server-up/server-up/core/updater.py

"""Update-check tegen een Forgejo/GitHub Releases-API (stdlib only).

Twee kanalen:

* **stable** — alleen echte releases (`v0.5.10`)
* **beta**   — ook pre-releases (`v0.5.10-beta1`), zoals `release.yml` die
               markeert voor tags met een streepje erin

Een beta-gebruiker ziet dus zowel betas als stabiele releases; wie op stable zit
ziet alleen releases. Schakel je van beta terug naar stable terwijl je een beta
draait, dan wordt de eerstvolgende stabiele release aangeboden zodra die hoger
is (`0.5.10-beta1` < `0.5.10`).
"""
from __future__ import annotations

import json
import re
import threading
import time
import urllib.request

import core as cfg

_SEMVER = re.compile(r"^v?(\d+)\.(\d+)\.(\d+)(?:[-+.](.+))?$")

CHANNELS = ("stable", "beta")

# Resultaat van de laatste check, zodat niet elke paginalading een
# netwerkverzoek doet. De UI kan met force=True alsnog verversen.
_CACHE: dict = {}
_CACHE_TS: float = 0.0
_CACHE_TTL = 3600
_LOCK = threading.Lock()


def _pre_key(pre: str):
    parts = [p for p in re.split(r"[.+-]", pre) if p] if pre else []
    return tuple((0, int(p)) if p.isdigit() else (1, p) for p in parts)


def _parse(v: str):
    """Naar vergelijkbare tuple. Release sorteert hoger dan pre-release."""
    if not v:
        return None
    m = _SEMVER.match(v.strip())
    if not m:
        return None
    pre = m.group(4) or ""
    return (int(m.group(1)), int(m.group(2)), int(m.group(3)),
            1 if not pre else 0, _pre_key(pre))


def channel() -> str:
    """Het ingestelde kanaal, met migratie vanaf de oude boolean."""
    c = cfg.load()
    ch = (c.get("UPDATE_CHANNEL") or "").strip().lower()
    if ch in CHANNELS:
        return ch
    # Oude instelling: UPDATE_INCLUDE_PRERELEASE=true betekende 'beta'.
    if c.get("UPDATE_INCLUDE_PRERELEASE"):
        return "beta"
    return "stable"


def is_prerelease(version: str) -> bool:
    t = _parse(version)
    return bool(t and t[3] == 0)


def image_tag(release_tag: str) -> str:
    """Vertaal een Forgejo-release-tag naar de tag uit onze registry.

    Git-tags heten volgens afspraak `v0.8.13`; de build-workflows verwijderen
    die voorloop-v en pushen `:0.8.13`. Alleen een geldige semver wordt
    aangepast, zodat een handmatige registry-tag zoals `beta` intact blijft.
    """
    tag = (release_tag or "").strip()
    return tag[1:] if tag.startswith("v") and _parse(tag) else tag


def api_url(url: str) -> str:
    """Maak van een geplakt adres het API-adres.

    Wie in Forgejo of GitHub naar zijn releases kijkt en dat adres uit de
    adresbalk plakt, geeft het adres van een wébpagina op. Dat levert HTML op,
    en de enige melding die je daarna kreeg was `JSONDecodeError: Expecting
    value: line 1 column 1` — waaruit niemand opmaakt dat er `/api/v1/` had
    moeten staan. Dus rekenen we het om in plaats van erover te vallen.

        https://git.voorbeeld.nl/bes-r/server-up/releases
          → https://git.voorbeeld.nl/api/v1/repos/bes-r/server-up/releases
        https://github.com/bes-r/server-up/releases
          → https://api.github.com/repos/bes-r/server-up/releases
    """
    schoon = (url or "").strip().rstrip("/")
    if not schoon or "/api/" in schoon:
        return schoon
    m = re.match(r"^(https?)://([^/]+)/([^/]+)/([^/]+?)(?:/releases.*|\.git)?$",
                 schoon)
    if not m:
        return schoon
    schema, host, eigenaar, repo = m.groups()
    if host.lower() in ("github.com", "www.github.com"):
        return f"https://api.github.com/repos/{eigenaar}/{repo}/releases"
    return f"{schema}://{host}/api/v1/repos/{eigenaar}/{repo}/releases"


def _releases(url: str) -> list[dict]:
    req = urllib.request.Request(
        url, headers={"Accept": "application/json", "User-Agent": "server-up"})
    with urllib.request.urlopen(req, timeout=8) as r:
        rauw = r.read().decode("utf-8", "replace")
    try:
        data = json.loads(rauw)
    except ValueError:
        # Bijna altijd een webpagina in plaats van de API. Zeg dát, in plaats
        # van de gebruiker met een JSON-fout te laten zitten.
        raise ValueError(
            f"{url} geeft geen JSON terug maar een webpagina. Vul het "
            "API-adres in, met /api/v1/repos/ erin.") from None
    return data if isinstance(data, list) else [data]


def check(current: str, force: bool = False) -> dict:
    """Zoek de nieuwste release in het ingestelde kanaal.

    Het resultaat wordt een uur gecached; `force=True` omzeilt de cache.
    """
    global _CACHE, _CACHE_TS
    with _LOCK:
        if not force and _CACHE and (time.time() - _CACHE_TS) < _CACHE_TTL:
            if _CACHE.get("current") == current:
                return dict(_CACHE)

    c = cfg.load()
    url = api_url(c.get("UPDATE_API_URL") or "")
    ch = channel()
    res = {"enabled": bool(url), "current": current, "channel": ch,
           "current_is_prerelease": is_prerelease(current),
           "latest": None, "latest_tag": "", "image_tag": "",
           "update_available": False,
           "url": "", "notes": "", "prerelease": False, "error": "",
           "checked_at": time.time()}
    if not url:
        return res

    try:
        rels = _releases(url)
    except Exception as e:
        res["error"] = f"{type(e).__name__}: {e}"
        return res

    best, best_t = None, None
    for rel in rels:
        if not isinstance(rel, dict) or rel.get("draft"):
            continue
        # Op stable tellen pre-releases niet mee; op beta wél.
        if rel.get("prerelease") and ch != "beta":
            continue
        t = _parse(rel.get("tag_name") or rel.get("name") or "")
        if not t:
            continue
        if best_t is None or t > best_t:
            best, best_t = rel, t
    if not best:
        res["error"] = f"Geen releases gevonden in kanaal '{ch}'"
        return res

    tag = (best.get("tag_name") or best.get("name") or "")
    res["latest_tag"] = tag
    res["image_tag"] = image_tag(tag)
    res["latest"] = res["image_tag"]
    res["url"] = best.get("html_url") or best.get("url") or ""
    res["notes"] = best.get("body") or ""
    res["prerelease"] = bool(best.get("prerelease"))
    cur_t = _parse(current)
    if cur_t and best_t and best_t > cur_t:
        res["update_available"] = True

    with _LOCK:
        _CACHE = dict(res)
        _CACHE_TS = time.time()
    return res


def invalidate():
    """Gooi de cache weg (na een instellingswijziging of na een update)."""
    global _CACHE, _CACHE_TS
    with _LOCK:
        _CACHE = {}
        _CACHE_TS = 0.0