chemavx-seo-tools/seo_watch.py

#!/usr/bin/env python3
"""
Tool D — vigilante SEO periódico de los dos blogs (EN + ES, Ghost).

READ-ONLY sobre Ghost. Corre los chequeos que detectan REGRESIONES nuevas a
medida que se publica (2 posts/semana), no el inventario completo de siempre:

  1. Enlaces internos rotos          (destino 404 / slug inexistente)
  2. Enlaces internos no canónicos   (apex, http, o sin barra final → salto 301)
  3. Enlaces prematuros              (a un post que se publica DESPUÉS que el que
                                      lo enlaza → 404 en la ventana intermedia)
  4. Enlaces envejecidos             (el anchor nombra a otro artículo que se
                                      publicó DESPUÉS: el corpus crece y el
                                      mejor destino cambia)
  5. Sitemap                         (toda URL listada debe dar 200 directo)
  6. Violaciones accionables         (delta vs la última ejecución, vía seo_rules)

Solo avisa por Telegram cuando hay hallazgos o cuando algo CAMBIA respecto a la
ejecución anterior — un informe semanal de "todo bien" es ruido que se ignora.

Uso:
    python3 seo_watch.py               # chequea y notifica si procede
    python3 seo_watch.py --dry-run     # imprime, no notifica ni guarda estado
    python3 seo_watch.py --force       # notifica aunque no haya cambios

Diseño (lecciones del 2026-07-18, ver memoria project-en-seo-recovery):
  - Lo determinista va primero: fue lo que encontró los problemas reales.
  - Gemini NO sugiere enlaces temáticos (0/6 acertó). Si se le llama, es para
    EXTRAER entidades del texto que casen con slugs del corpus, pasándole
    siempre los waivers de seo_exceptions para que no reabra decisiones cerradas.
  - Nada de esto modifica Ghost. Las correcciones las decide un humano.
"""
import argparse
import json
import os
import re
import subprocess
import sys
import tempfile
import urllib.parse
import urllib.request
from concurrent.futures import ThreadPoolExecutor

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import seo_rules as R
import seo_exceptions as X

# Los dos sitios. Ojo al canónico: EN es www, ES es el APEX — al revés.
SITES = {
    "en": {"cli": "ghst-en", "host": "theexclusionzone.com",
           "url": "https://www.theexclusionzone.com/"},
    "es": {"cli": "ghst-es", "host": "zonadeexclusion.com",
           "url": "https://zonadeexclusion.com/"},
}
HOST = SITES["en"]["url"]          # el sitio activo; lo fija use_site()
CLI = SITES["en"]["cli"]
STATE = os.path.expanduser("~/.local/state/seo-watch/last.json")
TG_SECRET = ("monitoring", "grafana-telegram-infisical")
TIMEOUT = 20


def use_site(site):
    """Fija el sitio activo para todos los checks.

    seo_rules.SITE_HOST está clavado a EN y NO se parametriza ahí a propósito:
    ese fichero se vendoriza byte a byte en ResearchOwl y la CI lo verifica.
    Se reasigna en runtime, que deja intacto el contrato compartido.
    """
    global HOST, CLI
    cfg = SITES[site]
    HOST, CLI = cfg["url"], cfg["cli"]
    R.SITE_HOST = cfg["host"]
    return cfg


# ---------- utilidades ----------

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


def fetch_posts():
    """Todos los posts (published + scheduled) con html. Salida a fichero: la
    respuesta pesa cientos de KB y un PIPE se trunca a 64 KB."""
    fd, path = tempfile.mkstemp(suffix=".json", prefix="seo_watch_")
    os.close(fd)
    try:
        r = sh(f"{CLI} post list --limit 100 --formats html --json > {path}", timeout=240)
        with open(path) as f:
            data = json.load(f)
        return data.get("posts", data) if isinstance(data, dict) else data
    finally:
        os.unlink(path)


