diff --git a/seo-watch.service b/seo-watch.service new file mode 100644 index 0000000..0cf6f4b --- /dev/null +++ b/seo-watch.service @@ -0,0 +1,15 @@ +[Unit] +Description=SEO watch — regresiones de enlaces/sitemap en theexclusionzone.com +Documentation=file:///home/chemavx/seo-tools/seo_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/seo_watch.py +TimeoutStartSec=900 diff --git a/seo-watch.timer b/seo-watch.timer new file mode 100644 index 0000000..cb50333 --- /dev/null +++ b/seo-watch.timer @@ -0,0 +1,10 @@ +[Unit] +Description=SEO watch semanal (lunes 05:30 UTC, antes de la publicación de las 06:00) + +[Timer] +OnCalendar=Mon *-*-* 05:30:00 UTC +Persistent=true +RandomizedDelaySec=300 + +[Install] +WantedBy=timers.target diff --git a/seo_watch.py b/seo_watch.py new file mode 100644 index 0000000..a8cdb39 --- /dev/null +++ b/seo_watch.py @@ -0,0 +1,243 @@ +#!/usr/bin/env python3 +""" +Tool D — vigilante SEO periódico de theexclusionzone.com (EN, 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. Sitemap (toda URL listada debe dar 200 directo) + 5. 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 + +HOST = "https://www.theexclusionzone.com/" +STATE = os.path.expanduser("~/.local/state/seo-watch/last.json") +TG_SECRET = ("monitoring", "grafana-telegram-infisical") +TIMEOUT = 20 + + +# ---------- 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"ghst-en 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 + + +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"(.*?)", 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): + """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"]): + continue + out.setdefault(v.rule, []).append(p["slug"]) + return out + + +# ---------- salida ---------- + +def build_report(roto, nocanon, prematuro, 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 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, 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), + "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 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") + a = ap.parse_args() + + posts = fetch_posts() + roto, nocanon, prematuro = check_links(posts) + n_sm, sm_malos = check_sitemap() + reglas = check_rules(posts) + + body = build_report(roto, nocanon, prematuro, n_sm, sm_malos, reglas) + pub = sum(1 for p in posts if p.get("status") == "published") + header = f"🔎 SEO watch — theexclusionzone.com\n({pub} publicados, {len(posts)} totales, sitemap {n_sm} URLs)\n" + report = header + "\n" + body + + fp = fingerprint(roto, nocanon, prematuro, sm_malos, reglas) + prev = "" + try: + with open(STATE) as f: + prev = json.load(f).get("fingerprint", "") + except FileNotFoundError: + pass + changed = fp != prev + hallazgos = bool(roto or nocanon or prematuro or sm_malos) + + 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())