"""Startup update-check helpers. The first release only decides whether the app may enter the main window. It never downloads, overwrites, deletes, or migrates local program/data files. """ from __future__ import annotations import json import re import urllib.request from dataclasses import dataclass from . import client_policy from .version import APP_CODE_NAME, APP_UPDATE_CHECK_URL, APP_VERSION DEFAULT_TIMEOUT_SECONDS = 3.0 class UpdateCheckError(RuntimeError): """Raised for invalid update-check inputs or server responses.""" @dataclass(frozen=True) class UpdateInfo: latest_version: str = "" min_supported_version: str = "" force_update: bool = False download_url: str = "" sha256: str = "" size_bytes: int = 0 package_format: str = "" updater_protocol: int = 0 min_updater_protocol: int = 0 signature_algorithm: str = "" manifest_signature: str = "" message: str = "" @dataclass(frozen=True) class UpdateCheckResult: current_version: str checked: bool = False forced: bool = False latest_version: str = "" min_supported_version: str = "" download_url: str = "" sha256: str = "" size_bytes: int = 0 package_format: str = "" updater_protocol: int = 0 min_updater_protocol: int = 0 signature_algorithm: str = "" manifest_signature: str = "" automatic_update_error: str = "" message: str = "" error: str = "" client_policy: client_policy.ClientPolicy | None = None client_policy_error: str = "" @property def can_enter(self) -> bool: return not self.forced def parse_version(version) -> tuple[int, ...]: """Parse semantic numeric version segments for comparison.""" text = str(version or "").strip() if text.lower().startswith("v"): text = text[1:].strip() if not text: raise UpdateCheckError("版本号不能为空") segments = [] for part in text.split("."): match = re.match(r"^(\d+)", part.strip()) if match is None: raise UpdateCheckError(f"版本号格式不正确：{version}") segments.append(int(match.group(1))) return tuple(segments) def compare_versions(left, right) -> int: """Return -1/0/1 for numeric semantic version comparison.""" left_segments = parse_version(left) right_segments = parse_version(right) length = max(len(left_segments), len(right_segments)) left_padded = left_segments + (0,) * (length - len(left_segments)) right_padded = right_segments + (0,) * (length - len(right_segments)) if left_padded < right_padded: return -1 if left_padded > right_padded: return 1 return 0 def _as_bool(value) -> bool: if isinstance(value, bool): return value if isinstance(value, (int, float)): return value != 0 if isinstance(value, str): return value.strip().lower() in {"1", "true", "yes", "y", "on"} return False def parse_update_info(payload) -> UpdateInfo: if not isinstance(payload, dict): raise UpdateCheckError("版本接口返回内容不是 JSON 对象") release = payload.get("release") if not isinstance(release, dict): release = {} latest_version = str(payload.get("latest_version") or release.get("version") or "").strip() min_supported_version = str( payload.get("min_supported_version") or release.get("min_supported_version") or "" ).strip() if not latest_version and not min_supported_version: raise UpdateCheckError("版本接口缺少 latest_version 或 min_supported_version") return UpdateInfo( latest_version=latest_version, min_supported_version=min_supported_version, force_update=_as_bool(payload.get("force_update", release.get("force_update"))), download_url=str(payload.get("download_url") or release.get("download_url") or "").strip(), sha256=str(payload.get("sha256") or release.get("sha256") or "").strip(), size_bytes=int(payload.get("size_bytes") or release.get("size_bytes") or 0), package_format=str( payload.get("package_format") or release.get("package_format") or "" ).strip(), updater_protocol=int( payload.get("updater_protocol") or release.get("updater_protocol") or 0 ), min_updater_protocol=int( payload.get("min_updater_protocol") or release.get("min_updater_protocol") or 0 ), signature_algorithm=str( payload.get("signature_algorithm") or release.get("signature_algorithm") or "" ).strip(), manifest_signature=str( payload.get("manifest_signature") or release.get("manifest_signature") or "" ).strip(), message=str( payload.get("message") or payload.get("release_notes") or release.get("message") or release.get("release_notes") or "" ).strip(), ) def is_forced_update(info: UpdateInfo, current_version: str) -> bool: if info.min_supported_version and compare_versions(current_version, info.min_supported_version) < 0: return True if ( info.force_update and info.latest_version and compare_versions(current_version, info.latest_version) < 0 ): return True return False def _decode_payload(raw_payload): if isinstance(raw_payload, (bytes, bytearray)): raw_payload = raw_payload.decode("utf-8") if isinstance(raw_payload, str): try: return json.loads(raw_payload) except ValueError as exc: raise UpdateCheckError("版本接口返回内容不是合法 JSON") from exc return raw_payload def fetch_update_payload(url: str, timeout: float = DEFAULT_TIMEOUT_SECONDS): request = urllib.request.Request( url, headers={ "Accept": "application/json", "User-Agent": f"{APP_CODE_NAME}/{APP_VERSION}", }, ) with urllib.request.urlopen(request, timeout=timeout) as response: return _decode_payload(response.read()) def check_for_update( *, current_version: str = APP_VERSION, url: str = APP_UPDATE_CHECK_URL, timeout: float = DEFAULT_TIMEOUT_SECONDS, fetcher=None, ) -> UpdateCheckResult: if not str(url or "").strip(): return UpdateCheckResult(current_version=current_version, checked=False) try: payload = _decode_payload((fetcher or fetch_update_payload)(url, timeout)) except Exception as exc: return UpdateCheckResult( current_version=current_version, checked=True, forced=False, error=f"启动版本检查失败，已允许继续使用：{exc}", ) resolved_policy = None policy_error = "" if isinstance(payload, dict) and "client_policy" in payload: try: resolved_policy = client_policy.parse_client_policy( payload.get("client_policy") ) policy_error = resolved_policy.warning except client_policy.ClientPolicyError as exc: policy_error = str(exc) try: if isinstance(payload, dict) and payload.get("release", object()) is None: info = UpdateInfo() else: info = parse_update_info(payload) forced = is_forced_update(info, current_version) return UpdateCheckResult( current_version=current_version, checked=True, forced=forced, latest_version=info.latest_version, min_supported_version=info.min_supported_version, download_url=info.download_url, sha256=info.sha256, size_bytes=info.size_bytes, package_format=info.package_format, updater_protocol=info.updater_protocol, min_updater_protocol=info.min_updater_protocol, signature_algorithm=info.signature_algorithm, manifest_signature=info.manifest_signature, message=info.message, client_policy=resolved_policy, client_policy_error=policy_error, ) except Exception as exc: return UpdateCheckResult( current_version=current_version, checked=True, forced=False, error=f"启动版本检查失败，已允许继续使用：{exc}", client_policy=resolved_policy, client_policy_error=policy_error, )