chemavx-seo-tools/seo_link.py

#!/usr/bin/env python3
"""
Tool F — inserta enlaces internos en el CUERPO de un post (EN o ES, Ghost).

Es la pieza que faltaba: el resto de herramientas leen `html` (que Ghost
renderiza sea cual sea el formato de origen) y por eso les da igual el formato.
Escribir en el cuerpo, no: hay que tocar la fuente, y la fuente viene en tres
sabores distintos según cuándo y cómo se creó el post.

    1. lexical con UN bloque `html`   — los 29 del ES y casi todos los del EN
    2. lexical con nodos NATIVOS      — los dos forenses del EN (roswell-341,
                                        yellow-sea); el texto vive en nodos
                                        `extended-text` dentro de `paragraph`
    3. mobiledoc con tarjetas `html`  — lo que genera ResearchOwl HOY en los
                                        borradores nuevos (descubierto el
                                        2026-07-21 con el Canarias de prueba)

Comprobaciones ANTES de escribir (todas son cicatrices de errores reales):
  - el slug destino existe en el corpus de ESE sitio (no del otro)
  - el destino está publicado; si está programado avisa de enlace PREMATURO
    (daría 404 en la ventana intermedia — caso Wilson-Davis → MJ-12)
  - la URL se construye con el canónico del sitio: EN es www, ES es el APEX
  - el anchor aparece en la prosa y NO dentro de un <a> ya existente
    (casi metemos un <a> anidado en Levelland con «Project Blue Book»)
  - tras la edición: anclas balanceadas y ninguna anidada

Uso:
    python3 seo_link.py <slug> --site es --link "incidente de Roswell=roswell-1947-..."
    python3 seo_link.py <slug> --site es --link "A=slug-a" --link "B=slug-b" --apply

Sin --apply no escribe: imprime lo que haría con el contexto de cada inserción.
"""
import argparse
import copy
import datetime
import json
import os
import re
import subprocess
import sys
import tempfile

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

SITES = {
    "en": {"cli": "ghst-en", "base": "https://www.theexclusionzone.com"},
    "es": {"cli": "ghst-es", "base": "https://zonadeexclusion.com"},
}
BDIR = os.path.expanduser("~/link-batch-backup")

# Esquema exacto de los nodos lexical nativos, copiado de un enlace real del
# corpus. No inventar campos: Ghost valida y un nodo mal formado se pierde.
TEXT_NODE = {"detail": 0, "format": 0, "mode": "normal", "style": "",
             "text": "", "type": "extended-text", "version": 1}
LINK_NODE = {"children": [], "direction": "ltr", "format": "", "indent": 0,
             "type": "link", "version": 1, "rel": None, "target": None,
             "title": None, "url": ""}


def sh(cmd, timeout=240):
    return subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=timeout)


def fetch_corpus(cli):
    """Corpus completo (published + scheduled). A fichero: un PIPE se trunca.

    `--limit all` + comprobación del total (ver seo_watch.fetch_posts). Aquí un
    corpus truncado es peor todavía que en el vigilante: haría fallar el
    «¿existe el destino?» y rechazaría un enlace perfectamente válido, o peor,
    daría por inexistente un post que sí está."""
    fd, path = tempfile.mkstemp(suffix=".json")
    os.close(fd)
    try:
        sh(f"{cli} post list --limit all --formats lexical,mobiledoc,html --json > {path}")
        with open(path) as f:
            d = json.load(f)
        posts = d.get("posts", d) if isinstance(d, dict) else d
        total = ((d.get("meta") or {}).get("pagination") or {}).get("total") \
            if isinstance(d, dict) else None
        if total is not None and len(posts) != total:
            sys.exit(f"✗ corpus truncado: {len(posts)} de {total} posts — abortado "
                     f"sin escribir (el chequeo de destinos sería mentira)")
        return posts
    finally:
        os.unlink(path)


# ---------- detección de formato ----------

