RedBear-OS/local/scripts/lib/external_version_lib.py

"""Shared upstream-version resolution helpers for Red Bear external-package tooling.

Consumers (add local/scripts/lib to sys.path, then import):
  - local/scripts/check-external-versions.sh (embedded Python)
  - local/scripts/bump-graphics-recipes.sh (embedded Python)

Single source of truth for: index fetching (cache + fallback), version sort
keys, sed-style version transforms, index resolution, tar-template rendering,
group member helpers, recipe.toml source pins, and git ls-remote tag
resolution. Behavior changes belong here, not in the consumers.
"""

import hashlib
import os
import re
import subprocess
from pathlib import Path
from typing import Optional

_CACHE_DIR: Optional[Path] = None
_NO_FETCH = False


def configure(cache_dir: Path, no_fetch: bool) -> None:
    global _CACHE_DIR, _NO_FETCH
    _CACHE_DIR = Path(cache_dir)
    _NO_FETCH = bool(no_fetch)


def cache_key_for(url: str) -> Path:
    assert _CACHE_DIR is not None, "configure() must be called before fetch use"
    safe = re.sub(r'[^a-zA-Z0-9_.-]', '_', url)
    if len(safe) > 180:
        # Stable hash (blake2b) — NOT Python's hash(), which is randomized
        # per-process via PYTHONHASHSEED and would orphan cache entries.
        digest = hashlib.blake2b(url.encode(), digest_size=8).hexdigest()
        safe = safe[:80] + "___" + digest + "___" + safe[-80:]
    return _CACHE_DIR / f"{safe}.cache"


def fetch(url: str):
    """Return (text, source); source is fetch/cache/cache-fallback*/error."""
    assert _CACHE_DIR is not None, "configure() must be called before fetch use"
    cache = cache_key_for(url)
    if _NO_FETCH:
        if cache.exists():
            return cache.read_text(errors="replace"), "cache"
        return None, "no-cache"
    try:
        result = subprocess.run(
            ["curl", "-sLf", "--retry", "3", "--retry-delay", "2",
             "--connect-timeout", "20", "--max-time", "90", url],
            capture_output=True, text=True, timeout=120,
        )
        if result.returncode == 0 and result.stdout:
            _CACHE_DIR.mkdir(parents=True, exist_ok=True)
            tmp = cache.with_suffix(cache.suffix + ".tmp")
            tmp.write_text(result.stdout)
            os.replace(str(tmp), str(cache))
            return result.stdout, "fetch"
        if cache.exists():
            return cache.read_text(errors="replace"), "cache-fallback"
        err = (result.stderr or "").strip()
        return None, err or f"curl-exit-{result.returncode}"
    except subprocess.TimeoutExpired:
        if cache.exists():
            return cache.read_text(errors="replace"), "cache-fallback-timeout"
        return None, "timeout"
    except Exception as e:
        if cache.exists():
            return cache.read_text(errors="replace"), "cache-fallback-error"
        return None, str(e)


def version_sort_key(v: str):
    """Sort key that orders pre-releases BEFORE the corresponding final.

    Handles forms like 1.0.0, 1.0.0-rc1, 1.0.0-alpha1, 1.0.0-beta2.
    A final release sorts after any pre-release of the same X.Y.Z base.
    """
    m = re.match(r'^(\d+(?:[._]\d+)*)(?:[-.]([0-9A-Za-z]+))?$', v)
    if not m:
        parts = re.split(r'[._+-]', v)
        return [(1, 0, p) if not p.isdigit() else (0, int(p), "") for p in parts]
    main_str, pre = m.group(1), m.group(2)
    main_key = [(0, int(p), "") if p.isdigit() else (1, 0, p)
                for p in re.split(r'[._]', main_str)]
    if pre is None or pre == "":
        pre_key = [(1, 0, "zzzzzz_final")]
    else:
        pre_key = [(-1, 0, pre)]
    return main_key + pre_key


