hreflang: aplicarlo solo cuando el espejo publica, sin que nadie se acuerde

Las parejas se curan a mano, pero solo se pueden declarar cuando los dos
lados están publicados: un hreflang a una URL programada apunta a un 404.
Había cinco esperando, repartidas entre el 4 y el 27 de agosto, y la
alternativa era acordarse de correr `aplicar` cinco veces en las fechas
justas. Eso no es una solución: es una lista que se cumple a medias, y el
hreflang que falta no da error en ninguna parte, así que nadie se entera.

- hreflang_watch.py + timer diario a las 06:40 UTC (40 min tras la cola,
  antes de canibal-watch). Telegram SOLO si algo se ha abierto o ha
  fallado: 26 días de cada 31 no habrá nada, y si esos días avisara, el
  día que importa se leería igual que los otros.
- `aplica()` extraída de cmd_aplicar: el automático escribe por el MISMO
  camino que el manual, no por una copia suya.
- `comprueba` separa PENDIENTE de ROTO. Antes daba diez ✗ de 404 fijos por
  los espejos programados; diez rojos permanentes que no se pueden
  arreglar son un informe que se deja de leer.

Verificado por el camino real, no en seco: se le quitó el bloque al lado ES
de la pareja de Varginha (avería visible en la web pública, 0 hreflang), el
vigilante la detectó, la escribió, la verificó contra las URLs exactas y
avisó por Telegram. Restaurada y comprobada después. Y `verifica()` se ve
fallar con una pareja cruzada a propósito, así que el ✅ no es vacuo.

Nueve tests. Ensayado además arrancando el unit por systemd, que es donde
los CLIs de node mueren con 127 si falta el bin de nvm en el PATH.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

hreflang-watch.service

@@ -0,0 +1,17 @@
+[Unit]
+Description=hreflang — declara la pareja ES↔EN en cuanto el espejo se publica
+Documentation=file:///home/chemavx/seo-tools/hreflang_watch.py
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=oneshot
+User=chemavx
+WorkingDirectory=/home/chemavx/seo-tools
+Environment=HOME=/home/chemavx
+Environment=KUBECONFIG=/home/chemavx/.kube/config
+# El bin de nvm es obligatorio: ghst-en/ghst-es son CLIs de node y sin él el
+# unit muere con 127 sin decir por qué (cicatriz compartida con los demás).
+Environment=PATH=/home/chemavx/.local/bin:/home/chemavx/.nvm/versions/node/v24.14.1/bin:/usr/local/bin:/usr/bin:/bin
+ExecStart=/usr/bin/python3 /home/chemavx/seo-tools/hreflang_watch.py
+TimeoutStartSec=900

hreflang-watch.timer

@@ -0,0 +1,13 @@
+[Unit]
+Description=hreflang, a diario a las 06:40 UTC (40 min después de que publique la cola)
+
+[Timer]
+# La cola publica a las 06:00 UTC. 06:40 da margen a que Ghost haya publicado y
+# la página esté servida, y va antes que canibal-watch (07:00) para que el resto
+# del día ya vea el hreflang puesto.
+OnCalendar=*-*-* 06:40:00 UTC
+Persistent=true
+RandomizedDelaySec=300
+
+[Install]
+WantedBy=timers.target

hreflang_watch.py

