diff --git a/src/launcher.py b/src/launcher.py new file mode 100644 index 0000000..a48dd4a --- /dev/null +++ b/src/launcher.py @@ -0,0 +1,229 @@ +"""CMBot update launcher (docs/10-lan-update.md stage 3/4). + +Compiled to Launcher.exe (PyInstaller onefile) and run instead of launching +app/CMBot.exe directly. On each start it: + + 1. seeds default config/templates into ~/.cmbot on first run; + 2. reads update_source / credentials from ~/.cmbot/config/app_config.json; + 3. checks the manifest and, when a newer version is advertised, downloads the + release zip, verifies SHA-256, extracts it, and swaps it into app\\ while + keeping the previous version in app.old\\ for rollback; + 4. launches app\\CMBot.exe. + +Degrades safely: an unreachable source, a bad download, or a non-writable +install root leaves the existing app\\ in place and launches it. The running +program's data lives in ~/.cmbot (get_data_dir), so updates never touch it. +""" +import argparse +import hashlib +import logging +import shutil +import subprocess +import sys +import zipfile +from pathlib import Path +from urllib import error + +# Make 'services' importable both frozen (PyInstaller --paths src) and from source. +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from services.update_service import check_for_update, download # noqa: E402 +from services.file_service import get_data_dir # noqa: E402 +from services.config_service import load_config # noqa: E402 + +APP_EXE = "CMBot.exe" +VERSION_FILE = "version.txt" +CONFIG_FILES = ("app_config.json", "templates.json") + +logger = logging.getLogger("launcher") + + +# ── helpers ────────────────────────────────────────────────────────────────── + +def _read_version(app_dir): + """Return the version recorded in app/version.txt, or '' if absent.""" + try: + # utf-8-sig tolerates a BOM written by PowerShell. + return (app_dir / VERSION_FILE).read_text(encoding="utf-8-sig").strip() + except OSError: + return "" + + +def seed_defaults(app_dir, data_dir): + """First run: copy factory config/templates from app\\config into the data + root, without overwriting any existing user file.""" + src = app_dir / "config" + if not src.is_dir(): + return + dst = data_dir / "config" + dst.mkdir(parents=True, exist_ok=True) + for name in CONFIG_FILES: + s, d = src / name, dst / name + if s.exists() and not d.exists(): + shutil.copy2(str(s), str(d)) + logger.info("Seeded default %s into %s", name, dst) + + +def is_writable(path): + """True if a file can be created under *path* (so self-update can proceed).""" + try: + path.mkdir(parents=True, exist_ok=True) + probe = path / ".write_test" + probe.write_text("x", encoding="ascii") + probe.unlink() + return True + except OSError: + return False + + +def _sha256(path): + h = hashlib.sha256() + with open(str(path), "rb") as f: + for chunk in iter(lambda: f.read(65536), b""): + h.update(chunk) + return h.hexdigest() + + +def _rm(path): + if path.is_dir(): + shutil.rmtree(str(path), ignore_errors=True) + elif path.exists(): + path.unlink() + + +def _find_package_root(extract_dir): + """Locate the folder that holds CMBot.exe inside an extracted zip. + + Handles both a flat zip (exe at root) and one wrapped in a single folder. + """ + if (extract_dir / APP_EXE).exists(): + return extract_dir + subs = [p for p in extract_dir.iterdir() if p.is_dir()] + if len(subs) == 1 and (subs[0] / APP_EXE).exists(): + return subs[0] + return None + + +def _swap_in(app_dir, old_dir, package_root): + """Rename app->app.old, then package_root->app. Roll back on failure. + + Never overwrites a running app dir: if app\\CMBot.exe is locked (running), + the rename raises and the caller degrades to the existing version. + """ + if old_dir.exists(): + shutil.rmtree(str(old_dir)) + if app_dir.exists(): + app_dir.rename(old_dir) + try: + package_root.rename(app_dir) + except OSError: + if not app_dir.exists() and old_dir.exists(): + old_dir.rename(app_dir) # restore previous version + raise + + +# ── update flow ────────────────────────────────────────────────────────────── + +def _try_update(install_root, app_dir, source, user, password, local_version): + info = check_for_update(source, local_version, user, password) + if info is None: + return # no update / unreachable / not newer — already logged + + if not is_writable(install_root): + logger.warning("Install root not writable, skipping update: %s", install_root) + return + + staging = install_root / "staging" + staging.mkdir(parents=True, exist_ok=True) + zip_path = staging / "{}.zip".format(info.version) + extract_dir = staging / "app.new" + _rm(zip_path) + _rm(extract_dir) + + logger.info("Downloading v%s from %s", info.version, info.source) + download(info.source, zip_path, user, password) + + if info.size and zip_path.stat().st_size != info.size: + raise ValueError("download size mismatch: got {}, expected {}".format( + zip_path.stat().st_size, info.size)) + if info.sha256 and _sha256(zip_path).lower() != info.sha256.lower(): + raise ValueError("sha256 mismatch") + + with zipfile.ZipFile(str(zip_path)) as z: + z.extractall(str(extract_dir)) + package_root = _find_package_root(extract_dir) + if not package_root: + raise ValueError("downloaded package does not contain {}".format(APP_EXE)) + pkg_version = _read_version(package_root) + if pkg_version != info.version: + raise ValueError("package version '{}' != manifest '{}'".format( + pkg_version, info.version)) + + _swap_in(app_dir, install_root / "app.old", package_root) + _rm(zip_path) + _rm(extract_dir) + logger.info("Installed and switched to v%s", info.version) + + +def run(install_root, no_launch=False): + """Seed config, attempt an update, then launch the app. Returns exit code.""" + install_root = Path(install_root) + app_dir = install_root / "app" + data_dir = get_data_dir() + + seed_defaults(app_dir, data_dir) + + cfg = load_config() + source = cfg.get("update_source", "") + user = cfg.get("update_user", "") + password = cfg.get("update_pass", "") + + local_version = _read_version(app_dir) + if source: + try: + _try_update(install_root, app_dir, source, user, password, local_version) + except (OSError, ValueError, error.URLError) as exc: + logger.warning("Update skipped (using local version): %s", exc) + else: + logger.info("No update_source configured, skipping update check.") + + exe = app_dir / APP_EXE + if not exe.exists(): + logger.error("Executable not found: %s", exe) + return 1 + if no_launch: + logger.info("NoLaunch: would start %s (data=%s)", exe, data_dir) + return 0 + logger.info("Launching %s", exe) + subprocess.Popen([str(exe)], cwd=str(exe.parent)) + return 0 + + +def _default_install_root(): + if getattr(sys, "frozen", False): + return Path(sys.executable).resolve().parent + return Path.cwd() + + +def main(argv=None): + parser = argparse.ArgumentParser(description="CMBot update launcher") + parser.add_argument("--install-root", default=None, + help="install root (default: Launcher.exe's folder)") + parser.add_argument("--no-launch", action="store_true", + help="update only, do not start the app") + args = parser.parse_args(argv) + + root = Path(args.install_root) if args.install_root else _default_install_root() + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(message)s", + handlers=[ + logging.StreamHandler(), + logging.FileHandler(str(root / "launcher.log"), encoding="utf-8"), + ], + ) + return run(root, no_launch=args.no_launch) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/services/update_service.py b/src/services/update_service.py index 9138a54..a553abb 100644 --- a/src/services/update_service.py +++ b/src/services/update_service.py @@ -26,9 +26,12 @@ MANIFEST_NAME = "manifest.json" class UpdateInfo: """A newer version advertised by the update source.""" version: str - source: str # folder or URL the user opens to get the new version + source: str # absolute zip URL (http) or folder/path to the new version notes: str = "" mandatory: bool = False + sha256: str = "" # expected hash of the release zip (launcher verifies) + size: int = 0 # release zip size in bytes (0 = unknown) + min_supported: str = "" def parse_version(text) -> tuple: @@ -70,17 +73,35 @@ def _manifest_url(update_source): return parse.urljoin(source, MANIFEST_NAME) +def make_auth_header(update_user="", update_pass=""): + """Return an HTTP Basic Auth header dict, or {} when no credentials.""" + if not (update_user or update_pass): + return {} + raw = "{}:{}".format(update_user or "", update_pass or "").encode("utf-8") + return {"Authorization": "Basic {}".format(base64.b64encode(raw).decode("ascii"))} + + +def download(url, dest, update_user="", update_pass="", timeout=120): + """Download *url* to *dest* (Path/str) with optional Basic Auth. + + Streams in chunks so large release zips don't load fully into memory. + Raises (OSError / URLError) on failure — callers handle degradation. + """ + headers = {"User-Agent": "CMBot"} + headers.update(make_auth_header(update_user, update_pass)) + req = request.Request(url, headers=headers) + with request.urlopen(req, timeout=timeout) as response, open(str(dest), "wb") as f: + while True: + chunk = response.read(65536) + if not chunk: + break + f.write(chunk) + + def _load_http_manifest(update_source, update_user="", update_pass=""): url = _manifest_url(update_source) - headers = { - "Accept": "application/json", - "User-Agent": "CMBot", - } - if update_user or update_pass: - raw = "{}:{}".format(update_user or "", update_pass or "").encode("utf-8") - headers["Authorization"] = "Basic {}".format( - base64.b64encode(raw).decode("ascii") - ) + headers = {"Accept": "application/json", "User-Agent": "CMBot"} + headers.update(make_auth_header(update_user, update_pass)) req = request.Request(url, headers=headers) with request.urlopen(req, timeout=5) as response: @@ -136,11 +157,18 @@ def check_for_update( source = str(data.get("url", "") or data.get("source", "")).strip() if not source: source = str(update_source) + elif _is_http_source(update_source): + # Resolve a relative release URL against the manifest URL so callers + # (in-app banner / launcher) always get a directly usable absolute URL. + source = parse.urljoin(_manifest_url(update_source), source) info = UpdateInfo( version=version, source=source, notes=str(data.get("notes", "")), mandatory=bool(data.get("mandatory", False)), + sha256=str(data.get("sha256", "")).strip(), + size=int(data.get("size", 0) or 0), + min_supported=str(data.get("min_supported", "")).strip(), ) logger.info("Update available: v%s (current v%s)", version, current_version) return info diff --git a/tasks.md b/tasks.md index e8719f2..c31ecd9 100644 --- a/tasks.md +++ b/tasks.md @@ -933,7 +933,7 @@ - [x] 文档：`docs/10` 改为便携 + `Launcher.exe` + `~/.cmbot` 模型（§3/§4/§5/§8/§9/§11/§16） - [x] `get_data_dir()` 三级回退：`CMBOT_DATA_DIR` → 打包态 `~/.cmbot` → 开发态项目根；`tests/test_file_service.py` 5 个单测 -- [ ] `src/launcher.py`：复用 `update_service`，下载 zip→SHA-256→解压→`app/app.old` 切换→启动；安装根可写性检测；首次把 `app\config\` 默认模板播种到 `~/.cmbot` +- [x] `src/launcher.py`：复用 `update_service`，下载 zip→SHA-256→解压→`app/app.old` 切换→启动；安装根可写性检测；首次把 `app\config\` 默认模板播种到 `~/.cmbot`；`update_service` 扩展（`UpdateInfo.sha256/size/min_supported`、绝对 url 解析、`make_auth_header`/`download`）；`tests/test_launcher.py` 11 个单测 + 本地 HTTP server 真实端到端验证 - [ ] `build.ps1` 增产 `Launcher.exe`（PyInstaller onefile），发布 zip 含 `Launcher.exe` + `app\` - [ ] 退休 `scripts/update.ps1` 与 `scripts/install_local.ps1` - [ ] 端到端实测（解压到 D 盘运行、自更新、回滚） diff --git a/tests/test_launcher.py b/tests/test_launcher.py new file mode 100644 index 0000000..fbde8f8 --- /dev/null +++ b/tests/test_launcher.py @@ -0,0 +1,153 @@ +"""Tests for launcher — no GUI/network dependency (download is mocked).""" +import hashlib +import json +import os +import shutil +import sys +import tempfile +import unittest +import zipfile +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +import launcher +from services.update_service import UpdateInfo + + +def _make_release_zip(dst_zip, version, exe_body): + """Build a flat release zip (CMBot.exe + version.txt) and return its sha256.""" + tmp = Path(tempfile.mkdtemp()) + try: + (tmp / "CMBot.exe").write_text(exe_body, encoding="ascii") + (tmp / "version.txt").write_text(version, encoding="ascii") + with zipfile.ZipFile(str(dst_zip), "w", zipfile.ZIP_DEFLATED) as z: + z.write(str(tmp / "CMBot.exe"), "CMBot.exe") + z.write(str(tmp / "version.txt"), "version.txt") + finally: + shutil.rmtree(str(tmp), ignore_errors=True) + return hashlib.sha256(dst_zip.read_bytes()).hexdigest() + + +class _LauncherBase(unittest.TestCase): + def setUp(self): + self.tmp = Path(tempfile.mkdtemp()) + self.install = self.tmp / "install" + self.app = self.install / "app" + (self.app / "config").mkdir(parents=True) + (self.app / "CMBot.exe").write_text("OLD-1.0.0", encoding="ascii") + (self.app / "version.txt").write_text("1.0.0", encoding="ascii") + # factory defaults shipped inside app\config + (self.app / "config" / "app_config.json").write_text( + json.dumps({"update_source": "http://x/manifest.json", "last_template": "正方形"}), + encoding="utf-8") + (self.app / "config" / "templates.json").write_text("{}", encoding="utf-8") + # data root -> a temp dir via the env override + self.data = self.tmp / "data" + self._prev_env = os.environ.get("CMBOT_DATA_DIR") + os.environ["CMBOT_DATA_DIR"] = str(self.data) + + def tearDown(self): + if self._prev_env is None: + os.environ.pop("CMBOT_DATA_DIR", None) + else: + os.environ["CMBOT_DATA_DIR"] = self._prev_env + shutil.rmtree(str(self.tmp), ignore_errors=True) + + +class TestHelpers(_LauncherBase): + def test_seed_defaults_copies_when_missing(self): + launcher.seed_defaults(self.app, self.data) + self.assertTrue((self.data / "config" / "app_config.json").exists()) + self.assertTrue((self.data / "config" / "templates.json").exists()) + + def test_seed_defaults_does_not_overwrite(self): + (self.data / "config").mkdir(parents=True) + (self.data / "config" / "app_config.json").write_text( + '{"update_source":"USER"}', encoding="utf-8") + launcher.seed_defaults(self.app, self.data) + kept = json.loads((self.data / "config" / "app_config.json").read_text(encoding="utf-8")) + self.assertEqual(kept["update_source"], "USER") + + def test_is_writable(self): + self.assertTrue(launcher.is_writable(self.install)) + + def test_find_package_root_flat(self): + d = self.tmp / "flat" + d.mkdir() + (d / "CMBot.exe").write_text("x", encoding="ascii") + self.assertEqual(launcher._find_package_root(d), d) + + def test_find_package_root_wrapped(self): + d = self.tmp / "wrap" + (d / "CMBot-1.1.0").mkdir(parents=True) + (d / "CMBot-1.1.0" / "CMBot.exe").write_text("x", encoding="ascii") + self.assertEqual(launcher._find_package_root(d), d / "CMBot-1.1.0") + + def test_find_package_root_missing(self): + d = self.tmp / "empty" + d.mkdir() + self.assertIsNone(launcher._find_package_root(d)) + + def test_read_version_tolerates_bom(self): + (self.app / "version.txt").write_text("1.0.0", encoding="utf-8-sig") + self.assertEqual(launcher._read_version(self.app), "1.0.0") + + +class TestRun(_LauncherBase): + def _info(self, version, zip_path): + sha = hashlib.sha256(zip_path.read_bytes()).hexdigest() + return UpdateInfo(version=version, source="http://x/{}.zip".format(version), + sha256=sha, size=zip_path.stat().st_size) + + def test_update_applied_and_swapped(self): + new_zip = self.tmp / "new.zip" + _make_release_zip(new_zip, "1.1.0", "NEW-1.1.0") + info = self._info("1.1.0", new_zip) + + def fake_download(url, dest, user="", password="", timeout=120): + shutil.copy(str(new_zip), str(dest)) + + with patch("launcher.check_for_update", return_value=info), \ + patch("launcher.download", side_effect=fake_download): + rc = launcher.run(self.install, no_launch=True) + + self.assertEqual(rc, 0) + self.assertEqual((self.app / "version.txt").read_text(encoding="ascii"), "1.1.0") + self.assertEqual((self.app / "CMBot.exe").read_text(encoding="ascii"), "NEW-1.1.0") + self.assertEqual(((self.install / "app.old") / "version.txt").read_text(encoding="ascii"), "1.0.0") + self.assertFalse((self.install / "staging" / "1.1.0.zip").exists()) + + def test_no_update_keeps_local(self): + with patch("launcher.check_for_update", return_value=None): + rc = launcher.run(self.install, no_launch=True) + self.assertEqual(rc, 0) + self.assertEqual((self.app / "version.txt").read_text(encoding="ascii"), "1.0.0") + self.assertFalse((self.install / "app.old").exists()) + + def test_sha256_mismatch_degrades(self): + new_zip = self.tmp / "new.zip" + _make_release_zip(new_zip, "1.1.0", "NEW") + info = self._info("1.1.0", new_zip) + info.sha256 = "deadbeef" # force mismatch + + def fake_download(url, dest, user="", password="", timeout=120): + shutil.copy(str(new_zip), str(dest)) + + with patch("launcher.check_for_update", return_value=info), \ + patch("launcher.download", side_effect=fake_download): + rc = launcher.run(self.install, no_launch=True) + + self.assertEqual(rc, 0) + self.assertEqual((self.app / "version.txt").read_text(encoding="ascii"), "1.0.0") + self.assertFalse((self.install / "app.old").exists()) + + def test_seeds_config_into_data_root(self): + with patch("launcher.check_for_update", return_value=None): + launcher.run(self.install, no_launch=True) + self.assertTrue((self.data / "config" / "app_config.json").exists()) + + +if __name__ == "__main__": + unittest.main()