SageTube/ytsage/core/ytsage_app_update.py

"""
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())