@@ -0,0 +1,150 @@
+#!/usr/bin/env python3
+"""hreflang-watch — pone el hreflang en cuanto el espejo se publica.
+
+Por qué existe. Las parejas se curan a mano en hreflang-parejas.md, pero solo se
+pueden DECLARAR cuando los dos lados están publicados: un hreflang a una URL
+programada apunta a un 404. El 2026-07-30 había cinco parejas esperando a que
+publicase el espejo español, repartidas entre el 4 y el 27 de agosto:
+
+    04-ago  oleada belga        11-ago  proyecto Libro Azul
+    24-ago  Colares             25-ago  JAL 1628        27-ago  Teherán
+
+La alternativa a esto era que Jose se acordase de correr `seo_hreflang.py
+aplicar` cinco veces en agosto, cada una el día justo. Eso no es una solución,
+es una lista de tareas que se cumple a medias: el hreflang se queda a medio
+poner y encima nadie se entera, porque el que falta no da error en ningún sitio.
+
+Qué hace. Una vez al día pregunta si alguna pareja ha pasado a publicada por los
+dos lados, y si es así la escribe y la verifica en la web pública. Escribe por
+`seo_hreflang.aplica()`, el MISMO camino que el comando manual — no una copia.
+
+Disciplina de aviso (la de la casa): Telegram solo si ha PASADO algo. Nada de
+recordar a diario las que faltan; el silencio significa «ninguna se ha abierto
+hoy», y ese es el estado normal 26 días de cada 31. Si el silencio fuera el
+aviso, el aviso de verdad no se distinguiría.
+
+    python3 hreflang_watch.py --seco     # dice qué haría, no escribe ni avisa
+    python3 hreflang_watch.py
+"""
+import base64
+import json
+import os
+import subprocess
+import sys
+import urllib.parse
+import urllib.request
+from pathlib import Path
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import seo_hreflang as H  # noqa: E402
+
+TG_SECRET = ("monitoring", "grafana-telegram-infisical")
+LOG_DIR = Path.home() / ".local/state/hreflang-watch"
+
+
+def sh(cmd):
+    return subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=60)
+
+
+def telegram(texto):
+    """Mismo camino que canibal-watch e indexnow-watch: el token sale del secret
+    que gestiona Infisical, nunca de disco."""
+    t = sh(f"kubectl get secret -n {TG_SECRET[0]} {TG_SECRET[1]} "
+           f"-o jsonpath='{{.data.TELEGRAM_BOT_TOKEN}}'")
+    c = sh(f"kubectl get secret -n {TG_SECRET[0]} {TG_SECRET[1]} "
+           f"-o jsonpath='{{.data.TELEGRAM_CHAT_ID}}'")
+    if not t.stdout or not c.stdout:
+        print("✗ no se pudo leer el secret de Telegram", file=sys.stderr)
+        return False
+    token = base64.b64decode(t.stdout).decode()
+    chat = base64.b64decode(c.stdout).decode()
+    data = urllib.parse.urlencode({"chat_id": chat, "text": texto[:3900]}).encode()
+    req = urllib.request.Request(
+        f"https://api.telegram.org/bot{token}/sendMessage", data=data)
+    with urllib.request.urlopen(req, timeout=30) as r:
+        return json.loads(r.read()).get("ok", False)
+
+
+def verifica(parejas_hechas):
+    """Comprueba en la web PÚBLICA solo las parejas que acabamos de escribir.
+
+    Que Ghost devuelva 200 al guardar no significa que la página lo emita: el
+    codeinjection_head podría no llegar a renderizarse. Y se comprueban las URLs
+    EXACTAS, no «que haya hreflang»: un par cruzado también tendría dos.
+    """
+    malos = []
+    for slug_en, slug_es in sorted(parejas_hechas):
+        esperado = {"en": f"{H.SITIOS['en']['base']}/{slug_en}/",
+                    "es": f"{H.SITIOS['es']['base']}/{slug_es}/"}
+        esperado["x-default"] = esperado["en"]
+        for site, slug in (("en", slug_en), ("es", slug_es)):
+            url = f"{H.SITIOS[site]['base']}/{slug}/"
+            try:
+                req = urllib.request.Request(
+                    url, headers={"User-Agent": "hreflang-watch/1.0"})
+                html = urllib.request.urlopen(req, timeout=30).read().decode("utf8", "ignore")
+            except Exception as exc:
+                malos.append(f"{slug}: no se pudo leer ({exc})")
+                continue
+            enc = dict(H.re.findall(
+                r'<link[^>]+hreflang="([^"]+)"[^>]+href="([^"]+)"', html))
+            for k, v in esperado.items():
+                if enc.get(k) != v:
+                    malos.append(f"{slug}: {k} dice {enc.get(k)}, debería ser {v}")
+    return malos
+
+
+def main():
+    seco = "--seco" in sys.argv
+    plan, avisos, parejas = H.cambios()
+
+    espera = H.pendientes(parejas)
+    for (a, b), estado in sorted(espera.items()):
+        print(f"  … pendiente ({estado}): {a[:52]}")
+
+    if not plan:
+        print(f"nada que abrir hoy · {len(parejas) - len(espera)} vivas, "
+              f"{len(espera)} esperando al espejo")
+        return 0
+
+    print(f"\n{len(plan)} posts a escribir:")
+    for site, p, _ in plan:
+        print(f"  [{site}] {p['slug'][:56]}")
+    if seco:
+        print("\n(--seco: no se ha escrito nada)")
+        return 0
+
+    LOG_DIR.mkdir(parents=True, exist_ok=True)
+    hechos, fallos, copia = H.aplica(plan)
+
+    # De los slugs escritos, deducir qué PAREJAS se han abierto, para verificar
+    # solo esas: el resto ya se verificó el día que se pusieron.
+    tocados = {s for _, s in hechos}
+    abiertas = {(a, b) for a, b in parejas if a in tocados or b in tocados}
+    malos = verifica(abiertas) if hechos else []
+
+    lineas = [f"🌐 hreflang: {len(abiertas)} pareja(s) abiertas"]
+    for a, b in sorted(abiertas):
+        lineas.append(f"• {a[:44]} ↔ {b[:44]}")
+    if fallos:
+        lineas.append(f"\n🚨 {len(fallos)} no se pudieron escribir:")
+        lineas += [f"• [{s}] {sl}" for s, sl in fallos]
+    if malos:
+        lineas.append("\n🚨 escrito en Ghost pero MAL en la web pública:")
+        lineas += [f"• {m}" for m in malos]
+    if not fallos and not malos:
+        lineas.append("\n✅ verificado en la web pública, URLs exactas")
+    if espera:
+        lineas.append(f"\n… quedan {len(espera)} esperando a que publique el espejo")
+    lineas.append(f"\ncopia previa: {copia}")
+
+    texto = "\n".join(lineas)
+    print("\n" + texto)
+    if not telegram(texto):
+        print("✗ no se pudo avisar por Telegram", file=sys.stderr)
+        return 1
+    return 1 if (fallos or malos) else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

