chemavx-seo-tools/canibal_watch.py

#!/usr/bin/env python3
"""
canibal-watch — avisa cuando dos artículos del MISMO blog empiezan a competir.

Corre a diario, calcula los pares por encima del umbral con seo_semantic y
manda Telegram SOLO si hay algo NUEVO que contar. Un vigilante que avisa todos
los días se convierte en ruido y se deja de leer.

Qué cuenta como novedad:
  · un par que antes no pasaba el umbral y ahora sí (típico: se publica un
    artículo que solapa con otro ya publicado)
  · un par que ya estaba pero se ha quedado SIN enlace en ninguna dirección
    (una edición se llevó por delante el enlace que ordenaba la jerarquía)

Lo que NO dispara: que un par siga igual que ayer. Se guarda en el estado y se
calla.

No escribe en el blog ni en el corpus: solo lee y avisa. Arreglar el solape es
decisión editorial de Jose.
"""
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_semantic as S  # noqa: E402

ESTADO_DIR = Path.home() / ".local/state/canibal-watch"
ESTADO = ESTADO_DIR / "state.json"
TG_SECRET = ("monitoring", "grafana-telegram-infisical")
UMBRAL = float(os.environ.get("CANIBAL_UMBRAL", "0.78"))


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


def telegram(texto):
    """Mismo camino que indexnow-watch y remate-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 clave(par):
    return f"{par['a']['slug']}|{par['b']['slug']}"


def main():
    seco = "--seco" in sys.argv          # calcula y enseña, pero no manda ni guarda
    ESTADO_DIR.mkdir(parents=True, exist_ok=True)
    previo = {}
    if ESTADO.exists():
        previo = json.loads(ESTADO.read_text())

    ahora, nuevos, perdieron_enlace = {}, [], []
    for sitio in ("es", "en"):
        for par in S.canibalizacion(sitio, UMBRAL):
            k = f"{sitio}:{clave(par)}"
            ahora[k] = {"similitud": par["similitud"], "enlace": par["enlace"]}
            antes = previo.get(k)
            # Solo es ACCIONABLE si además no se enlazan: un par muy parecido
            # pero cosido con un enlace ya le dice a Google cuál manda. Los
            # enlazados se siguen guardando en el estado (hacen falta para
            # detectar que MÁS TARDE pierdan el enlace), pero no se avisan: un
            # vigilante que saca 19 líneas para 6 cosas que hacer se deja de
            # leer. Para verlos todos está `seo_semantic.py canibalizacion`.
            if antes is None:
                if par["enlace"] == "ninguno":
                    nuevos.append((sitio, par))
            elif antes["enlace"] != "ninguno" and par["enlace"] == "ninguno":
                perdieron_enlace.append((sitio, par))

    print(f"pares por encima de {UMBRAL}: {len(ahora)}  "
          f"nuevos: {len(nuevos)}  han perdido el enlace: {len(perdieron_enlace)}")

    if nuevos or perdieron_enlace:
        lineas = ["🔎 Canibalización en los blogs"]
        for titulo, grupo in (("NUEVOS pares que compiten", nuevos),
                              ("Se han quedado SIN enlace", perdieron_enlace)):
            if not grupo:
                continue
            lineas.append(f"\n{titulo}:")
            for sitio, p in grupo[:6]:
                lineas.append(
                    f"\n[{sitio.upper()}] {p['similitud']:.3f} — sin enlace"
                    f"\n  A: {p['a']['titulo'][:80]}"
                    f"\n  B: {p['b']['titulo'][:80]}")
            if len(grupo) > 6:
                lineas.append(f"\n…y {len(grupo)-6} más "
                              f"(verlos: seo_semantic.py canibalizacion --site es|en)")
        lineas.append("\n\nDecidir a mano: coser con enlaces internos hacia el "
                      "que deba mandar, o fusionar. Yo no toco el blog.")
        texto = "\n".join(lineas)
        if seco:
            print("\n--- lo que se enviaría ---\n" + texto)
        else:
            print("enviado a Telegram" if telegram(texto) else "✗ fallo al enviar")
    else:
        print("sin novedades: nadie avisa a nadie")

    if not seco:
        ESTADO.write_text(json.dumps(ahora, indent=1, ensure_ascii=False))


if __name__ == "__main__":
    main()