chemavx-seo-tools/indexnow_watch.py

#!/usr/bin/env python3
"""indexnow_watch — avisa a IndexNow (Bing/DDG/Yandex/Naver) de URLs nuevas o
actualizadas en los dos blogs, comparando el sitemap de posts con el estado.

Patrón seo-watch/remate-watch: timer systemd en el master, estado en disco,
Telegram solo cuando pasa algo (📡 envío, 🚨 fallo). Google NO participa en
IndexNow: a Google lo cubre el sitemap + crawl diario, esto es la vía rápida
del resto de buscadores.

Las keys NO viven en este fichero (el repo podría publicarse): están en
CONFIG (~/.local/state/indexnow/config.json) y el fichero {key}.txt se sirve
desde content/files/indexnow/ del PVC de cada Ghost (verificado 200 por el
dominio real el 2026-07-24).

Uso:
  indexnow_watch.py --dry-run          # qué enviaría, sin enviar
  indexnow_watch.py --bootstrap        # primera vez: envía TODO el sitemap
  indexnow_watch.py                    # normal: solo lo nuevo/cambiado
Ensayo de fallo: INDEXNOW_ENDPOINT=https://127.0.0.1:9/ indexnow_watch.py
"""
import argparse
import base64
import json
import os
import re
import subprocess
import sys
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from pathlib import Path

STATE_DIR = Path.home() / ".local/state/indexnow"
STATE_FILE = STATE_DIR / "state.json"
CONFIG_FILE = STATE_DIR / "config.json"
ENDPOINT = os.environ.get("INDEXNOW_ENDPOINT", "https://api.indexnow.org/indexnow")
TG_SECRET = ("monitoring", "grafana-telegram-infisical")

# host y sitemap por sitio; key/keyLocation vienen de CONFIG_FILE
SITES = {
    "es": {"host": "zonadeexclusion.com",
           "sitemap": "https://zonadeexclusion.com/sitemap-posts.xml"},
    "en": {"host": "www.theexclusionzone.com",
           "sitemap": "https://www.theexclusionzone.com/sitemap-posts.xml"},
}


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


def telegram(text):
    r = sh(f"kubectl get secret -n {TG_SECRET[0]} {TG_SECRET[1]} -o jsonpath='{{.data.TELEGRAM_BOT_TOKEN}}'")
    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 fetch_sitemap(url):
    """Devuelve {url: lastmod} del sitemap de posts."""
    req = urllib.request.Request(url, headers={"User-Agent": "indexnow-watch/1.0"})
    with urllib.request.urlopen(req, timeout=30) as resp:
        xml = resp.read().decode()
    out = {}
    for m in re.finditer(r"<url>(.*?)</url>", xml, re.S):
        block = m.group(1)
        loc = re.search(r"<loc>\s*(.*?)\s*</loc>", block)
        lastmod = re.search(r"<lastmod>\s*(.*?)\s*</lastmod>", block)
        if loc:
            out[loc.group(1)] = lastmod.group(1) if lastmod else ""
    return out


def submit(site_cfg, urls):
    """POST a IndexNow. 200/202 = aceptado."""
    body = json.dumps({
        "host": site_cfg["host"],
        "key": site_cfg["key"],
        "keyLocation": site_cfg["keyLocation"],
        "urlList": urls,
    }).encode()
    req = urllib.request.Request(ENDPOINT, data=body,
                                 headers={"Content-Type": "application/json; charset=utf-8"})
    with urllib.request.urlopen(req, timeout=30) as resp:
        return resp.status


def load_json(path, default):
    try:
        return json.loads(path.read_text())
    except (FileNotFoundError, json.JSONDecodeError):
        return default


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--dry-run", action="store_true")
    ap.add_argument("--bootstrap", action="store_true",
                    help="permite el primer envío masivo con estado vacío")
    ap.add_argument("--site", choices=["en", "es", "both"], default="both")
    args = ap.parse_args()

    STATE_DIR.mkdir(parents=True, exist_ok=True)
    config = load_json(CONFIG_FILE, {})
    state = load_json(STATE_FILE, {})
    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
    report, failed = [], False

    for site in (["en", "es"] if args.site == "both" else [args.site]):
        cfg = dict(SITES[site])
        if site not in config or "key" not in config[site]:
            print(f"[{site}] sin key en {CONFIG_FILE}, me lo salto", file=sys.stderr)
            failed = True
            continue
        cfg.update(config[site])
        known = state.get(site, {})

        try:
            current = fetch_sitemap(cfg["sitemap"])
        except Exception as e:
            report.append(f"🚨 indexnow-watch [{site.upper()}]: no pude leer el sitemap: {e}")
            failed = True
            continue

        pending = [u for u, lm in current.items() if known.get(u) != lm]

        if not known and pending and not args.bootstrap:
            print(f"[{site}] estado vacío y {len(pending)} URLs: usa --bootstrap "
                  "para el primer envío masivo", file=sys.stderr)
            continue

        if not pending:
            print(f"[{site}] sin novedades ({len(current)} URLs en sitemap)")
            # poda de URLs despublicadas para que una republicación reenvíe
            state[site] = {u: lm for u, lm in current.items() if u in known}
            continue

        print(f"[{site}] {len(pending)} URL(s) para IndexNow:")
        for u in pending:
            print(f"  {u}")
        if args.dry_run:
            continue

        try:
            status = submit(cfg, pending)
        except Exception as e:
            status, err = None, e
        if status in (200, 202):
            state[site] = dict(current)
            slugs = ", ".join(u.rstrip("/").rsplit("/", 1)[-1] for u in pending[:6])
            extra = f" (+{len(pending) - 6} más)" if len(pending) > 6 else ""
            report.append(f"📡 IndexNow [{site.upper()}]: {len(pending)} URL(s) "
                          f"enviadas (HTTP {status}): {slugs}{extra}")
        else:
            failed = True
            detail = f"HTTP {status}" if status else f"{err}"
            report.append(f"🚨 indexnow-watch [{site.upper()}]: envío FALLÓ ({detail}); "
                          f"{len(pending)} URL(s) quedan pendientes para el próximo tick")

    if not args.dry_run:
        STATE_FILE.write_text(json.dumps(state, indent=1))
        if report:
            try:
                telegram("\n\n".join(report))
            except Exception as e:
                print(f"telegram falló: {e}", file=sys.stderr)
                failed = True
    for line in report:
        print(line)
    sys.exit(1 if failed else 0)


if __name__ == "__main__":
    main()