""" Application update checks against SageTube's own releases ========================================================= SageTube is a fork of oop7/YTSage. It keeps upstream's git history, its internal `ytsage` package name and its inherited tags -- but it is a different program with a different release stream, and this module is the line between them. **This checks SageTube. Nothing here touches upstream.** The three other updaters in this codebase legitimately track their own upstreams and are deliberately left alone: - yt-dlp binary -> gui/ytsage_gui_dialogs/ytsage_dialogs_update.py - Deno runtime -> core/ytsage_deno.py - ffmpeg -> utils/ytsage_constants.py They share only the word "update". Gitea's release API is shaped like GitHub's -- `tag_name`, `body`, `html_url`, `draft`, `prerelease` -- so the consuming code needs no special cases. The repository is anonymously readable, so no token is involved. Tag notes --------- The repository carries upstream's tags up to `v5.3.0b` alongside SageTube's own. `packaging` reads a trailing `b` as a beta marker (`5.3.0b` -> `5.3.0b0`), so those inherited tags are correctly treated as pre-releases and are invisible to a stable instance. Parsing is still defensive: one unparseable tag must not take the whole check down. """ from __future__ import annotations import time from dataclasses import dataclass from typing import Any, Dict, List, Optional import requests from packaging.version import InvalidVersion, Version, parse as parse_version from .. import __version__ from ..utils.ytsage_config_manager import ConfigManager from ..utils.ytsage_logger import logger API_BASE: str = "https://git.houmeres.sk/api/v1/repos/Houmeres/SageTube" RELEASES_PAGE: str = "https://git.houmeres.sk/Houmeres/SageTube/releases" DEFAULT_TIMEOUT: float = 8.0 #: Don't ask the forge more than once a day, however often the app is started. CHECK_INTERVAL_SECONDS: int = 24 * 60 * 60 #: How many releases to pull when the beta channel is on. BETA_PAGE_SIZE: int = 20 @dataclass(frozen=True) class Release: """One published release, normalised.""" version: Version tag: str url: str body: str prerelease: bool def _headers() -> Dict[str, str]: return { "Accept": "application/json", "User-Agent": f"SageTube/{__version__}", } def _to_release(payload: Dict[str, Any]) -> Optional[Release]: """Normalise one API object, or None if it is unusable.""" tag = (payload.get("tag_name") or "").strip() if not tag: return None try: version = parse_version(tag.lstrip("vV")) except (InvalidVersion, TypeError) as e: # A hand-made or inherited tag that isn't a version. Skip it rather # than letting it abort the whole check. logger.debug(f"Ignoring unparseable release tag {tag!r}: {e}") return None return Release( version=version, tag=tag, url=payload.get("html_url") or f"{RELEASES_PAGE}/tag/{tag}", body=(payload.get("body") or "").strip(), prerelease=bool(payload.get("prerelease")) or version.is_prerelease, ) def _get(path: str, timeout: float, **params: Any) -> Optional[Any]: """GET a JSON endpoint. Returns None on any failure -- never raises.""" try: response = requests.get( f"{API_BASE}{path}", headers=_headers(), params=params or None, timeout=timeout, ) except requests.RequestException as e: logger.debug(f"Update check request failed: {e}") return None if response.status_code == 404: # No release matches (e.g. /releases/latest on a repo whose only # releases are pre-releases). That is "nothing to offer", not an error. logger.debug("Update check: no matching release published.") return None if response.status_code != 200: logger.debug(f"Update check: API returned {response.status_code}") return None try: return response.json() except ValueError as e: logger.debug(f"Update check: malformed JSON: {e}") return None def fetch_latest(include_prerelease: bool = False, timeout: float = DEFAULT_TIMEOUT) -> Optional[Release]: """ The newest published release, or None if there is nothing or the check failed. With `include_prerelease` the whole release list is pulled and filtered here rather than trusting server-side filters, because the highest *version* is wanted -- not the most recently published, which is what the API orders by. """ if not include_prerelease: payload = _get("/releases/latest", timeout) if not isinstance(payload, dict): return None release = _to_release(payload) # /releases/latest should never return a draft or pre-release, but a # stable instance must not be handed one even if it does. if release is None or release.prerelease: return None return release payload = _get("/releases", timeout, limit=BETA_PAGE_SIZE, draft=False) if not isinstance(payload, list): return None candidates: List[Release] = [] for item in payload: if not isinstance(item, dict) or item.get("draft"): continue release = _to_release(item) if release is not None: candidates.append(release) if not candidates: return None return max(candidates, key=lambda r: r.version) def is_newer(release: Release, current_version: str) -> bool: """ Whether `release` should be offered over the running version. Honours the version the user chose to skip, so "Skip this version" is not silently undone by the next start. """ try: current = parse_version(str(current_version).lstrip("vV")) except (InvalidVersion, TypeError): logger.debug(f"Running version {current_version!r} is unparseable; skipping update check.") return False if release.version <= current: return False skipped = ConfigManager.get("skipped_update_version") if skipped and str(skipped) == str(release.version): logger.debug(f"Update {release.tag} available but skipped by the user.") return False return True def should_check(now: Optional[float] = None) -> bool: """Rate-limit to CHECK_INTERVAL_SECONDS, so a restart is not a new check.""" if now is None: now = time.time() try: last = float(ConfigManager.get("last_update_check") or 0) except (TypeError, ValueError): last = 0.0 # A clock moved backwards must not lock the check out until it catches up. if last > now: return True return (now - last) >= CHECK_INTERVAL_SECONDS def mark_checked(now: Optional[float] = None) -> None: ConfigManager.set("last_update_check", now if now is not None else time.time())