def body_format(post):
    lex = post.get("lexical")
    if lex:
        kids = json.loads(lex)["root"]["children"]
        if len(kids) == 1 and kids[0].get("type") == "html":
            return "lexical-html"
        return "lexical-nativo"
    if post.get("mobiledoc"):
        return "mobiledoc"
    return None


# ---------- inserción en HTML (vale para lexical-html y mobiledoc) ----------

def link_in_html(html, anchor, url):
    """Primer <p> con el anchor FUERA de cualquier <a>. -> (html, ok, contexto)"""
    hecho = [False]
    ctx = [""]

    def en_parrafo(m):
        if hecho[0]:
            return m.group(0)
        trozos = re.split(r"(<a\b.*?</a>)", m.group(2), flags=re.S)
        for i, tr in enumerate(trozos):
            if tr.startswith("<a"):
                continue
            mm = re.search(r"(?<![\w-])" + re.escape(anchor) + r"(?![\w-])", tr)
            if not mm:
                continue
            trozos[i] = tr[:mm.start()] + f'<a href="{url}">{anchor}</a>' + tr[mm.end():]
            a, b = max(0, mm.start() - 55), mm.end() + 55
            ctx[0] = re.sub(r"\s+", " ", re.sub(r"<[^>]+>", "", tr[a:b]))
            hecho[0] = True
            break
        return m.group(1) + "".join(trozos) + m.group(3)

    out = re.sub(r"(<p>)(.*?)(</p>)", en_parrafo, html, flags=re.S)
    return out, hecho[0], ctx[0]


# ---------- inserción en lexical nativo ----------

def link_in_lexical_nodes(root_children, anchor, url):
    """Parte el nodo de texto en (antes | link | después). Solo mira los hijos
    DIRECTOS del párrafo, así que un texto que ya viva dentro de un `link`
    nunca se toca — que es justo lo que evita el anidamiento."""
    for parrafo in root_children:
        if parrafo.get("type") != "paragraph":
            continue
        hijos = parrafo.get("children") or []
        for i, nodo in enumerate(hijos):
            if nodo.get("type") != "extended-text":
                continue
            txt = nodo.get("text") or ""
            mm = re.search(r"(?<![\w-])" + re.escape(anchor) + r"(?![\w-])", txt)
            if not mm:
                continue
            antes, despues = txt[:mm.start()], txt[mm.end():]
            enlace = copy.deepcopy(LINK_NODE)
            enlace["url"] = url
            hijo = copy.deepcopy(TEXT_NODE)
            hijo["text"] = anchor
            hijo["format"] = nodo.get("format", 0)
            enlace["children"] = [hijo]

            nuevos = []
            if antes:
                n = copy.deepcopy(nodo)
                n["text"] = antes
                nuevos.append(n)
            nuevos.append(enlace)
            if despues:
                n = copy.deepcopy(nodo)
                n["text"] = despues
                nuevos.append(n)
            parrafo["children"] = hijos[:i] + nuevos + hijos[i + 1:]
            a, b = max(0, mm.start() - 55), mm.end() + 55
            return True, re.sub(r"\s+", " ", txt[a:b])
    return False, ""


# ---------- salvaguardas ----------

def revisar_html(html, slug):
    if re.search(r"<a [^>]*>(?:(?!</a>).)*<a ", html, re.S):
        sys.exit(f"✗ {slug}: ancla anidada — abortado sin escribir")
    if html.count("<a ") != html.count("</a>"):
        sys.exit(f"✗ {slug}: <a> descuadrados — abortado sin escribir")


