304 lines
9.9 KiB
Python
304 lines
9.9 KiB
Python
"""发布包的轻量启动器和下次启动更新器。
|
|
|
|
Launcher 不联网。主程序已经下载并校验好 ``data/update/app.new`` 后，本模块在
|
|
主程序未运行时替换 ``app``，失败时恢复 ``app.old``。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ctypes
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Callable, Optional
|
|
|
|
|
|
PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
|
|
HEALTH_WAIT_SECONDS = 15.0
|
|
|
|
|
|
class UpdateApplyError(RuntimeError):
|
|
"""Launcher 无法安全应用或恢复更新。"""
|
|
|
|
|
|
def install_root(executable: str | None = None) -> Path:
|
|
"""返回 Launcher.exe 所在的发布根目录。"""
|
|
|
|
return Path(executable or sys.executable).resolve().parent
|
|
|
|
|
|
def app_executable(root: Path) -> Path:
|
|
"""返回主程序路径。"""
|
|
|
|
return root / "app" / "CMAutoBuy.exe"
|
|
|
|
|
|
def update_directory(root: Path) -> Path:
|
|
"""返回在线更新状态目录。"""
|
|
|
|
return root / "data" / "update"
|
|
|
|
|
|
def show_error(message: str) -> None:
|
|
"""使用 Windows 原生对话框显示启动错误。"""
|
|
|
|
ctypes.windll.user32.MessageBoxW(0, message, "商品采集采购工具", 0x10)
|
|
|
|
|
|
def show_information(message: str) -> None:
|
|
"""使用 Windows 原生对话框显示普通提示。"""
|
|
|
|
ctypes.windll.user32.MessageBoxW(0, message, "商品采集采购工具", 0x40)
|
|
|
|
|
|
def _read_json(path: Path) -> Optional[dict]:
|
|
try:
|
|
value = json.loads(path.read_text(encoding="utf-8-sig"))
|
|
except (OSError, UnicodeDecodeError, json.JSONDecodeError):
|
|
return None
|
|
return value if isinstance(value, dict) else None
|
|
|
|
|
|
def _write_json_atomic(path: Path, value: dict) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = path.with_suffix(path.suffix + ".tmp")
|
|
temporary.write_text(
|
|
json.dumps(value, ensure_ascii=False, indent=2) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
os.replace(str(temporary), str(path))
|
|
|
|
|
|
def _safe_remove(root: Path, path: Path) -> None:
|
|
"""只删除发布根目录内明确的更新目标。"""
|
|
|
|
root_resolved = root.resolve()
|
|
path_resolved = path.resolve()
|
|
if path_resolved == root_resolved or not path_resolved.is_relative_to(root_resolved):
|
|
raise UpdateApplyError(f"拒绝删除发布目录之外的路径：{path_resolved}")
|
|
if not path.exists():
|
|
return
|
|
if path.is_dir():
|
|
shutil.rmtree(path)
|
|
else:
|
|
path.unlink()
|
|
|
|
|
|
def _health_matches(directory: Path, version: str) -> bool:
|
|
health = _read_json(directory / "healthy.json")
|
|
return health is not None and health.get("version") == version
|
|
|
|
|
|
def is_process_running(pid: int, expected_executable: Path) -> bool:
|
|
"""确认 PID 仍存在，并且确实是本发布目录的主程序。"""
|
|
|
|
if pid <= 0 or sys.platform != "win32":
|
|
return False
|
|
kernel32 = ctypes.windll.kernel32
|
|
kernel32.OpenProcess.argtypes = [
|
|
ctypes.c_ulong,
|
|
ctypes.c_int,
|
|
ctypes.c_ulong,
|
|
]
|
|
kernel32.OpenProcess.restype = ctypes.c_void_p
|
|
kernel32.QueryFullProcessImageNameW.argtypes = [
|
|
ctypes.c_void_p,
|
|
ctypes.c_ulong,
|
|
ctypes.c_wchar_p,
|
|
ctypes.POINTER(ctypes.c_ulong),
|
|
]
|
|
kernel32.QueryFullProcessImageNameW.restype = ctypes.c_int
|
|
kernel32.CloseHandle.argtypes = [ctypes.c_void_p]
|
|
kernel32.CloseHandle.restype = ctypes.c_int
|
|
handle = kernel32.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
|
|
if not handle:
|
|
return False
|
|
try:
|
|
buffer = ctypes.create_unicode_buffer(32768)
|
|
size = ctypes.c_ulong(len(buffer))
|
|
if not kernel32.QueryFullProcessImageNameW(handle, 0, buffer, ctypes.byref(size)):
|
|
return False
|
|
actual = Path(buffer.value).resolve()
|
|
return str(actual).casefold() == str(expected_executable.resolve()).casefold()
|
|
finally:
|
|
kernel32.CloseHandle(handle)
|
|
|
|
|
|
def is_main_program_running(
|
|
root: Path,
|
|
checker: Callable[[int, Path], bool] = is_process_running,
|
|
) -> bool:
|
|
"""读取上次启动 PID，判断主程序是否仍在运行。"""
|
|
|
|
directory = update_directory(root)
|
|
running_file = directory / "running.json"
|
|
running = _read_json(running_file)
|
|
if running is None:
|
|
return False
|
|
pid = running.get("pid")
|
|
if isinstance(pid, int) and checker(pid, app_executable(root)):
|
|
return True
|
|
try:
|
|
running_file.unlink(missing_ok=True)
|
|
except OSError:
|
|
pass
|
|
return False
|
|
|
|
|
|
def _validate_pending(pending: dict) -> tuple[str, str]:
|
|
version = pending.get("version")
|
|
state = pending.get("state")
|
|
if (
|
|
pending.get("schema_version") != 1
|
|
or not isinstance(version, str)
|
|
or state not in {"ready", "applied"}
|
|
):
|
|
raise UpdateApplyError("待更新状态文件无效")
|
|
return version, state
|
|
|
|
|
|
def finalize_applied_update(root: Path) -> None:
|
|
"""健康启动后清除待更新状态，保留一份 app.old。"""
|
|
|
|
directory = update_directory(root)
|
|
(directory / "pending.json").unlink(missing_ok=True)
|
|
_safe_remove(root, directory / "app.new")
|
|
|
|
|
|
def rollback_applied_update(root: Path, version: str, reason: str) -> None:
|
|
"""把未健康启动的新 app 移走，并恢复 app.old。"""
|
|
|
|
directory = update_directory(root)
|
|
current_app = root / "app"
|
|
old_app = root / "app.old"
|
|
failed_app = directory / "app.failed"
|
|
if not old_app.is_dir():
|
|
raise UpdateApplyError("新版本启动失败，但没有可恢复的 app.old")
|
|
|
|
_safe_remove(root, failed_app)
|
|
if current_app.exists():
|
|
os.replace(str(current_app), str(failed_app))
|
|
try:
|
|
os.replace(str(old_app), str(current_app))
|
|
except OSError as exc:
|
|
if failed_app.exists() and not current_app.exists():
|
|
os.replace(str(failed_app), str(current_app))
|
|
raise UpdateApplyError(f"恢复旧版本失败：{exc}") from exc
|
|
|
|
(directory / "pending.json").unlink(missing_ok=True)
|
|
_write_json_atomic(
|
|
directory / "last_error.json",
|
|
{"schema_version": 1, "version": version, "reason": reason},
|
|
)
|
|
|
|
|
|
def apply_pending_update(root: Path) -> Optional[str]:
|
|
"""应用已暂存更新，返回刚应用的版本；没有更新时返回 None。"""
|
|
|
|
directory = update_directory(root)
|
|
pending_path = directory / "pending.json"
|
|
if not pending_path.exists():
|
|
return None
|
|
pending = _read_json(pending_path)
|
|
if pending is None:
|
|
raise UpdateApplyError("待更新状态文件损坏，请删除 data/update/pending.json 后重试")
|
|
version, state = _validate_pending(pending)
|
|
|
|
if state == "applied":
|
|
if _health_matches(directory, version):
|
|
finalize_applied_update(root)
|
|
return None
|
|
rollback_applied_update(root, version, "新版本上次启动未通过健康检查")
|
|
return None
|
|
|
|
current_app = root / "app"
|
|
staged_app = directory / "app.new"
|
|
old_app = root / "app.old"
|
|
if not current_app.is_dir() or not (current_app / "CMAutoBuy.exe").is_file():
|
|
raise UpdateApplyError("当前 app 目录不完整，不能应用更新")
|
|
if not staged_app.is_dir() or not (staged_app / "CMAutoBuy.exe").is_file():
|
|
raise UpdateApplyError("已下载更新不完整，请在设置页重新下载")
|
|
|
|
_safe_remove(root, old_app)
|
|
(directory / "healthy.json").unlink(missing_ok=True)
|
|
os.replace(str(current_app), str(old_app))
|
|
try:
|
|
os.replace(str(staged_app), str(current_app))
|
|
pending["state"] = "applied"
|
|
_write_json_atomic(pending_path, pending)
|
|
except Exception as exc:
|
|
if current_app.exists() and not staged_app.exists():
|
|
os.replace(str(current_app), str(staged_app))
|
|
if old_app.exists() and not current_app.exists():
|
|
os.replace(str(old_app), str(current_app))
|
|
raise UpdateApplyError(f"应用更新失败，已恢复旧版本：{exc}") from exc
|
|
return version
|
|
|
|
|
|
def wait_for_health(process, root: Path, version: str) -> Optional[bool]:
|
|
"""等待健康标记；成功=True，提前退出=False，超时仍运行=None。"""
|
|
|
|
directory = update_directory(root)
|
|
deadline = time.monotonic() + HEALTH_WAIT_SECONDS
|
|
while time.monotonic() < deadline:
|
|
if _health_matches(directory, version):
|
|
return True
|
|
if process.poll() is not None:
|
|
return False
|
|
time.sleep(0.25)
|
|
return None
|
|
|
|
|
|
def _start_main_program(root: Path):
|
|
target = app_executable(root)
|
|
if not target.is_file():
|
|
raise OSError(f"主程序不存在：{target}")
|
|
process = subprocess.Popen([str(target)], cwd=str(target.parent))
|
|
_write_json_atomic(
|
|
update_directory(root) / "running.json",
|
|
{"schema_version": 1, "pid": process.pid},
|
|
)
|
|
return process
|
|
|
|
|
|
def main() -> int:
|
|
"""安全应用待更新目录，然后启动主程序。"""
|
|
|
|
root = install_root()
|
|
if is_main_program_running(root):
|
|
show_information("商品采集采购工具已经在运行，请勿重复启动。")
|
|
return 0
|
|
|
|
try:
|
|
applied_version = apply_pending_update(root)
|
|
process = _start_main_program(root)
|
|
except (OSError, UpdateApplyError) as exc:
|
|
show_error(f"主程序启动或更新失败。\n\n{exc}")
|
|
return 1
|
|
|
|
if applied_version:
|
|
health = wait_for_health(process, root, applied_version)
|
|
if health is True:
|
|
finalize_applied_update(root)
|
|
elif health is False:
|
|
try:
|
|
rollback_applied_update(
|
|
root,
|
|
applied_version,
|
|
"新版本启动后提前退出",
|
|
)
|
|
_start_main_program(root)
|
|
except (OSError, UpdateApplyError) as exc:
|
|
show_error(f"新版本启动失败，恢复旧版本时发生错误。\n\n{exc}")
|
|
return 1
|
|
show_error("新版本未能正常启动，已自动恢复并启动旧版本。")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|