src/webui/services/system.py

Rama main · réplica de lectura del repositorio original.

"""System metadata and update checks."""

from __future__ import annotations

import logging
import os
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable

import aiohttp

from src.version import DEFAULT_UPDATE_REPOSITORY, __version__
from src.runtime_diagnostics import build_runtime_log_archive
from src.runtime_logging import clear_runtime_logs as clear_logs
from src.runtime_logging import runtime_log_status as log_status

logger = logging.getLogger("trpg")

GITHUB_API = "https://api.github.com"
_VERSION_RE = re.compile(
    r"^\s*[vV]?(\d+)(?:\.(\d+))?(?:\.(\d+))?"
    r"(?:-([0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*))?"
    r"(?:\+[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*)?\s*$"
)


class NoReleaseError(RuntimeError):
    """Raised when the repository exists but has no public releases yet."""


@dataclass(frozen=True)
class SystemDependencies:
    data_dir: Path
    config_state: Callable[[], dict[str, Any]]
    proxy_url: Callable[[], str]
    mirrors: Callable[[], Any | None]


def runtime_log_status(data_dir: Path) -> dict[str, Any]:
    return log_status(data_dir)


def clear_runtime_logs(data_dir: Path) -> dict[str, Any]:
    return clear_logs(data_dir)


def export_runtime_logs(data_dir: Path) -> tuple[bytes, int]:
    return build_runtime_log_archive(data_dir)


def _parse_version(value: str) -> tuple[tuple[int, int, int], tuple[str, ...] | None] | None:
    match = _VERSION_RE.match(value or "")
    if not match:
        return None
    major, minor, patch, prerelease = match.groups()
    return (
        (int(major), int(minor or 0), int(patch or 0)),
        tuple(prerelease.split(".")) if prerelease else None,
    )


def _compare_prerelease(left: tuple[str, ...] | None, right: tuple[str, ...] | None) -> int:
    if left is None or right is None:
        if left is right:
            return 0
        return 1 if left is None else -1
    for left_part, right_part in zip(left, right):
        if left_part == right_part:
            continue
        left_numeric = left_part.isdigit()
        right_numeric = right_part.isdigit()
        if left_numeric and right_numeric:
            return 1 if int(left_part) > int(right_part) else -1
        if left_numeric != right_numeric:
            return -1 if left_numeric else 1
        return 1 if left_part > right_part else -1
    return (len(left) > len(right)) - (len(left) < len(right))


def is_newer_version(latest: str, current: str) -> bool:
    latest_version = _parse_version(latest)
    current_version = _parse_version(current)
    if latest_version is None or current_version is None:
        return False
    latest_core, latest_prerelease = latest_version
    current_core, current_prerelease = current_version
    if latest_core != current_core:
        return latest_core > current_core
    return _compare_prerelease(latest_prerelease, current_prerelease) > 0


async def check_updates(
    dependencies: SystemDependencies,
    include_prerelease: bool | None = None,
) -> dict[str, Any]:
    repo = str(os.getenv("TRPG_UPDATE_REPOSITORY") or DEFAULT_UPDATE_REPOSITORY).strip()
    if not repo or "/" not in repo:
        return {
            "ok": False,
            "error": "更新仓库配置无效",
            "current_version": __version__,
            "repository": repo,
            "update_available": False,
        }

    if include_prerelease is None:
        config_state = dependencies.config_state()
        include_prerelease = str(config_state.get("update_channel") or "stable") == "preview"
    channel = "preview" if include_prerelease else "stable"

    check_source: dict[str, Any] = {"mode": "github-api", "mirror_name": "GitHub API"}
    try:
        proxy_url = dependencies.proxy_url()
        release, check_source = await _fetch_release_with_mirrors(
            dependencies.mirrors(), repo, include_prerelease, proxy_url,
        )
    except NoReleaseError as exc:
        return {
            "ok": True,
            "message": str(exc),
            "current_version": __version__,
            "repository": repo,
            "update_available": False,
            "no_release": True,
            "channel": channel,
            "releases_url": f"https://github.com/{repo}/releases",
            "source_url": f"https://github.com/{repo}",
            "install_hint": _install_hint(repo),
            "check_source": check_source,
        }
    except Exception as exc:
        logger.warning("检查更新失败: %s", exc)
        return {
            "ok": False,
            "error": str(exc),
            "current_version": __version__,
            "repository": repo,
            "update_available": False,
        }

    latest_version = str(release.get("tag_name") or "").lstrip("vV")
    latest = {
        "version": latest_version,
        "tag_name": release.get("tag_name", ""),
        "name": release.get("name", ""),
        "body": release.get("body", ""),
        "html_url": release.get("html_url", f"https://github.com/{repo}/releases"),
        "published_at": release.get("published_at", ""),
        "prerelease": bool(release.get("prerelease")),
        "assets": [
            {
                "name": asset.get("name", ""),
                "download_url": asset.get("browser_download_url", ""),
                "size": asset.get("size", 0),
            }
            for asset in release.get("assets", [])
            if isinstance(asset, dict)
        ],
    }
    return {
        "ok": True,
        "current_version": __version__,
        "repository": repo,
        "update_available": is_newer_version(latest_version, __version__),
        "no_release": False,
        "channel": channel,
        "latest": latest,
        "release_url": latest["html_url"],
        "releases_url": f"https://github.com/{repo}/releases",
        "source_url": f"https://github.com/{repo}",
        "install_hint": _install_hint(repo),
        "check_source": check_source,
    }


async def _fetch_release(repo: str, include_prerelease: bool, proxy_url: str = "") -> dict[str, Any]:
    endpoint = f"{GITHUB_API}/repos/{repo}/releases"
    if not include_prerelease:
        endpoint += "/latest"

    headers = {
        "Accept": "application/vnd.github+json",
        "User-Agent": "DiceFrame update checker",
    }
    timeout = aiohttp.ClientTimeout(total=12)
    async with aiohttp.ClientSession(timeout=timeout, headers=headers) as session:
        request_kwargs = {"proxy": proxy_url} if proxy_url else {}
        async with session.get(endpoint, **request_kwargs) as resp:
            data = await resp.json(content_type=None)
            if resp.status == 404:
                raise NoReleaseError("暂无公开 Release")
            if resp.status >= 400:
                message = data.get("message") if isinstance(data, dict) else await resp.text()
                raise RuntimeError(f"GitHub 返回 HTTP {resp.status}: {message}")
            if not include_prerelease:
                return data
            if not isinstance(data, list):
                raise RuntimeError("GitHub Release 返回格式异常")
            for release in data:
                if isinstance(release, dict) and not release.get("draft"):
                    return release
    raise RuntimeError("没有可用的 Release")


async def _fetch_release_with_mirrors(
    mirrors: Any | None,
    repo: str,
    include_prerelease: bool,
    proxy_url: str = "",
) -> tuple[dict[str, Any], dict[str, Any]]:
    api_path = f"/repos/{repo}/releases"
    if not include_prerelease:
        api_path += "/latest"
    if mirrors is not None:
        result = await mirrors.fetch_github_api(api_path, official_first=True)
        if result.ok and isinstance(result.data, str):
            try:
                data = await _parse_release_payload(result.data, include_prerelease)
                source = result.to_dict()
                source.pop("data", None)
                return data, {"mode": "mirror", **source}
            except Exception as exc:
                logger.warning("镜像源返回的 Release 数据无法解析，回退 GitHub API：%s", exc)
        logger.warning("通过镜像源检查更新失败，回退 GitHub API：%s", result.error)
    release = await _fetch_release(repo, include_prerelease, proxy_url)
    return release, {"mode": "github-api", "mirror_name": "GitHub API"}


async def _parse_release_payload(payload: str, include_prerelease: bool) -> dict[str, Any]:
    import json

    data = json.loads(payload)
    if isinstance(data, dict) and data.get("message") == "Not Found":
        raise NoReleaseError("暂无公开 Release")
    if not include_prerelease:
        if not isinstance(data, dict):
            raise RuntimeError("GitHub Release 返回格式异常")
        return data
    if not isinstance(data, list):
        raise RuntimeError("GitHub Release 返回格式异常")
    for release in data:
        if isinstance(release, dict) and not release.get("draft"):
            return release
    raise RuntimeError("没有可用的 Release")


def _install_hint(repo: str) -> dict[str, str]:
    return {
        "windows": "下载新版源码包或 Release 附件，保留 data/ 目录后替换程序文件，再重新运行 web_ui.bat。",
        "docker": "Compose 部署请先进入存放 docker-compose.yml 的目录，再拉取镜像并重新 docker compose up -d；docker run 部署需重新创建容器，源码镜像需用新版源码重新 build。",
        "source": f"源码用户可从 https://github.com/{repo}/releases 下载新版，升级前先备份 data/。",
    }


class SystemService:
    """Runtime diagnostics and release checks with explicit host dependencies."""

    def __init__(self, dependencies: SystemDependencies) -> None:
        self._dependencies = dependencies

    async def check_updates(
        self, include_prerelease: bool | None = None,
    ) -> dict[str, Any]:
        return await check_updates(self._dependencies, include_prerelease)

    def runtime_log_status(self) -> dict[str, Any]:
        return runtime_log_status(self._dependencies.data_dir)

    def clear_runtime_logs(self) -> dict[str, Any]:
        return clear_runtime_logs(self._dependencies.data_dir)

    def export_runtime_logs(self) -> tuple[bytes, int]:
        return export_runtime_logs(self._dependencies.data_dir)