def http_status(url):
    """(código, destino_redirección). No sigue la redirección."""
    req = urllib.request.Request(url, method="HEAD", headers={"User-Agent": "seo-watch/1.0"})
    try:
        class NoRedirect(urllib.request.HTTPRedirectHandler):
            def redirect_request(self, req, fp, code, msg, headers, newurl):
                raise urllib.error.HTTPError(req.full_url, code, msg, headers, None)
        op = urllib.request.build_opener(NoRedirect)
        with op.open(req, timeout=TIMEOUT) as r:
            return r.status, None
    except urllib.error.HTTPError as e:
        return e.code, e.headers.get("Location")
    except Exception as e:
        return 0, str(e)[:60]


def internal_urls(html):
    return {u.split("#")[0] for u in re.findall(r'href="(https?://[^"]*theexclusionzone\.com[^"]*)"', html or "")}


def slug_of(url):
    p = urllib.parse.urlparse(url).path.strip("/")
    return p.split("/")[0] if p else ""


# ---------- chequeos ----------

def check_links(posts):
    """Rotos, no canónicos y prematuros, en una pasada."""
    live = {p["slug"] for p in posts if p.get("status") == "published"}
    known = {p["slug"]: p for p in posts}
    roto, nocanon, prematuro = [], [], []

    for p in posts:
        src, src_pub = p["slug"], p.get("published_at") or ""
        for u in internal_urls(p.get("html")):
            # no canónico: debe ser https + www + barra final
            if not u.startswith(HOST) or not u.endswith("/"):
                nocanon.append((src, u))
                continue
            s = slug_of(u)
            if not s or s in ("tag", "author", "about", "complete-guide-uap-cases"):
                continue
            tgt = known.get(s)
            if tgt is None:
                roto.append((src, u))
            elif s not in live:
                # el destino existe pero aún no está publicado
                tgt_pub = tgt.get("published_at") or ""
                if tgt_pub > src_pub:
                    prematuro.append((src, s, str(src_pub)[:10], str(tgt_pub)[:10]))
    return roto, nocanon, prematuro


# Tokens que no distinguen a un post de otro: aparecen en medio corpus.
_GENERIC = {
    "the", "a", "of", "and", "in", "on", "to", "that", "its", "an", "at",
    "ufo", "ufos", "uap", "incident", "files", "file", "declassified", "secret",
    "analysis", "forensic", "video", "case", "cases", "mystery", "history",
    "new", "us", "american", "america", "americas", "government", "official",
    "report", "reports", "still", "what", "how", "why", "who", "cant", "explain",
}


def _distinctive(slug):
    """Tokens del slug que de verdad identifican a ESE post."""
    return {t for t in slug.split("-")
            if t not in _GENERIC and not t.isdigit() and len(t) > 2}


def check_link_aging(posts):
    """Enlaces cuyo anchor nombra inequívocamente a OTRO artículo del corpus.

    El corpus crece 2 posts/semana, así que un enlace apunta al mejor destino
    que existía el día en que se escribió — y envejece. Caso real (2026-07-18):
    levelland enlazaba el anchor 'Project Blue Book' al artículo de AARO, porque
    cuando se generó (30-jun) el artículo de Blue Book aún no existía (1-jul).

    Heurística deliberadamente CONSERVADORA: solo avisa si el anchor contiene
    TODOS los tokens distintivos de otro post. Prefiere callarse a dar la lata:
    un vigilante desatendido que grita en falso se acaba ignorando.
    """
    live = {p["slug"] for p in posts if p.get("status") == "published"}
    tokmap = {s: _distinctive(s) for s in live}
    out = []
    for p in posts:
        html = p.get("html") or ""
        for m in re.finditer(
                r'<a href="' + re.escape(HOST) + r'([^"/]+)/?"[^>]*>(.*?)</a>', html, re.S):
            target = m.group(1)
            anchor = re.sub(r"<[^>]+>", "", m.group(2)).strip().lower()
            if not anchor:
                continue
            # Si el anchor ya casa con el destino actual, el enlace está bien:
            # no hay nada que "envejecer". (Evita marcar p.ej. un anchor sobre
            # el vídeo viral de Roswell que apunta —correctamente— a ese post
            # solo porque también contiene la palabra "roswell".)
            tgt_toks = tokmap.get(target) or set()
            if tgt_toks and all(t in anchor for t in tgt_toks):
                continue
            for other, toks in tokmap.items():
                # <2 tokens distintivos = identificador débil (p.ej. {roswell}),
                # casa con demasiadas cosas. No es señal suficiente.
                if other in (target, p["slug"]) or len(toks) < 2:
                    continue
                if all(t in anchor for t in toks):
                    out.append((p["slug"], anchor[:40], target, other))
    return out


