cmshoppe/app/update_check.py

"""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}",
        )