197 lines
8.6 KiB
Python
197 lines
8.6 KiB
Python
"""甄 AI 客服独立更新器；仅依赖 Python 标准库，打包为单文件 EXE。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import ctypes
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from installer_payload import (MANIFEST_NAME, REQUIRED_FILES, apply_payload,
|
|
fresh_process_environment, validate_installation)
|
|
|
|
|
|
def _sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as source:
|
|
for chunk in iter(lambda: source.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _process_running(pid: int) -> bool:
|
|
if pid <= 0:
|
|
return False
|
|
if os.name == "nt":
|
|
process_query_limited_information = 0x1000
|
|
handle = ctypes.windll.kernel32.OpenProcess(
|
|
process_query_limited_information, False, pid
|
|
)
|
|
if not handle:
|
|
return False
|
|
ctypes.windll.kernel32.CloseHandle(handle)
|
|
return True
|
|
try:
|
|
os.kill(pid, 0)
|
|
return True
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def _write_result(path: Path, payload: dict) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = path.with_suffix(path.suffix + ".tmp")
|
|
temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
os.replace(temporary, path)
|
|
|
|
|
|
def _wait_for_parent(pid: int, timeout: int = 120) -> None:
|
|
deadline = time.monotonic() + timeout
|
|
while _process_running(pid):
|
|
if time.monotonic() >= deadline:
|
|
raise RuntimeError("主程序在 120 秒内没有退出")
|
|
time.sleep(0.25)
|
|
|
|
|
|
def _start_available_version(install_dir: Path, preferred: Path | None = None) -> bool:
|
|
"""Never start a missing runtime or pair an old EXE with a different payload."""
|
|
try:
|
|
manifest_file = install_dir / MANIFEST_NAME
|
|
if manifest_file.is_file():
|
|
manifest = validate_installation(install_dir, hashes=False)
|
|
candidate = install_dir / manifest["executable"]
|
|
if preferred is not None and candidate != preferred:
|
|
return False
|
|
else:
|
|
# Legacy installations have no manifest; refuse an incomplete Qt runtime.
|
|
if any(not (install_dir / name).is_file() for name in REQUIRED_FILES):
|
|
return False
|
|
candidates = sorted(install_dir.glob("ZhenAI-WeCom-Assistant-v*.exe"),
|
|
key=lambda path: path.stat().st_mtime, reverse=True)
|
|
candidate = preferred if preferred is not None else next(iter(candidates), None)
|
|
if candidate is None or not candidate.is_file():
|
|
return False
|
|
process = subprocess.Popen([str(candidate)], cwd=install_dir, close_fds=True,
|
|
env=fresh_process_environment())
|
|
try:
|
|
return process.wait(timeout=3) == 0
|
|
except subprocess.TimeoutExpired:
|
|
return True
|
|
except (OSError, ValueError, KeyError, RuntimeError):
|
|
return False
|
|
|
|
|
|
def _apply_installer_payload(arguments: list[str]) -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--apply-payload", required=True)
|
|
parser.add_argument("--install-dir", required=True)
|
|
parser.add_argument("--target-version", required=True)
|
|
parser.add_argument("--close-script", required=True)
|
|
parser.add_argument("--result-file", required=True)
|
|
args = parser.parse_args(arguments)
|
|
result_file = Path(args.result_file).resolve()
|
|
install_dir = Path(args.install_dir).absolute()
|
|
result = {"target_version": args.target_version, "install_dir": str(install_dir),
|
|
"status": "failed", "error": "", "updated_at": time.strftime("%Y-%m-%d %H:%M:%S")}
|
|
def close_application():
|
|
powershell = Path(os.environ["SystemRoot"]) / "System32/WindowsPowerShell/v1.0/powershell.exe"
|
|
process = subprocess.run([str(powershell), "-NoLogo", "-NoProfile", "-NonInteractive",
|
|
"-ExecutionPolicy", "Bypass", "-File", args.close_script,
|
|
"-InstallDir", str(install_dir)], capture_output=True, text=True,
|
|
errors="replace", timeout=120, env=fresh_process_environment(),
|
|
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
|
|
if process.returncode:
|
|
raise RuntimeError((process.stdout + process.stderr).strip() or
|
|
f"Cannot stop installed application: {process.returncode}")
|
|
try:
|
|
backup = apply_payload(Path(args.apply_payload), install_dir,
|
|
args.target_version, close_application, move_payload=True)
|
|
result.update(status="installed", backup_dir=str(backup or ""))
|
|
_write_result(result_file, result)
|
|
return 0
|
|
except Exception as exc:
|
|
result["error"] = str(exc)[:2000]
|
|
_write_result(result_file, result)
|
|
return 2
|
|
|
|
|
|
def main() -> int:
|
|
# External Windows tools must not inherit PyInstaller's private DLL directory.
|
|
if os.name == "nt":
|
|
ctypes.windll.kernel32.SetDllDirectoryW(None)
|
|
if "--apply-payload" in sys.argv:
|
|
return _apply_installer_payload(sys.argv[1:])
|
|
parser = argparse.ArgumentParser(description="甄 AI 客服独立更新器")
|
|
parser.add_argument("--installer", required=True)
|
|
parser.add_argument("--parent-pid", type=int, required=True)
|
|
parser.add_argument("--install-dir", required=True)
|
|
parser.add_argument("--target-version", required=True)
|
|
parser.add_argument("--result-file", required=True)
|
|
parser.add_argument("--expected-sha256", default="")
|
|
parser.add_argument("--from-version", default="")
|
|
args = parser.parse_args()
|
|
|
|
installer = Path(args.installer).resolve()
|
|
install_dir = Path(args.install_dir).resolve()
|
|
result_file = Path(args.result_file).resolve()
|
|
result = {
|
|
"target_version": args.target_version,
|
|
"from_version": args.from_version,
|
|
"status": "failed",
|
|
"error": "",
|
|
"updated_at": time.strftime("%Y-%m-%d %H:%M:%S"),
|
|
}
|
|
attempted_start = False
|
|
try:
|
|
_wait_for_parent(args.parent_pid)
|
|
if not installer.is_file():
|
|
raise RuntimeError("升级安装包不存在")
|
|
if args.expected_sha256 and _sha256(installer) != args.expected_sha256.lower():
|
|
raise RuntimeError("安装前 SHA256 复核失败")
|
|
# NSIS requires /D last and unquoted, including when the path has spaces.
|
|
command = subprocess.list2cmdline([str(installer), "/S"]) + " /D=" + str(install_dir)
|
|
install_started = time.time()
|
|
completed = subprocess.run(
|
|
command, timeout=1800, env=fresh_process_environment(),
|
|
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
|
)
|
|
if completed.returncode != 0:
|
|
detail = "磁盘空间不足，旧版本已保留，请释放空间后重试" if completed.returncode == 112 else ""
|
|
install_result = result_file.parent / "install-result.json"
|
|
try:
|
|
if install_result.stat().st_mtime >= install_started - 2:
|
|
record = json.loads(install_result.read_text(encoding="utf-8"))
|
|
if (record.get("target_version") == args.target_version
|
|
and Path(record.get("install_dir", "")) == install_dir):
|
|
detail = str(record.get("error") or "")[:1000]
|
|
except (OSError, ValueError, TypeError):
|
|
pass
|
|
raise RuntimeError(detail or f"静默安装失败，退出码 {completed.returncode}")
|
|
target = install_dir / f"ZhenAI-WeCom-Assistant-v{args.target_version}.exe"
|
|
if not target.is_file():
|
|
raise RuntimeError("安装完成但没有找到新版本主程序")
|
|
validate_installation(install_dir, args.target_version)
|
|
attempted_start = True
|
|
if not _start_available_version(install_dir, target):
|
|
raise RuntimeError("安装完成，但新版本未能正常启动；请手动重新安装")
|
|
result["status"] = "installed"
|
|
_write_result(result_file, result)
|
|
return 0
|
|
except Exception as exc:
|
|
result["status"] = "failed"
|
|
result["error"] = str(exc)[:1000]
|
|
# A failed installer may leave the previous payload intact. Start it only
|
|
# when its runtime validates; never launch a partial/new-old mixture.
|
|
result["restarted"] = False if attempted_start else _start_available_version(install_dir)
|
|
_write_result(result_file, result)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|