alexei.dolgolyov
4ef11c8f00
Lint & Test / test (push) Successful in 10s
- Rename GITEA_TOKEN to DEPLOY_TOKEN in release workflow - Extract shared version detection into build-common.sh - Use importlib.metadata for runtime version instead of hardcoded string - Use PEP 440 parsing (packaging lib) for update version comparison - Add packaging>=23.0 to dependencies - Fix update banner close button alignment (CSS) - Update CLAUDE.md with versioning docs and frontend rebuild notes
127 lines
4.1 KiB
Python
127 lines
4.1 KiB
Python
"""Provider-agnostic update checker service."""
|
|
|
|
import asyncio
|
|
import logging
|
|
import re
|
|
from typing import Any, Optional
|
|
|
|
from packaging.version import Version
|
|
|
|
from .release_provider import ReleaseProvider
|
|
from .websocket_manager import ws_manager
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_PRE_PATTERN = re.compile(
|
|
r"^(\d+\.\d+\.\d+)[-.]?(alpha|beta|rc)[.-]?(\d+)$", re.IGNORECASE
|
|
)
|
|
_PRE_MAP = {"alpha": "a", "beta": "b", "rc": "rc"}
|
|
|
|
|
|
def _parse_version(raw: str) -> Version:
|
|
"""Normalize a version tag to PEP 440 for correct comparison.
|
|
|
|
Examples:
|
|
v0.3.0-alpha.1 → 0.3.0a1 (pre-release, sorts below 0.3.0)
|
|
v0.3.0-rc.3 → 0.3.0rc3
|
|
v1.0.0 → 1.0.0
|
|
"""
|
|
cleaned = raw.lstrip("v").strip()
|
|
m = _PRE_PATTERN.match(cleaned)
|
|
if m:
|
|
base, pre_label, pre_num = m.group(1), m.group(2).lower(), m.group(3)
|
|
cleaned = f"{base}{_PRE_MAP[pre_label]}{pre_num}"
|
|
return Version(cleaned)
|
|
|
|
|
|
class UpdateChecker:
|
|
"""Periodically checks for new releases using a ReleaseProvider."""
|
|
|
|
def __init__(self, provider: ReleaseProvider, current_version: str) -> None:
|
|
self._provider = provider
|
|
self._current_version = current_version
|
|
self._current_parsed = _parse_version(current_version)
|
|
self._task: Optional[asyncio.Task] = None
|
|
self._cached_update: Optional[dict[str, Any]] = None
|
|
|
|
@property
|
|
def cached_update(self) -> Optional[dict[str, Any]]:
|
|
"""Return the cached update info, or None if up-to-date."""
|
|
return self._cached_update
|
|
|
|
async def check_for_update(self) -> Optional[dict[str, Any]]:
|
|
"""Check for a newer release.
|
|
|
|
Returns:
|
|
Dict with current/latest/url if an update exists, None otherwise.
|
|
"""
|
|
release = await self._provider.get_latest_release()
|
|
if release is None:
|
|
return None
|
|
|
|
latest_parsed = _parse_version(release.version)
|
|
if latest_parsed <= self._current_parsed:
|
|
return None
|
|
|
|
return {
|
|
"current": self._current_version,
|
|
"latest": release.version,
|
|
"url": release.url,
|
|
}
|
|
|
|
async def start(self, interval: int) -> None:
|
|
"""Start periodic update checking.
|
|
|
|
Checks immediately on start, then every `interval` seconds.
|
|
"""
|
|
if self._task is not None:
|
|
return
|
|
|
|
self._task = asyncio.create_task(self._check_loop(interval))
|
|
logger.info("Update checker started (interval: %ds)", interval)
|
|
|
|
async def stop(self) -> None:
|
|
"""Stop periodic update checking."""
|
|
if self._task is not None:
|
|
self._task.cancel()
|
|
try:
|
|
await self._task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
self._task = None
|
|
logger.info("Update checker stopped")
|
|
|
|
async def _check_loop(self, interval: int) -> None:
|
|
"""Background loop that checks for updates periodically."""
|
|
# Initial check with a small delay to let the server finish starting
|
|
await asyncio.sleep(5)
|
|
|
|
while True:
|
|
try:
|
|
update = await self.check_for_update()
|
|
|
|
if update and update != self._cached_update:
|
|
self._cached_update = update
|
|
logger.info(
|
|
"New version available: %s → %s (%s)",
|
|
update["current"],
|
|
update["latest"],
|
|
update["url"],
|
|
)
|
|
await ws_manager.broadcast(
|
|
{"type": "update_available", "data": update}
|
|
)
|
|
elif update is None and self._cached_update is not None:
|
|
# Version was updated (or release removed) — clear cache
|
|
self._cached_update = None
|
|
|
|
except asyncio.CancelledError:
|
|
break
|
|
except Exception as e:
|
|
logger.warning("Update check failed: %s", e)
|
|
|
|
try:
|
|
await asyncio.sleep(interval)
|
|
except asyncio.CancelledError:
|
|
break
|