buff3.0/backend-client/app/services/client_version_service.py

"""
客户端版本策略服务
"""
from __future__ import annotations

import re
from typing import Dict, Any, Optional, Tuple

from sqlalchemy.orm import Session

from app.models.client_app_version import ClientAppVersion
from app.models.system_config import SystemConfig

DEFAULT_PLATFORM = "windows"
DEFAULT_CHANNEL = "stable"
DEFAULT_CHECK_INTERVAL_SECONDS = 60


def _normalize_platform(platform: Optional[str]) -> str:
    value = (platform or "").strip().lower()
    if value in ("windows", "win32", "win", "darwin", "mac", "macos", "linux"):
        if value in ("darwin", "mac"):
            return "macos"
        if value in ("win32", "win"):
            return "windows"
        return value
    return DEFAULT_PLATFORM


def _normalize_channel(channel: Optional[str]) -> str:
    value = (channel or "").strip().lower()
    return value if value else DEFAULT_CHANNEL


def _parse_version(version: Optional[str]) -> Optional[Tuple[int, ...]]:
    text = (version or "").strip()
    if not text:
        return None
    parts = [p for p in re.split(r"[.\-+_]", text) if p != ""]
    parsed = []
    for p in parts:
        m = re.match(r"^\d+", p)
        if not m:
            break
        parsed.append(int(m.group(0)))
    if not parsed:
        return None
    return tuple(parsed)


def _compare_version(a: Optional[str], b: Optional[str]) -> int:
    """
    比较版本号:
    -1: a < b
     0: a == b
     1: a > b
    """
    ta = _parse_version(a)
    tb = _parse_version(b)
    if ta is None and tb is None:
        return 0
    if ta is None:
        return -1
    if tb is None:
        return 1
    max_len = max(len(ta), len(tb))
    aa = list(ta) + [0] * (max_len - len(ta))
    bb = list(tb) + [0] * (max_len - len(tb))
    if aa < bb:
        return -1
    if aa > bb:
        return 1
    return 0


class ClientVersionService:
    def __init__(self, db: Session):
        self.db = db

    def get_check_interval_seconds(self) -> int:
        row = self.db.query(SystemConfig).filter(
            SystemConfig.config_key == "client_version_check_interval_seconds",
            SystemConfig.app_name.contains("backend-client"),
            SystemConfig.is_deleted == 0
        ).first()
        if not row or not row.config_value:
            return DEFAULT_CHECK_INTERVAL_SECONDS
        try:
            seconds = int(row.config_value)
            return seconds if seconds > 0 else DEFAULT_CHECK_INTERVAL_SECONDS
        except (TypeError, ValueError):
            return DEFAULT_CHECK_INTERVAL_SECONDS

    def get_latest_published(self, platform: Optional[str], channel: Optional[str]) -> Optional[ClientAppVersion]:
        pf = _normalize_platform(platform)
        ch = _normalize_channel(channel)
        rows = self.db.query(ClientAppVersion).filter(
            ClientAppVersion.platform == pf,
            ClientAppVersion.channel == ch,
            ClientAppVersion.status == "published",
            ClientAppVersion.is_deleted == 0
        ).all()
        if not rows:
            return None
        return max(rows, key=lambda r: _parse_version(r.version) or (0,))

    def build_check_result(
        self,
        client_version: Optional[str],
        platform: Optional[str],
        channel: Optional[str],
    ) -> Dict[str, Any]:
        latest = self.get_latest_published(platform, channel)
        interval = self.get_check_interval_seconds()
        if not latest:
            return {
                "latest_version": None,
                "latest_build": None,
                "min_supported_version": None,
                "has_update": False,
                "is_force_update": False,
                "update_title": None,
                "release_notes_html": "",
                "download_url": "",
                "published_at": None,
                "check_interval_seconds": interval,
                "platform": _normalize_platform(platform),
                "channel": _normalize_channel(channel),
            }

        current = (client_version or "").strip()
        latest_version = latest.version
        min_supported = (latest.min_supported_version or "").strip() or latest.version
        has_update = _compare_version(current, latest_version) < 0
        force_by_min = _compare_version(current, min_supported) < 0
        is_force = bool(latest.is_force_update or force_by_min)

        return {
            "latest_version": latest_version,
            "latest_build": latest.build_number,
            "min_supported_version": min_supported,
            "has_update": has_update,
            "is_force_update": bool(is_force and has_update),
            "update_title": latest.release_title or f"发现新版本 {latest_version}",
            "release_notes_html": latest.release_notes_html or "",
            "download_url": latest.download_url or "",
            "published_at": latest.published_at.isoformat() if latest.published_at else None,
            "check_interval_seconds": interval,
            "platform": latest.platform,
            "channel": latest.channel,
        }

    def should_block_client(
        self,
        client_version: Optional[str],
        platform: Optional[str],
        channel: Optional[str],
    ) -> Tuple[bool, Dict[str, Any]]:
        result = self.build_check_result(client_version, platform, channel)
        return bool(result.get("is_force_update")), result