chengma/cmbot
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: src/launcher.py — Python update launcher (HTTP, app/app.old swap)

Browse Source 
Launcher core that reuses services/update_service: seed default config into
~/.cmbot on first run, check the manifest, download the release zip, verify
SHA-256, extract, validate version, swap into app/ (keeping app.old/ for
rollback), then launch app/CMBot.exe. Degrades to the local version on any
failure; checks install-root writability before updating.

- update_service: UpdateInfo gains sha256/size/min_supported; release URL
  resolved to absolute; extract make_auth_header + add download() helper
- tests/test_launcher.py: 11 tests (seed, find-root, swap, sha mismatch,
  no-update); also verified end-to-end against a local HTTP server

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
admin
2026-06-18 10:49:45 +08:00
co-authored by Claude Opus 4.8
parent 3d01bed98c
commit 1d8e4fc240
4 changed files with 421 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/launcher.py
+229
View File
@@ -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())
src/services/update_service.py
+38 -10
View File
@@ -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