feadulta/tools/e2e/verify/classify_gaps.py

#!/usr/bin/env python3
"""
classify_gaps.py — Clasifica los huecos de traducción aflorados por el crawler
(#121/#132) en DESCARGAR (bíblico/litúrgico) vs TRADUCIR (comentarios/artículos),
para coordinar con Minimax.

Entrada: report.csv (de analyze.py). Cruza las páginas marcadas SPANISH_LEAK /
LANG_MISMATCH / CROSS_LANG_REDIRECT con las categorías WP del post, vía wp eval.

Salida:
  gaps_traducir.csv   — posts ES a traducir (comentarios/artículos)
  gaps_descargar.csv  — posts bíblicos/litúrgicos (lecturas, evangelios, salmos)
  gaps_dossier.json   — todo, estructurado
"""
import csv, json, subprocess, re, os, sys
from collections import defaultdict

DIR = os.path.dirname(os.path.abspath(__file__))
REPORT = os.path.join(DIR, "report.csv")

# Categorías que se DESCARGAN (texto oficial), no se traducen con LLM.
DESCARGAR_CATS = {
    1645,  # Lecturas bíblicas
    1648,  # Eucaristía (lecturas/salmos litúrgicos)
    28,    # evangelios-y-comentarios (los 4 evangelistas: descarga)
}
# term_id de las categorías que SÍ se traducen
TRADUCIR_CATS = {1646, 1647, 1650, 6}  # editorial, comentarios evangelio, artículos, carta

FLAGS_INTERES = ("SPANISH_LEAK", "LANG_MISMATCH", "HTML_LANG_MISMATCH", "CROSS_LANG_REDIRECT")


def slug_from_url(url):
    path = re.sub(r"https?://[^/]+", "", url).strip("/")
    parts = [p for p in path.split("/") if p]
    if parts and parts[0] in ("en", "fr", "it", "pt", "es"):
        parts = parts[1:]
    # saltar prefijos no-post
    if parts and parts[0] in ("category", "tag", "author", "page"):
        return None
    return parts[-1] if parts else None


def wp_resolve(slugs):
    """Devuelve {slug: {id, lang, cats:[ids], title, es_id}} vía wp eval."""
    if not slugs:
        return {}
    payload = json.dumps(list(slugs))
    php = r'''
$slugs=json_decode(file_get_contents("php://stdin"),true);
$out=[];
foreach($slugs as $s){
  $p=get_posts(["name"=>$s,"post_type"=>"post","post_status"=>["publish","draft"],"numberposts"=>1,"suppress_filters"=>true]);
  if(!$p){$out[$s]=null;continue;}
  $id=$p[0]->ID;
  $lang=function_exists("pll_get_post_language")?pll_get_post_language($id):"";
  $es=$id;
  if($lang!=="es"&&function_exists("pll_get_post")){$t=pll_get_post($id,"es");if($t)$es=$t;}
  $out[$s]=["id"=>$id,"lang"=>$lang,"cats"=>wp_get_post_categories($id),"title"=>get_the_title($id),"es_id"=>$es,
            "es_cats"=>wp_get_post_categories($es),"es_title"=>get_the_title($es)];
}
echo json_encode($out);
'''
    r = subprocess.run(
        ["docker", "exec", "-i", "wordpress-web", "wp", "eval", php, "--allow-root"],
        input=payload, capture_output=True, text=True)
    try:
        return json.loads(r.stdout.strip().splitlines()[-1])
    except Exception as e:
        print("wp_resolve error:", e, r.stdout[:300], r.stderr[:300], file=sys.stderr)
        return {}


def main():
    if not os.path.exists(REPORT):
        print("Falta report.csv — ejecuta analyze.py primero", file=sys.stderr)
        return 1
    rows = list(csv.DictReader(open(REPORT, encoding="utf-8")))
    flagged = [r for r in rows if any(f in (r.get("flags") or "") for f in FLAGS_INTERES)]
    print(f"Páginas marcadas: {len(flagged)} / {len(rows)}")

    slugs = {}
    for r in flagged:
        s = slug_from_url(r.get("url", ""))
        if s:
            slugs.setdefault(s, r)

    meta = wp_resolve(set(slugs))

    traducir, descargar, otros = [], [], []
    for s, r in slugs.items():
        m = meta.get(s)
        if not m:
            otros.append({"slug": s, "url": r["url"], "flags": r.get("flags"), "motivo": "no resuelve post"})
            continue
        cats = set(m.get("es_cats") or m.get("cats") or [])
        row = {"slug": s, "lang": r.get("lang"), "flags": r.get("flags"),
               "es_id": m["es_id"], "es_title": m.get("es_title"), "cats": sorted(cats), "url": r["url"]}
        if cats & DESCARGAR_CATS:
            descargar.append(row)
        elif cats & TRADUCIR_CATS:
            traducir.append(row)
        else:
            otros.append(row)

    # dedup por es_id para traducir (lo que va a Minimax)
    by_es = {}
    for t in traducir:
        by_es.setdefault(t["es_id"], t)

    def write_csv(path, rows, cols):
        with open(path, "w", newline="", encoding="utf-8") as f:
            w = csv.DictWriter(f, fieldnames=cols, extrasaction="ignore")
            w.writeheader()
            w.writerows(rows)

    write_csv(os.path.join(DIR, "gaps_traducir.csv"), list(by_es.values()),
              ["es_id", "es_title", "lang", "cats", "flags", "url"])
    write_csv(os.path.join(DIR, "gaps_descargar.csv"), descargar,
              ["es_id", "es_title", "lang", "cats", "flags", "url"])
    json.dump({"traducir": list(by_es.values()), "descargar": descargar, "otros": otros},
              open(os.path.join(DIR, "gaps_dossier.json"), "w"), ensure_ascii=False, indent=2)

    print(f"\nTRADUCIR (Minimax): {len(by_es)} posts ES únicos")
    print(f"DESCARGAR (bíblico/litúrgico): {len(descargar)} páginas")
    print(f"OTROS (revisar a mano): {len(otros)}")
    print("\nIDs ES a traducir:", ",".join(str(i) for i in by_es))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())