takana/scripts/fuentes/fuentes-vigia.py

#!/usr/bin/env python3
"""Comprueba que las URLs de fuentes de las recetas siguen vivas. Ver ADR 0013 §3.

No descarga: pide cabeceras. Para `git` no hay HEAD que valga, así que usa `git ls-remote
--exit-code`, que es la pregunta equivalente («¿este remoto sigue ahí y responde?»).

Escribe `docs/state/fuentes-vigia.json`. NO falla si hay URLs muertas: su trabajo es INFORMAR.
Un vigía que aborta el latido en la primera URL caída deja de correr y se vuelve inútil justo
cuando más hace falta — que es cuando hay varias caídas.
"""
import json, os, re, subprocess, sys, glob
from concurrent.futures import ThreadPoolExecutor

ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
os.chdir(ROOT)
LIMITE = int(os.environ.get("LIMITE", "0"))
JOBS = int(os.environ.get("JOBS", "12"))
TIMEOUT = int(os.environ.get("TIMEOUT", "25"))

def fuentes():
    out = []
    for f in sorted(glob.glob("recipes/**/*.toml", recursive=True)):
        s = open(f, errors="ignore").read()
        nombre = (re.search(r'^\s*name\s*=\s*"([^"]+)"', s, re.M) or [None, os.path.basename(f)[:-5]])[1]
        t = re.search(r'^\s*tarball\s*=\s*"([^"]+)"', s, re.M)
        g = re.search(r'^\s*repo\s*=\s*"([^"]+)"', s, re.M)
        if t:   out.append({"receta": nombre, "fichero": f, "tipo": "tarball", "url": t.group(1)})
        elif g: out.append({"receta": nombre, "fichero": f, "tipo": "git", "url": g.group(1)})
    return out

def revisa(e):
    url, tipo = e["url"], e["tipo"]
    try:
        if tipo == "tarball":
            # -I pide cabeceras; algunos servidores no admiten HEAD y responden 405 -> se reintenta
            # con -r 0-0 (un rango de UN byte), que es un GET mínimo y no descarga el tarball.
            r = subprocess.run(["curl", "-sIL", "-o", "/dev/null", "-w", "%{http_code}",
                                "--connect-timeout", str(TIMEOUT), "--max-time", str(TIMEOUT * 2), url],
                               capture_output=True, text=True, timeout=TIMEOUT * 3)
            code = (r.stdout or "").strip()[-3:]
            if code in ("405", "501", "403"):
                r = subprocess.run(["curl", "-sL", "-r", "0-0", "-o", "/dev/null", "-w", "%{http_code}",
                                    "--connect-timeout", str(TIMEOUT), "--max-time", str(TIMEOUT * 2), url],
                                   capture_output=True, text=True, timeout=TIMEOUT * 3)
                code = (r.stdout or "").strip()[-3:]
            viva = code.startswith("2") or code.startswith("3")
            return {**e, "codigo": code or "000", "viva": viva}
        else:
            r = subprocess.run(["git", "ls-remote", "--exit-code", url, "HEAD"],
                               capture_output=True, text=True, timeout=TIMEOUT * 2)
            return {**e, "codigo": f"git:{r.returncode}", "viva": r.returncode == 0}
    except Exception as ex:
        return {**e, "codigo": f"error:{type(ex).__name__}", "viva": False}

def main():
    fs = fuentes()
    if LIMITE:
        fs = fs[:LIMITE]
    print(f"== vigía de fuentes: {len(fs)} recetas ({sum(1 for f in fs if f['tipo']=='tarball')} tarball, "
          f"{sum(1 for f in fs if f['tipo']=='git')} git), {JOBS} en paralelo", flush=True)
    with ThreadPoolExecutor(max_workers=JOBS) as ex:
        res = list(ex.map(revisa, fs))
    muertas = [r for r in res if not r["viva"]]
    # El informe se ordena por receta para que el diff de git sea legible entre corridas.
    inf = {
        "schema": "hammer-fuentes-vigia/1",
        "total": len(res),
        "vivas": len(res) - len(muertas),
        "muertas": len(muertas),
        "detalle_muertas": sorted(muertas, key=lambda r: r["receta"]),
    }
    os.makedirs("docs/state", exist_ok=True)
    with open("docs/state/fuentes-vigia.json", "w") as fh:
        json.dump(inf, fh, indent=1, ensure_ascii=False)
        fh.write("\n")
    print(f"== vivas {inf['vivas']} / {inf['total']}   MUERTAS {inf['muertas']}")
    for m in inf["detalle_muertas"][:40]:
        print(f"   {m['codigo']:>12}  {m['receta']:<24} {m['url'][:70]}")
    # Exit 0 SIEMPRE: informar, no abortar el latido. Ver la docstring.
    return 0

sys.exit(main())