alexei.dolgolyov
795a15cb8b
Lint & Test / test (push) Successful in 10s
- Abstract ReleaseProvider protocol for platform-agnostic version checking - GiteaReleaseProvider implementation using stdlib urllib - UpdateChecker service with periodic background checks and WS broadcast - Persistent dismissible banner in Web UI when a new version is detected - Health endpoint now returns cached update info - Configurable via update_check_enabled and update_check_interval settings - i18n support (EN/RU)
78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
"""Gitea release provider implementation."""
|
|
|
|
import json
|
|
import logging
|
|
import urllib.error
|
|
import urllib.request
|
|
from typing import Optional
|
|
|
|
from .release_provider import ReleaseInfo, ReleaseProvider
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Default repository coordinates
|
|
_DEFAULT_BASE_URL = "https://git.dolgolyov-family.by"
|
|
_DEFAULT_OWNER = "alexei.dolgolyov"
|
|
_DEFAULT_REPO = "media-player-server"
|
|
|
|
|
|
class GiteaReleaseProvider(ReleaseProvider):
|
|
"""Fetches the latest release from a Gitea repository."""
|
|
|
|
def __init__(
|
|
self,
|
|
base_url: str = _DEFAULT_BASE_URL,
|
|
owner: str = _DEFAULT_OWNER,
|
|
repo: str = _DEFAULT_REPO,
|
|
timeout: float = 10.0,
|
|
) -> None:
|
|
self._api_url = f"{base_url}/api/v1/repos/{owner}/{repo}/releases"
|
|
self._release_page_url = f"{base_url}/{owner}/{repo}/releases/tag"
|
|
self._timeout = timeout
|
|
|
|
async def get_latest_release(self) -> Optional[ReleaseInfo]:
|
|
"""Fetch the latest stable release from Gitea API.
|
|
|
|
Returns:
|
|
ReleaseInfo for the latest non-prerelease, or None on failure.
|
|
"""
|
|
import asyncio
|
|
|
|
try:
|
|
data = await asyncio.to_thread(self._fetch_releases)
|
|
except Exception as e:
|
|
logger.warning("Failed to check for updates: %s", e)
|
|
return None
|
|
|
|
if not data:
|
|
return None
|
|
|
|
# Find the first non-prerelease, non-draft release
|
|
for release in data:
|
|
if release.get("draft") or release.get("prerelease"):
|
|
continue
|
|
|
|
tag = release.get("tag_name", "")
|
|
version = tag.lstrip("v")
|
|
if not version:
|
|
continue
|
|
|
|
return ReleaseInfo(
|
|
version=version,
|
|
url=f"{self._release_page_url}/{tag}",
|
|
prerelease=False,
|
|
)
|
|
|
|
logger.debug("No stable releases found")
|
|
return None
|
|
|
|
def _fetch_releases(self) -> list[dict]:
|
|
"""Synchronous HTTP fetch of releases (run in thread)."""
|
|
url = f"{self._api_url}?limit=5"
|
|
req = urllib.request.Request(url, headers={"Accept": "application/json"})
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=self._timeout) as resp:
|
|
return json.loads(resp.read().decode())
|
|
except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError, OSError) as e:
|
|
raise RuntimeError(f"Gitea API request failed: {e}") from e
|