feat: add startup forced update check

2026-07-07 17:58:28 +08:00
parent 610a2304cc
commit 95b180e9cf
9 changed files with 527 additions and 12 deletions
@@ -4,8 +4,9 @@ from __future__ import annotations

 import os
 import sys
+import webbrowser

-from .. import appconfig
+from .. import appconfig, diagnostics, update_check
 from ..version import APP_NAME, display_name
 from . import widgets as _widgets
 from .widgets import *
@@ -42,6 +43,80 @@ def _ensure_offscreen_for_headless_tests():
        os.environ["QT_QPA_PLATFORM"] = "offscreen"


+def _write_update_check_diagnostic(message, *, result=None, exc=None):
+    payload = None
+    if result is not None:
+        payload = {
+            "checked": result.checked,
+            "forced": result.forced,
+            "current_version": result.current_version,
+            "latest_version": result.latest_version,
+            "min_supported_version": result.min_supported_version,
+            "download_url": result.download_url,
+            "error": result.error,
+        }
+    try:
+        diagnostics.write_diagnostic_log(
+            message,
+            level="WARNING",
+            step="startup_update_check",
+            payload=payload,
+            exc=exc,
+        )
+    except Exception:
+        # 版本检查第一版必须失败放行，诊断日志不可写也不能阻断启动。
+        pass
+
+
+def _forced_update_details(result):
+    online_version = result.latest_version or result.min_supported_version or "未知"
+    lines = [
+        f"当前版本：{result.current_version}",
+        f"线上版本：{online_version}",
+    ]
+    if result.min_supported_version:
+        lines.append(f"最低支持版本：{result.min_supported_version}")
+    if result.message:
+        lines.append(f"升级说明：{result.message}")
+    if result.download_url:
+        lines.append("请下载新版，关闭程序后覆盖程序文件和 _internal/，保留 data/ 目录。")
+    else:
+        lines.append("版本接口未提供下载地址，请联系管理员获取新版后再使用。")
+    return "\n".join(lines)
+
+
+def _show_forced_update_dialog(result, *, parent=None, opener=None) -> bool:
+    opener = opener or webbrowser.open
+    box = QMessageBox(parent)
+    box.setIcon(QMessageBox.Warning)
+    box.setWindowTitle("必须升级")
+    box.setText("当前版本已不能继续使用，请先升级到新版。")
+    box.setInformativeText(_forced_update_details(result))
+    download_button = box.addButton("下载新版", QMessageBox.AcceptRole)
+    exit_button = box.addButton("退出程序", QMessageBox.RejectRole)
+    if not result.download_url and hasattr(download_button, "setEnabled"):
+        download_button.setEnabled(False)
+    box.setDefaultButton(download_button if result.download_url else exit_button)
+    box.exec()
+    if box.clickedButton() is download_button and result.download_url:
+        opener(result.download_url)
+    return False
+
+
+def _run_startup_update_gate(*, checker=None, opener=None) -> bool:
+    try:
+        result = (checker or update_check.check_for_update)()
+    except Exception as exc:
+        _write_update_check_diagnostic("启动版本检查异常，已允许继续使用", exc=exc)
+        return True
+
+    if result.error:
+        _write_update_check_diagnostic("启动版本检查失败，已允许继续使用", result=result)
+    if result.forced:
+        return _show_forced_update_dialog(result, opener=opener)
+    return True
+
+
 def main() -> int:
    if QT_IMPORT_ERROR is not None:
        print(f"{APP_NAME} GUI 无法启动：当前 Python 环境未安装 PySide6。")
@@ -59,6 +134,8 @@ def main() -> int:
    except appconfig.ConfigError as exc:
        QMessageBox.critical(None, "启动配置错误", str(exc))
        return 1
+    if not _run_startup_update_gate():
+        return 1
    window = MainWindow()
    window.show()
    return app.exec()
@@ -0,0 +1,189 @@
+"""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 .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 = ""
+    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 = ""
+    message: str = ""
+    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(),
+        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 = (fetcher or fetch_update_payload)(url, timeout)
+        info = parse_update_info(_decode_payload(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,
+            message=info.message,
+        )
+    except Exception as exc:
+        return UpdateCheckResult(
+            current_version=current_version,
+            checked=True,
+            forced=False,
+            error=f"启动版本检查失败，已允许继续使用：{exc}",
+        )
@@ -3,6 +3,7 @@
 APP_NAME = "蝦皮圈優化助手"
 APP_CODE_NAME = "cmshopee"
 APP_VERSION = "0.1.0"
+APP_UPDATE_CHECK_URL = "https://cm.833729.com/api/v1/client/releases/latest?platform=windows"


 def display_name() -> str: