110 lines
3.5 KiB
Python
110 lines
3.5 KiB
Python
"""自动升级后的启动健康标记与失败版本熔断记录。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import time
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
|
|
HEALTH_STATUSES = {"process_started", "main_window_ready", "environment_blocked"}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class UpdateHealthContext:
|
|
install_root: Path
|
|
transaction_id: str
|
|
target_version: str
|
|
|
|
@property
|
|
def transaction_dir(self):
|
|
return self.install_root / ".cmshopee-update" / "transactions" / self.transaction_id
|
|
|
|
@property
|
|
def health_path(self):
|
|
return self.transaction_dir / "health.json"
|
|
|
|
|
|
def _atomic_json(path, payload):
|
|
path = Path(path)
|
|
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) + "\n", encoding="utf-8")
|
|
os.replace(str(temporary), str(path))
|
|
|
|
|
|
def context_from_argv(argv, install_root, app_version):
|
|
values = list(argv or [])
|
|
try:
|
|
transaction_index = values.index("--update-transaction")
|
|
version_index = values.index("--update-target-version")
|
|
transaction_id = values[transaction_index + 1]
|
|
target_version = values[version_index + 1]
|
|
except (ValueError, IndexError):
|
|
return None
|
|
if not re.fullmatch(r"[A-Za-z0-9-]{8,64}", transaction_id):
|
|
return None
|
|
if target_version != app_version:
|
|
return None
|
|
return UpdateHealthContext(Path(install_root).resolve(), transaction_id, target_version)
|
|
|
|
|
|
def write_health(context, status, message=""):
|
|
if context is None:
|
|
return None
|
|
if status not in HEALTH_STATUSES:
|
|
raise ValueError("未知升级健康状态")
|
|
payload = {
|
|
"schema_version": 1,
|
|
"transaction_id": context.transaction_id,
|
|
"target_version": context.target_version,
|
|
"status": status,
|
|
"message": str(message or "")[:300],
|
|
"updated_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
|
|
}
|
|
_atomic_json(context.health_path, payload)
|
|
return context.health_path
|
|
|
|
|
|
def failed_versions_path(install_root):
|
|
return Path(install_root).resolve() / ".cmshopee-update" / "failed-versions.json"
|
|
|
|
|
|
def load_failed_versions(install_root):
|
|
path = failed_versions_path(install_root)
|
|
try:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
return {"schema_version": 1, "releases": {}}
|
|
releases = payload.get("releases")
|
|
if not isinstance(releases, dict):
|
|
releases = {}
|
|
return {"schema_version": 1, "releases": releases}
|
|
|
|
|
|
def failed_release_key(version, sha256):
|
|
return "%s:%s" % (str(version or "").strip(), str(sha256 or "").strip().lower())
|
|
|
|
|
|
def record_failed_release(install_root, version, sha256, reason):
|
|
payload = load_failed_versions(install_root)
|
|
key = failed_release_key(version, sha256)
|
|
previous = payload["releases"].get(key) or {}
|
|
payload["releases"][key] = {
|
|
"version": str(version),
|
|
"sha256": str(sha256).lower(),
|
|
"reason": str(reason or "自动升级后新版未能正常启动")[:300],
|
|
"attempts": int(previous.get("attempts") or 0) + 1,
|
|
"failed_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
|
|
}
|
|
_atomic_json(failed_versions_path(install_root), payload)
|
|
return payload["releases"][key]
|
|
|
|
|
|
def get_failed_release(install_root, version, sha256):
|
|
payload = load_failed_versions(install_root)
|
|
return payload["releases"].get(failed_release_key(version, sha256))
|