def main():
    ap = argparse.ArgumentParser(description="Enlaces internos en el cuerpo (Tool F)")
    ap.add_argument("slug")
    ap.add_argument("--site", choices=("en", "es"), required=True)
    ap.add_argument("--link", action="append", required=True,
                    metavar="ANCHOR=slug-destino",
                    help="texto del anchor y slug al que apunta; repetible")
    ap.add_argument("--apply", action="store_true")
    a = ap.parse_args()

    cfg = SITES[a.site]
    corpus = fetch_corpus(cfg["cli"])
    by = {p["slug"]: p for p in corpus}
    if a.slug not in by:
        sys.exit(f"✗ «{a.slug}» no está en el corpus de [{a.site.upper()}] "
                 f"— ¿es un slug del otro sitio?")
    post = by[a.slug]
    fmt = body_format(post)
    if not fmt:
        sys.exit(f"✗ {a.slug}: sin cuerpo legible (ni lexical ni mobiledoc)")
    print(f"── {a.slug[:56]}  [{post.get('status')}]  formato: {fmt}")

    pares = []
    for spec in a.link:
        if "=" not in spec:
            sys.exit(f"✗ --link mal formado: {spec!r} (falta «=»)")
        anchor, destino = spec.split("=", 1)
        anchor, destino = anchor.strip(), destino.strip().strip("/")
        if destino not in by:
            sys.exit(f"✗ destino «{destino}» no existe en [{a.site.upper()}]")
        if destino == a.slug:
            sys.exit(f"✗ «{destino}» se enlazaría a sí mismo")
        tgt = by[destino]
        if tgt.get("status") != "published":
            print(f"   ⚠ PREMATURO: «{destino}» está en estado "
                  f"{tgt.get('status')} → 404 hasta que se publique")
        pares.append((anchor, f'{cfg["base"]}/{destino}/'))

    # --- inserción según formato ---
    if fmt == "lexical-nativo":
        lex = json.loads(post["lexical"])
        kids = lex["root"]["children"]
        for anchor, url in pares:
            ok, ctx = link_in_lexical_nodes(kids, anchor, url)
            print(f"   {'+' if ok else '⚠'} «{anchor}» → {url.split('/')[-2][:34]}"
                  + (f"\n     …{ctx}…" if ok else "  (no aparece en la prosa)"))
        campo, valor = "lexical", json.dumps(lex, ensure_ascii=False)
    else:
        if fmt == "lexical-html":
            doc = json.loads(post["lexical"])
            trozo = doc["root"]["children"][0]
            get_html, set_html = (lambda: trozo["html"]), (lambda h: trozo.__setitem__("html", h))
        else:
            doc = json.loads(post["mobiledoc"])
            cards = [c for c in doc.get("cards", []) if c and c[0] == "html"]
            if not cards:
                sys.exit(f"✗ {a.slug}: mobiledoc sin tarjetas html")
            tarjeta = cards[0][1]
            get_html, set_html = (lambda: tarjeta["html"]), (lambda h: tarjeta.__setitem__("html", h))
        html = get_html()
        for anchor, url in pares:
            html, ok, ctx = link_in_html(html, anchor, url)
            print(f"   {'+' if ok else '⚠'} «{anchor}» → {url.split('/')[-2][:34]}"
                  + (f"\n     …{ctx}…" if ok else "  (no aparece en la prosa)"))
        revisar_html(html, a.slug)
        set_html(html)
        campo = "lexical" if fmt == "lexical-html" else "mobiledoc"
        valor = json.dumps(doc, ensure_ascii=False)

    if not a.apply:
        print("\n(dry-run: no se ha escrito nada — repite con --apply)")
        return 0

    os.makedirs(BDIR, exist_ok=True)
    ts = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
    bpath = os.path.join(BDIR, f"{a.site}-{a.slug[:40]}-seolink-pristine-{ts}.json")
    with open(bpath, "w") as f:
        json.dump({"id": post["id"], "slug": a.slug, campo: post.get(campo)},
                  f, ensure_ascii=False)
    fd, ppath = tempfile.mkstemp(suffix=".json")
    with os.fdopen(fd, "w") as f:
        json.dump({campo: valor}, f, ensure_ascii=False)
    r = sh(f'{cfg["cli"]} post update {post["id"]} --from-json "{ppath}"')
    os.unlink(ppath)
    if "Slug:" not in r.stdout:
        print(r.stdout, r.stderr)
        sys.exit(f"✗ {a.slug}: {cfg['cli']} no confirmó la actualización")
    print(f"✓ aplicado (backup {os.path.basename(bpath)})")
    return 0


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