seo_hreflang.py

@@ -231,13 +231,14 @@ def cmd_plan(a):
     return 0
 
 
-def cmd_aplicar(a):
+def aplica(plan, log=print):
-    plan, avisos, _ = cambios()
+    """Escribe el plan en Ghost. Devuelve (hechos, fallos, ruta_de_la_copia).
-    for w in avisos:
+
-        print(f"  ⚠ {w}")
+    Vive fuera de cmd_aplicar para que el vigilante automático (hreflang_watch)
-    if not plan:
+    escriba por ESTE camino y no por una copia suya: si el manual y el
-        print("nada que hacer: ya está todo puesto")
+    automático no comparten la escritura, el ensayo de uno no dice nada del
-        return 0
+    otro.
+    """
     os.makedirs(BDIR, exist_ok=True)
     ts = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
     bpath = os.path.join(BDIR, f"hreflang-pristine-{ts}.json")
@@ -245,8 +246,8 @@ def cmd_aplicar(a):
         json.dump([{"site": s, "id": p["id"], "slug": p["slug"],
                     "codeinjection_head": p.get("codeinjection_head")}
                    for s, p, _ in plan], f, ensure_ascii=False, indent=1)
-    print(f"copia previa: {bpath}\n")
+    log(f"copia previa: {bpath}\n")
-    fallos = 0
+    hechos, fallos = [], []
     for site, p, nuevo in plan:
         fd, ppath = tempfile.mkstemp(suffix=".json")
         with os.fdopen(fd, "w") as f:
