diff --git a/canibal-watch.service b/canibal-watch.service new file mode 100644 index 0000000..90883f9 --- /dev/null +++ b/canibal-watch.service @@ -0,0 +1,15 @@ +[Unit] +Description=Canibalización — pares de artículos que compiten por la misma búsqueda +Documentation=file:///home/chemavx/seo-tools/canibal_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 +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/canibal_watch.py +TimeoutStartSec=900 diff --git a/canibal-watch.timer b/canibal-watch.timer new file mode 100644 index 0000000..3db8927 --- /dev/null +++ b/canibal-watch.timer @@ -0,0 +1,10 @@ +[Unit] +Description=Canibalización, a diario a las 07:00 UTC (una hora después de que publique la cola) + +[Timer] +OnCalendar=*-*-* 07:00:00 UTC +Persistent=true +RandomizedDelaySec=600 + +[Install] +WantedBy=timers.target diff --git a/canibal_watch.py b/canibal_watch.py new file mode 100644 index 0000000..8f3d57d --- /dev/null +++ b/canibal_watch.py @@ -0,0 +1,124 @@ +#!/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() diff --git a/remate_prompt.md b/remate_prompt.md index 02a43c5..d35ca4f 100644 --- a/remate_prompt.md +++ b/remate_prompt.md @@ -38,11 +38,22 @@ CHECKLIST — en este orden (el SEO se deriva del texto FINAL, nunca antes): entero serializado, cámbialo. 3) ENLACES INTERNOS — mínimo 2, SOLO a posts publicados de ESTE sitio (nunca a - pages, nunca a posts programados = enlace prematuro 404): + pages, nunca a posts programados = enlace prematuro 404). + + Para ELEGIR los destinos, pregunta primero a la capa semántica en vez de ir + a ojo (embebe el borrador al vuelo y lo compara con todo el corpus): + python3 ~/seo-tools/seo_semantic.py afines $slug --site $site -n 6 + Devuelve los publicados más afines por similitud y marca «(YA ENLAZADO)» los + que el borrador ya enlaza — ésos NO los repitas. Es una sugerencia, no una + orden: descarta los que no encajen de verdad en la prosa, y si ninguno pasa + de ~0,55 di en el informe que el corpus no da para enlaces buenos en vez de + forzar uno malo. Un enlace forzado es peor que ninguno. + + Para INSERTARLOS: python3 ~/seo-tools/seo_link.py --site $site \ --link "anchor=slug-destino" [--link ...] --apply Primero SIN --apply para revisar las inserciones. Elige anchors que ya - existan en la prosa y destinos temáticamente afines de verdad. + existan en la prosa. 4) METAS + ALT — con seo_finish.py: - `python3 ~/seo-tools/seo_finish.py prep --site $site` descarga la diff --git a/seo_semantic.py b/seo_semantic.py index f87026c..85efc5a 100644 --- a/seo_semantic.py +++ b/seo_semantic.py @@ -190,11 +190,49 @@ def matriz(regs): return slugs, M @ M.T +def post_suelto(sitio, slug): + """Un post por slug, INCLUIDOS los borradores. + + Hace falta porque `ghst post list` sin filtro devuelve solo publicados y + programados: el corpus NO tiene borradores. Y el caso de uso principal — + sugerirle enlaces a remate-watch mientras prepara un borrador— es + exactamente ese. Se embebe al vuelo contra el caché, sin guardarlo. + """ + import tempfile + cli = SITIOS[sitio]["cli"] + fd, path = tempfile.mkstemp(suffix=".json", prefix="seo_semantic_") + os.close(fd) + try: + subprocess.run(f'{cli} post list --filter "slug:{slug}" --limit all ' + f'--formats html --json > {path}', + shell=True, capture_output=True, text=True, timeout=240) + with open(path) as f: + d = json.load(f) + posts = d.get("posts", d) if isinstance(d, dict) else d + return posts[0] if posts else None + except (json.JSONDecodeError, IndexError): + return None + finally: + os.unlink(path) + + def afines(sitio, slug, n, refrescar=False): regs = carga(sitio, refrescar) + externo = None if slug not in regs: - sys.exit(f"✗ '{slug}' no está en el corpus de {sitio.upper()} " - f"({len(regs)} artículos). ¿Sitio equivocado?") + p = post_suelto(sitio, slug) + if p is None: + sys.exit(f"✗ '{slug}' no existe en {sitio.upper()} " + f"(ni publicado, ni programado, ni borrador). " + f"¿Sitio equivocado?") + externo = {"vec": embed([texto_de(p)], endpoint())[0], + "enlaza_a": enlaces_de(p, SITIOS[sitio]["base"]), + "estado": p.get("status")} + print(f" («{slug}» está en {externo['estado']}, fuera del corpus: " + f"embebido al vuelo)", file=sys.stderr) + regs = dict(regs) + regs[slug] = {**externo, "slug": slug, "titulo": p.get("title"), + "url": f"{SITIOS[sitio]['base']}/{slug}/"} slugs, S = matriz(regs) i = slugs.index(slug) fila = [(S[i][j], slugs[j]) for j in range(len(slugs))