def check_sitemap():
    malos = []
    urls = []
    for part in ("posts", "pages"):
        r = sh(f"curl -s --max-time 25 {HOST}sitemap-{part}.xml")
        urls += re.findall(r"<loc>(.*?)</loc>", r.stdout)
    with ThreadPoolExecutor(max_workers=8) as ex:
        for u, (code, _) in zip(urls, ex.map(lambda x: http_status(x), urls)):
            if code != 200:
                malos.append((u, code))
    return len(urls), malos


def check_rules(posts, site):
    """Violaciones accionables (excluye waivers e INFO), agrupadas por regla."""
    out = {}
    for p in posts:
        if p.get("status") != "published":
            continue
        for v in R.check_post(p):
            if v.severity == R.INFO or X.accepted_reason(v.rule, p["slug"], site):
                continue
            out.setdefault(v.rule, []).append(p["slug"])
    return out


# ---------- salida ----------

def build_report(roto, nocanon, prematuro, envejecido, n_sitemap, sm_malos, reglas):
    L = []
    if roto:
        L.append(f"🔴 {len(roto)} enlace(s) interno(s) ROTO(S):")
        L += [f"   {s[:34]} → {u.replace(HOST,'/')}" for s, u in roto[:6]]
    if prematuro:
        L.append(f"🟠 {len(prematuro)} enlace(s) PREMATURO(S) (destino se publica después → 404 temporal):")
        L += [f"   {s[:28]} ({sp}) → {t[:28]} ({tp})" for s, t, sp, tp in prematuro[:6]]
    if nocanon:
        L.append(f"🟡 {len(nocanon)} enlace(s) NO CANÓNICO(S) (salto 301 evitable):")
        L += [f"   {s[:34]} → {u}" for s, u in nocanon[:6]]
    if envejecido:
        L.append(f"🔵 {len(envejecido)} enlace(s) ENVEJECIDO(S) (ya existe mejor destino):")
        L += [f"   {s[:26]}: «{a}» → /{t[:24]}/ ⇒ /{m[:26]}/" for s, a, t, m in envejecido[:6]]
    if sm_malos:
        L.append(f"🔴 sitemap: {len(sm_malos)} de {n_sitemap} URL(s) no dan 200:")
        L += [f"   [{c}] {u.replace(HOST,'/')}" for u, c in sm_malos[:6]]
    if reglas:
        tot = sum(len(v) for v in reglas.values())
        L.append(f"📋 {tot} violación(es) accionable(s) del auditor:")
        L += [f"   {k}: {len(v)}" for k, v in sorted(reglas.items(), key=lambda x: -len(x[1]))[:6]]
    if not L:
        L.append("✅ Sin hallazgos: enlaces, sitemap y reglas OK.")
    return "\n".join(L)