@@ -254,20 +255,56 @@ def cmd_aplicar(a):
         r = sh(f'{SITIOS[site]["cli"]} post update {p["id"]} --from-json "{ppath}"')
         os.unlink(ppath)
         ok = "Slug:" in r.stdout
-        fallos += 0 if ok else 1
+        (hechos if ok else fallos).append((site, p["slug"]))
-        print(f"  {'✓' if ok else '✗'} [{site}] {p['slug'][:56]}")
+        log(f"  {'✓' if ok else '✗'} [{site}] {p['slug'][:56]}")
         if not ok:
-            print(f"      {(r.stdout + r.stderr)[:200]}")
+            log(f"      {(r.stdout + r.stderr)[:200]}")
-    print(f"\n{len(plan) - fallos}/{len(plan)} aplicados")
+    return hechos, fallos, bpath
+
+
+def cmd_aplicar(a):
+    plan, avisos, _ = cambios()
+    for w in avisos:
+        print(f"  ⚠ {w}")
+    if not plan:
+        print("nada que hacer: ya está todo puesto")
+        return 0
+    hechos, fallos, _ = aplica(plan)
+    print(f"\n{len(hechos)}/{len(plan)} aplicados")
     return 1 if fallos else 0
 
 
+def pendientes(parejas):
+    """Parejas en las que algún lado no está publicado todavía.
+
+    Se separan de los fallos porque NO son un fallo: la cola de agosto tiene
+    cinco espejos programados, y comprobarlos daba diez ✗ de 404 en cada
+    ejecución. Un informe con diez rojos permanentes que no se pueden arreglar
+    es un informe que se deja de leer, y entonces el rojo de verdad tampoco se
+    ve. Mismo criterio que `cambios()`: aquí solo se juzga lo que ya vive.
+    """
+    en = trae("en", {a for a, _ in parejas})
+    es = trae("es", {b for _, b in parejas})
+    out = {}
+    for a, b in parejas:
+        pa, pb = en.get(a), es.get(b)
+        if not pa or not pb:
+            out[(a, b)] = "no encontrado en Ghost"
+        elif pa["status"] != "published" or pb["status"] != "published":
+            out[(a, b)] = f"{pa['status']}/{pb['status']}"
+    return out
+
+
 def cmd_comprueba(a):
     """La verdad está en la página pública: que Ghost lo guarde no basta."""
     import urllib.request
     parejas = lee_parejas()
+    espera = pendientes(parejas)
     malos = 0
     for slug_en, slug_es in parejas:
+        if (slug_en, slug_es) in espera:
+            print(f"  … [pendiente {espera[(slug_en, slug_es)]}] {slug_en[:46]}")
+            continue
         for site, slug in (("en", slug_en), ("es", slug_es)):
             url = f"{SITIOS[site]['base']}/{slug}/"
             try:
@@ -290,7 +327,9 @@ def cmd_comprueba(a):
             print(f"  {'✓' if not mal else '✗'} [{site}] {slug[:52]}")
             for k, (tiene, debe) in mal.items():
                 print(f"        {k}: dice {tiene} — debería ser {debe}")
-    print(f"\n{'✓ todas las páginas declaran las dos versiones' if not malos else f'✗ {malos} con problema'}")
+    vivas = len(parejas) - len(espera)
+    print(f"\n{f'✓ las {vivas} parejas vivas declaran las dos versiones' if not malos else f'✗ {malos} con problema'}"
+          + (f" · {len(espera)} pendientes de que publique el espejo" if espera else ""))
     return 1 if malos else 0
 
 

test_hreflang_watch.py

