#!/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"(.*?)", xml, re.S):
block = m.group(1)
loc = re.search(r"\s*(.*?)\s*", block)
lastmod = re.search(r"\s*(.*?)\s*", 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()