def apply_ver_transform(ver: str, transform: str) -> str:
    """Apply a sed-style transform like 's/-/./g' to a version token."""
    m = re.match(r'^s/(.*)/(.*)/([gi]*)$', transform)
    if not m:
        return ver
    pat, repl, flags = m.group(1), m.group(2), m.group(3)
    count = 0 if 'g' in flags else 1
    try:
        return re.sub(pat, repl, ver, count=count)
    except re.error:
        return ver


def resolve_index(index_url: str, regex_str: str, resolve_kind: str,
                  ver_transform: Optional[str] = None):
    """Resolve latest version vars from an index page -> (vars, source, error)."""
    text, src = fetch(index_url)
    if text is None:
        return None, src, "fetch-failed"
    try:
        pattern = re.compile(regex_str)
    except re.error as e:
        return None, src, f"bad-regex: {e}"
    matches = pattern.findall(text)
    if not matches:
        return None, src, "no-match"
    norm = []
    for mm in matches:
        if isinstance(mm, tuple):
            mm = "-".join(str(x) for x in mm)
        norm.append(mm)
    unique = list(dict.fromkeys(norm))
    unique.sort(key=version_sort_key, reverse=True)
    best = unique[0]
    if ver_transform:
        best = apply_ver_transform(best, ver_transform)
    if resolve_kind == "majmin":
        return {"majmin": best, "ver": f"{best}.0"}, src, None
    parts = best.split(".")
    majmin = ".".join(parts[:2]) if len(parts) >= 2 else best
    return {"majmin": majmin, "ver": best}, src, None


def render_template(template: str, variables: dict) -> str:
    out = template
    for k in ("ver", "majmin", "name"):
        out = out.replace("{" + k + "}", variables.get(k, ""))
    return out


def member_url_name(group: dict, member: str) -> str:
    overrides = group.get("member_url_names", {}) or {}
    if member in overrides:
        return overrides[member]
    strip = group.get("url_name_strip_prefix", "") or ""
    if strip and member.startswith(strip):
        return member[len(strip):]
    return member


def member_template(group: dict, member: str) -> str:
    mts = group.get("member_templates", {}) or {}
    if member in mts and "tar_template" in mts[member]:
        return mts[member]["tar_template"]
    return group["tar_template"]


def read_recipe_text(recipe_path: Path) -> str:
    toml = recipe_path / "recipe.toml"
    if not toml.exists():
        return ""
    return toml.read_text(errors="replace")


def current_tar_url(text: str):
    m = re.search(r'^\s*tar\s*=\s*"([^"]+)"', text, re.MULTILINE)
    return m.group(1) if m else None


def current_rev(text: str):
    m = re.search(r'^\s*rev\s*=\s*"([^"]+)"', text, re.MULTILINE)
    return m.group(1) if m else None


def resolve_git_rev(git_url: str, tag_regex_str: str):
    """Resolve latest tag + commit sha via git ls-remote --tags."""
    try:
        result = subprocess.run(
            ["git", "ls-remote", "--tags", git_url],
            capture_output=True, text=True, timeout=90,
        )
    except subprocess.TimeoutExpired:
        return None, "timeout"
    except Exception as e:
        return None, str(e)
    if result.returncode != 0:
        return None, (result.stderr or "git-error").strip()
    try:
        pattern = re.compile(tag_regex_str)
    except re.error as e:
        return None, f"bad-regex: {e}"
    shas = {}
    for line in result.stdout.splitlines():
        parts = line.split("\t")
        if len(parts) != 2:
            continue
        sha, ref = parts
        if not ref.startswith("refs/tags/"):
            continue
        tag = ref[len("refs/tags/"):]
        deref = tag.endswith("^{}")
        clean = tag[:-3] if deref else tag
        if pattern.fullmatch(clean):
            if clean not in shas or deref:
                shas[clean] = sha.strip()
    if not shas:
        return None, "no-tags-matched"
    best = max(shas.keys(), key=version_sort_key)
    return {"tag": best, "sha": shas[best]}, None