@@ -0,0 +1,175 @@
+#!/usr/bin/env python3
+"""Tests de hreflang_watch.py y de la separación pendiente/roto.
+
+Lo que se protege aquí no es «que no dé error»: es la disciplina de aviso. Un
+vigilante que canta a diario cinco rojos que nadie puede arreglar se convierte
+en ruido, y entonces el rojo de verdad tampoco se ve. Eso ya pasó con los
+og_*/twitter_* de las páginas.
+
+    python3 -m pytest test_hreflang_watch.py -q
+    python3 test_hreflang_watch.py            # sin pytest
+"""
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import seo_hreflang as H
+import hreflang_watch as W
+
+
+def _con_posts(en, es):
+    """Sustituye la lectura de Ghost por corpus de mentira, por sitio."""
+    orig = H.trae
+    H.trae = lambda site, slugs: (en if site == "en" else es)
+    return orig
+
+
+P = [("caso-en", "caso-es")]
+
+
+def _post(slug, status, head=""):
+    return {"id": "i-" + slug, "slug": slug, "status": status,
+            "codeinjection_head": head}
+
+
+# ─────────────── pendiente NO es roto (la cola de agosto) ───────────────
+
+def test_un_espejo_programado_sale_como_pendiente_no_como_fallo():
+    orig = _con_posts({"caso-en": _post("caso-en", "published")},
+                      {"caso-es": _post("caso-es", "scheduled")})
+    try:
+        esp = H.pendientes(P)
+    finally:
+        H.trae = orig
+    assert esp == {("caso-en", "caso-es"): "published/scheduled"}, esp
+
+
+def test_los_dos_publicados_no_estan_pendientes():
+    orig = _con_posts({"caso-en": _post("caso-en", "published")},
+                      {"caso-es": _post("caso-es", "published")})
+    try:
+        assert H.pendientes(P) == {}
+    finally:
+        H.trae = orig
+
+
+def test_un_slug_que_no_existe_en_ghost_tambien_es_pendiente():
+    """Si se renombra un slug en hreflang-parejas.md y no en Ghost, no hay que
+    tratarlo como una página rota: no hay nada que escribir todavía."""
+    orig = _con_posts({}, {"caso-es": _post("caso-es", "published")})
+    try:
+        assert H.pendientes(P) == {("caso-en", "caso-es"): "no encontrado en Ghost"}
+    finally:
+        H.trae = orig
+
+
+# ─────────────── que un programado NUNCA se escriba ───────────────
+
+def test_no_se_planifica_nada_mientras_el_espejo_este_programado():
+    """Es la salvaguarda de fondo: un hreflang a una URL futura es un 404, y un
+    404 declarado como versión alternativa es peor que no declarar nada."""
+    orig_t = _con_posts({"caso-en": _post("caso-en", "published")},
+                        {"caso-es": _post("caso-es", "scheduled")})
+    orig_p = H.lee_parejas
+    H.lee_parejas = lambda: P
+    try:
+        plan, avisos, _ = H.cambios()
+    finally:
+        H.trae, H.lee_parejas = orig_t, orig_p
+    assert plan == [], plan
+    assert any("404" in w for w in avisos), avisos
+
+
+def test_en_cuanto_se_publica_el_espejo_aparece_el_plan():
+    """La otra mitad: el vigilante no sirve si es tan prudente que nunca actúa."""
+    orig_t = _con_posts({"caso-en": _post("caso-en", "published")},
+                        {"caso-es": _post("caso-es", "published")})
+    orig_p = H.lee_parejas
+    H.lee_parejas = lambda: P
+    try:
+        plan, _, _ = H.cambios()
+    finally:
+        H.trae, H.lee_parejas = orig_t, orig_p
+    assert len(plan) == 2, plan                       # escribe en los DOS lados
+    assert {s for s, _, _ in plan} == {"en", "es"}, plan
+    for _, _, nuevo in plan:
+        assert H.INICIO in nuevo and H.FIN in nuevo
+        assert 'hreflang="x-default"' in nuevo        # sin él Google descarta el grupo
+
+
+# ─────────────── silencio cuando no hay nada que contar ───────────────
+
+def test_sin_nada_que_abrir_no_avisa_por_telegram():
+    """26 días de cada 31 no habrá nada. Si esos días mandara mensaje, el día que
+    importa se leería igual que los otros."""
+    orig_t = _con_posts({"caso-en": _post("caso-en", "published")},
+                        {"caso-es": _post("caso-es", "scheduled")})
+    orig_p, orig_tg = H.lee_parejas, W.telegram
+    H.lee_parejas = lambda: P
+    avisado = []
+    W.telegram = lambda t: avisado.append(t) or True
+    try:
+        rc = W.main()
+    finally:
+        H.trae, H.lee_parejas, W.telegram = orig_t, orig_p, orig_tg
+    assert rc == 0, rc
+    assert avisado == [], avisado
+
+
+def test_el_aviso_dice_que_pareja_y_que_queda_pendiente():
+    orig_t = _con_posts({"caso-en": _post("caso-en", "published")},
+                        {"caso-es": _post("caso-es", "published")})
+    orig_p, orig_tg, orig_a, orig_v = (H.lee_parejas, W.telegram,
+                                       H.aplica, W.verifica)
+    H.lee_parejas = lambda: P
+    avisado = []
+    W.telegram = lambda t: avisado.append(t) or True
+    H.aplica = lambda plan, log=print: ([(s, p["slug"]) for s, p, _ in plan],
+                                        [], "/tmp/copia.json")
+    W.verifica = lambda abiertas: []
+    try:
+        rc = W.main()
+    finally:
+        (H.trae, H.lee_parejas, W.telegram, H.aplica, W.verifica) = (
+            orig_t, orig_p, orig_tg, orig_a, orig_v)
+    assert rc == 0, rc
+    assert len(avisado) == 1, avisado
+    assert "caso-en" in avisado[0] and "caso-es" in avisado[0]
+    assert "verificado en la web pública" in avisado[0]
+
+
+def test_si_ghost_guarda_pero_la_web_no_lo_emite_grita_y_sale_con_error():
+    """El caso peligroso: Ghost dice OK y la página no lo tiene. Sin esto, el
+    vigilante daría por hecho algo que no está."""
+    orig_t = _con_posts({"caso-en": _post("caso-en", "published")},
+                        {"caso-es": _post("caso-es", "published")})
+    orig_p, orig_tg, orig_a, orig_v = (H.lee_parejas, W.telegram,
+                                       H.aplica, W.verifica)
+    H.lee_parejas = lambda: P
+    avisado = []
+    W.telegram = lambda t: avisado.append(t) or True
+    H.aplica = lambda plan, log=print: ([(s, p["slug"]) for s, p, _ in plan],
+                                        [], "/tmp/copia.json")
+    W.verifica = lambda abiertas: ["caso-es: en dice None, debería ser X"]
+    try:
+        rc = W.main()
+    finally:
+        (H.trae, H.lee_parejas, W.telegram, H.aplica, W.verifica) = (
+            orig_t, orig_p, orig_tg, orig_a, orig_v)
+    assert rc == 1, "un fallo silencioso deja exit 0 y systemd lo da por bueno"
+    assert "🚨" in avisado[0] and "MAL en la web pública" in avisado[0], avisado
+
+
+if __name__ == "__main__":
+    fallos = 0
+    for nombre, fn in sorted(globals().items()):
+        if not nombre.startswith("test_"):
+            continue
+        try:
+            fn()
+            print(f"  ✓ {nombre}")
+        except AssertionError as exc:
+            fallos += 1
+            print(f"  ✗ {nombre}: {exc}")
+    print(f"\n{'✓ todo pasa' if not fallos else f'✗ {fallos} fallo(s)'}")
+    sys.exit(1 if fallos else 0)