OmniNX/scripts/update-badge-data.py

#!/usr/bin/env python3
"""Schreibt .github/badges/data.json aus der Releases-API (OmniNX-Pack + NX_Firmware-Tag, Gitea/GitHub-kompatibel).

Download-Zähler: Summe aller *.zip-Assets (API). lifetime steigt nur bei current >= snapshot;
bei Asset-Tausch (current < snapshot) wird lifetime nicht reduziert.
"""

from __future__ import annotations

import json
import os
import sys
import urllib.error
import urllib.request

BADGE_PATH = os.path.join(
    os.path.dirname(os.path.dirname(__file__)),
    ".github",
    "badges",
    "data.json",
)


def api_get(url: str, token: str) -> bytes:
    req = urllib.request.Request(url)
    if token:
        req.add_header("Authorization", f"Bearer {token}")
    req.add_header("Accept", "application/json")
    with urllib.request.urlopen(req, timeout=120) as resp:
        return resp.read()


def fetch_releases(api_base: str, repo: str, token: str) -> list:
    out: list = []
    page = 1
    while True:
        url = f"{api_base}/repos/{repo}/releases?limit=100&page={page}"
        batch = json.loads(api_get(url, token))
        if not batch:
            break
        out.extend(batch)
        if len(batch) < 100:
            break
        page += 1
    return out


def _token_attempt_order(pack_token: str) -> list[str]:
    """Repos/branches the job token cannot see on Gitea often return 404; public /releases works without auth."""
    pat = os.environ.get("FIRMWARE_API_TOKEN", "").strip()
    order = [pat, "", pack_token] if pat else ["", pack_token]
    seen: set[str] = set()
    out: list[str] = []
    for t in order:
        if t in seen:
            continue
        seen.add(t)
        out.append(t)
    return out


def fetch_releases_with_token_fallback(
    api_base: str, repo: str, pack_token: str
) -> tuple[list | None, urllib.error.HTTPError | None]:
    last_err: urllib.error.HTTPError | None = None
    for t in _token_attempt_order(pack_token):
        try:
            return fetch_releases(api_base, repo, t), None
        except urllib.error.HTTPError as e:
            last_err = e
            if e.code in (401, 403, 404):
                continue
            raise
    return None, last_err


def sum_zip_downloads(releases: list) -> int:
    total = 0
    for rel in releases:
        for asset in rel.get("assets") or []:
            name = str(asset.get("name", ""))
            if name.endswith(".zip"):
                total += int(asset.get("download_count") or 0)
    return total


def first_non_draft_tag(releases: list) -> str | None:
    for rel in releases:
        if rel.get("draft"):
            continue
        tag = str(rel.get("tag_name") or "").strip()
        if tag:
            return tag
    return None


def load_existing_badge_data() -> dict:
    if not os.path.isfile(BADGE_PATH):
        return {}
    try:
        with open(BADGE_PATH, encoding="utf-8") as f:
            data = json.load(f)
            return data if isinstance(data, dict) else {}
    except (OSError, json.JSONDecodeError, TypeError):
        return {}


def load_existing_firmware() -> str | None:
    fw = load_existing_badge_data().get("switch_firmware")
    return str(fw).strip() if fw else None


def _int_or_none(value: object) -> int | None:
    if value is None:
        return None
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def load_existing_download_counters() -> tuple[int | None, int | None]:
    """Liest snapshot und lifetime; migriert alte release_zip_downloads_total."""
    data = load_existing_badge_data()
    snapshot = _int_or_none(data.get("release_zip_downloads_snapshot"))
    if snapshot is None:
        snapshot = _int_or_none(data.get("release_zip_downloads_total"))
    lifetime = _int_or_none(data.get("release_zip_downloads_lifetime"))
    if lifetime is None:
        lifetime = _int_or_none(data.get("release_zip_downloads_total"))
    return snapshot, lifetime


def update_download_lifetime(current: int, snapshot: int | None, lifetime: int | None) -> tuple[int, int]:
    """Gibt (neuer_snapshot, neue_lifetime) zurück."""
    if snapshot is None or lifetime is None:
        return current, current
    if current >= snapshot:
        lifetime += current - snapshot
    return current, lifetime


def main() -> int:
    api_base = os.environ.get("GITHUB_API_URL", "").rstrip("/")
    repo = os.environ.get("GITHUB_REPOSITORY", "")
    fw_repo = os.environ.get("FIRMWARE_REPOSITORY", "OmniNX/NX_Firmware").strip()
    token = os.environ.get("GITHUB_TOKEN", "")

    if not api_base or not repo:
        print("GITHUB_API_URL und GITHUB_REPOSITORY werden benötigt.", file=sys.stderr)
        return 1

    try:
        omninx_releases = fetch_releases(api_base, repo, token)
    except urllib.error.HTTPError as e:
        print(f"API: HTTP {e.code} — {e.reason}", file=sys.stderr)
        return 1

    if not omninx_releases:
        print("Keine Releases gefunden.", file=sys.stderr)
        return 1

    omninx_tag = first_non_draft_tag(omninx_releases)
    if not omninx_tag:
        print("Kein nicht-Draft-Release.", file=sys.stderr)
        return 1

    fw: str | None = None
    fw_releases, fw_err = fetch_releases_with_token_fallback(api_base, fw_repo, token)
    if fw_releases is not None:
        fw = first_non_draft_tag(fw_releases)
    elif fw_err is not None:
        print(f"Firmware-Repo ({fw_repo}): HTTP {fw_err.code} — {fw_err.reason}", file=sys.stderr)

    if not fw:
        fw = load_existing_firmware() or "unknown"

    current = sum_zip_downloads(omninx_releases)
    prev_snapshot, prev_lifetime = load_existing_download_counters()
    snapshot, lifetime = update_download_lifetime(current, prev_snapshot, prev_lifetime)

    data = {
        "switch_firmware": fw,
        "release_zip_downloads_snapshot": snapshot,
        "release_zip_downloads_lifetime": lifetime,
        "release_zip_downloads_total": current,
        "omninx_pack_version": omninx_tag,
    }

    with open(BADGE_PATH, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)
        f.write("\n")

    print(json.dumps(data, indent=2))
    return 0


if __name__ == "__main__":
    sys.exit(main())