chemavx-seo-tools/hreflang_watch.py

#!/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())