def fingerprint(roto, nocanon, prematuro, envejecido, sm_malos, reglas):
    return json.dumps({
        "roto": sorted(f"{s}|{u}" for s, u in roto),
        "nocanon": sorted(f"{s}|{u}" for s, u in nocanon),
        "prematuro": sorted(f"{s}|{t}" for s, t, _, _ in prematuro),
        "envejecido": sorted(f"{s}|{t}|{m}" for s, _, t, m in envejecido),
        "sitemap": sorted(u for u, _ in sm_malos),
        "reglas": {k: sorted(v) for k, v in reglas.items()},
    }, sort_keys=True)


def telegram(text):
    r = sh(f"kubectl get secret -n {TG_SECRET[0]} {TG_SECRET[1]} -o jsonpath='{{.data.TELEGRAM_BOT_TOKEN}}'")
    import base64
    token = base64.b64decode(r.stdout).decode()
    r2 = sh(f"kubectl get secret -n {TG_SECRET[0]} {TG_SECRET[1]} -o jsonpath='{{.data.TELEGRAM_CHAT_ID}}'")
    chat = base64.b64decode(r2.stdout).decode()
    data = urllib.parse.urlencode({"chat_id": chat, "text": text[:3900]}).encode()
    req = urllib.request.Request(f"https://api.telegram.org/bot{token}/sendMessage", data=data)
    with urllib.request.urlopen(req, timeout=30) as resp:
        return json.loads(resp.read()).get("ok", False)


def run_site(site):
    """Corre todos los checks sobre un sitio. Devuelve (informe, huella, hay_hallazgos)."""
    cfg = use_site(site)
    posts = fetch_posts()
    roto, nocanon, prematuro = check_links(posts)
    envejecido = check_link_aging(posts)
    n_sm, sm_malos = check_sitemap()
    reglas = check_rules(posts, site)

    body = build_report(roto, nocanon, prematuro, envejecido, n_sm, sm_malos, reglas)
    pub = sum(1 for p in posts if p.get("status") == "published")
    header = (f"── [{site.upper()}] {cfg['host']}\n"
              f"   {pub} publicados, {len(posts)} totales, sitemap {n_sm} URLs")
    fp = fingerprint(roto, nocanon, prematuro, envejecido, sm_malos, reglas)
    hallazgos = bool(roto or nocanon or prematuro or envejecido or sm_malos)
    return header + "\n" + body, fp, hallazgos


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--dry-run", action="store_true", help="imprime, no notifica ni guarda estado")
    ap.add_argument("--force", action="store_true", help="notifica aunque no haya cambios")
    ap.add_argument("--site", choices=("en", "es", "both"), default="both")
    a = ap.parse_args()

    sitios = ["en", "es"] if a.site == "both" else [a.site]
    partes, huellas, hallazgos = [], {}, False
    for s in sitios:
        try:
            txt, fp, hall = run_site(s)
        except Exception as e:  # un sitio caído no debe tumbar el informe del otro
            txt, fp, hall = f"── [{s.upper()}] ⚠ ERROR al revisar: {e}", f"error:{e}", True
        partes.append(txt); huellas[s] = fp; hallazgos |= hall

    report = "🔎 SEO watch\n\n" + "\n\n".join(partes)
    fp = json.dumps(huellas, sort_keys=True)

    prev = ""
    try:
        with open(STATE) as f:
            prev = json.load(f).get("fingerprint", "")
    except FileNotFoundError:
        pass
    changed = fp != prev

    print(report)
    print(f"\n[cambios vs última ejecución: {'sí' if changed else 'no'}]")

    if a.dry_run:
        return 0

    # Notifica si hay hallazgos y algo cambió (o si se fuerza)
    if a.force or (hallazgos and changed):
        sufijo = "" if changed else "\n\n(sin cambios desde la última revisión)"
        ok = telegram(report + sufijo)
        print(f"[telegram: {'enviado' if ok else 'FALLO'}]")

    os.makedirs(os.path.dirname(STATE), exist_ok=True)
    with open(STATE, "w") as f:
        json.dump({"fingerprint": fp}, f)
    return 0


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