seo-watch: vigila la indexación en Google (ventana 5-60 días, huella sin la antigüedad)

seo_gsc.py

@@ -599,6 +599,59 @@ def cmd_antes_despues(a):
 
 # ──────────────────────────────── utilidades ────────────────────────────────
 
+def url_canonica(site, slug_o_url):
+    """La URL tal y como Google la tiene. Ojo: EN canoniza en **www** y ES en el
+    APEX, al revés — declararlo en un solo sitio evita inspeccionar una URL que
+    la propiedad ni siquiera ve."""
+    if slug_o_url.startswith("http"):
+        return slug_o_url
+    base = ("https://www.theexclusionzone.com/" if site == "en"
+            else "https://zonadeexclusion.com/")
+    return f"{base}{slug_o_url.strip('/')}/"
+
+
+def inspecciona_url(tok, site, slug_o_url):
+    """Estado de indexación de UNA url según Google.
+
+    Devuelve el `indexStatusResult` con la `url` consultada añadida, o
+    `{"error": "<código>"}` si la API no contesta 200. Existe como función y no
+    solo como comando porque la usan dos consumidores —este CLI y el vigilante
+    diario— y el endpoint, la propiedad y el host canónico deben declararse una
+    vez, no dos.
+    """
+    import requests
+    url = url_canonica(site, slug_o_url)
+    r = requests.post(
+        "https://searchconsole.googleapis.com/v1/urlInspection/index:inspect",
+        headers={"Authorization": f"Bearer {tok}"}, timeout=60,
+        json={"inspectionUrl": url, "siteUrl": SITIOS[site]["propiedad"]})
+    if r.status_code != 200:
+        return {"error": str(r.status_code), "detalle": r.text[:200], "url": url}
+    d = dict(r.json()["inspectionResult"]["indexStatusResult"])
+    d["url"] = url
+    return d
+
+
+def etiqueta_indexacion(d):
+    """(icono, texto corto) de un `indexStatusResult`.
+
+    Los tres estados de «no indexado» NO son el mismo problema y por eso no se
+    colapsan: «no lo conoce» es que nunca llegó, «descubierto» es que está en
+    cola, y «rastreado sin indexar» es que lo leyó y decidió no quedárselo —
+    solo el último apunta al contenido.
+    """
+    if d.get("error"):
+        return "❓", f"error {d['error']}"
+    cob = d.get("coverageState", "?")
+    if d.get("verdict") == "PASS":
+        return "✅", "indexado"
+    if cob.startswith("Crawled"):
+        return "🟡", "rastreado, sin indexar"
+    if cob.startswith("Discovered"):
+        return "🔵", "descubierto, sin rastrear"
+    return "⏳", "Google no lo conoce"
+
+
 def cmd_inspecciona(a):
     """¿Conoce Google esta URL? Veredicto suyo, no deducción nuestra.
 
@@ -607,31 +660,22 @@ def cmd_inspecciona(a):
     unknown to Google»: nunca rastreados. Sin esto se habría confundido con un
     problema de contenido y se habrían reescrito titulares para nada.
     """
-    import requests
     tok = token(carga_credencial())
-    prop = SITIOS[a.site]["propiedad"]
-    base = ("https://www.theexclusionzone.com/" if a.site == "en"
-            else "https://zonadeexclusion.com/")
     malas = 0
     for slug in a.slugs:
-        url = slug if slug.startswith("http") else f"{base}{slug.strip('/')}/"
-        r = requests.post(
-            "https://searchconsole.googleapis.com/v1/urlInspection/index:inspect",
-            headers={"Authorization": f"Bearer {tok}"}, timeout=60,
-            json={"inspectionUrl": url, "siteUrl": prop})
-        if r.status_code != 200:
-            print(f"  ✗ {r.status_code} {url}\n     {r.text[:200]}")
+        i = inspecciona_url(tok, a.site, slug)
+        if i.get("error"):
+            print(f"  ✗ {i['error']} {i['url']}\n     {i.get('detalle','')}")
             malas += 1
             continue
-        i = r.json()["inspectionResult"]["indexStatusResult"]
         ok = i.get("verdict") == "PASS"
         malas += 0 if ok else 1
         print(f"  {'✓' if ok else '⚠'} {i.get('coverageState', '?')}")
         print(f"      rastreo {i.get('lastCrawlTime', '—')[:10]}  "
               f"robots {i.get('robotsTxtState', '?')}")
-        if i.get("googleCanonical") and i["googleCanonical"] != url:
+        if i.get("googleCanonical") and i["googleCanonical"] != i["url"]:
             print(f"      ⚠ Google canoniza a: {i['googleCanonical']}")
-        print(f"      {url}")
+        print(f"      {i['url']}")
     return 1 if malas else 0
 
 

seo_watch.py

@@ -31,6 +31,7 @@ Diseño (lecciones del 2026-07-18, ver memoria project-en-seo-recovery):
   - Nada de esto modifica Ghost. Las correcciones las decide un humano.
 """
 import argparse
+import datetime as dt
 import json
 import os
 import re
@@ -247,6 +248,70 @@ def check_sitemap():
     return len(urls), malos
 
 
+# ---------- indexación ----------
+
+# Ventana de vigilancia. Por debajo de GRACIA no hay noticia: Google tarda días
+# incluso cuando todo va bien, y avisar el día 1 convierte el vigilante en ruido.
+# Por encima de VENTANA tampoco: un post de hace meses sin indexar ya no es un
+# incidente que se arregle pidiendo indexación, es una conversación sobre la
+# autoridad del dominio, y repetirlo cada mañana no la adelanta.
+DIAS_GRACIA = 5
+DIAS_VENTANA = 60
+
+
+def check_indexacion(posts, site):
+    """Publicados recientes que Google todavía NO ha indexado.
+
+    El resto del vigilante es estructuralmente ciego a esto: mide enlaces,
+    títulos, reglas y sitemap, y un artículo impecable en las cuatro cosas puede
+    llevar semanas sin que Google sepa que existe. En agosto de 2026 fueron SEIS
+    del blog EN a la vez, uno de ellos con ocho días publicado, y el marcador
+    semanal no podía cantarlo porque mide impresiones — y lo que Google no
+    conoce no tiene impresiones que medir.
+
+    Devuelve [(slug, dias, estado)] con SOLO los no indexados, del más viejo al
+    más nuevo. Si la comprobación no se puede hacer devuelve un único
+    ("", 0, "no se pudo comprobar: …"): un chequeo que no corre tiene que verse,
+    no dejar el informe limpio por accidente.
+    """
+    import seo_gsc as G
+    ahora = dt.datetime.now(dt.timezone.utc)
+    cand = []
+    for p in posts:
+        if p.get("status") != "published" or not p.get("published_at"):
+            continue
+        try:
+            pub = dt.datetime.fromisoformat(p["published_at"].replace("Z", "+00:00"))
+        except ValueError:
+            continue
+        dias = (ahora - pub).days
+        if DIAS_GRACIA <= dias <= DIAS_VENTANA:
+            cand.append((p["slug"], dias))
+    if not cand:
+        return []
+    # Orden estable. El desempate por slug no es cosmético: con dos posts del
+    # mismo día el orden dependería del corpus y la huella cambiaría sola,
+    # reenviando hallazgos viejos como si fueran nuevos.
+    cand.sort(key=lambda x: (-x[1], x[0]))
+
+    try:
+        tok = G.token(G.carga_credencial())
+    except Exception as e:
+        return [("", 0, f"no se pudo comprobar: {type(e).__name__}")]
+
+    fuera = []
+    for slug, dias in cand:
+        try:
+            d = G.inspecciona_url(tok, site, slug)
+        except Exception as e:
+            fuera.append((slug, dias, f"error: {type(e).__name__}"))
+            continue
+        icono, txt = G.etiqueta_indexacion(d)
+        if icono != "✅":
+            fuera.append((slug, dias, txt))
+    return fuera
+
+
 # Encabezados en Title Case inglés: el generador del blog ES lo cuela a ratos
 # (el 2026-07-21 había 219 en 23 de 29 posts, pero otros del mismo mes salieron
 # limpios — es una fuga intermitente, no deuda cerrada). Estas dos listas son
@@ -431,8 +496,12 @@ def check_rules(posts, site):
 # ---------- salida ----------
 
 def build_report(roto, nocanon, prematuro, envejecido, n_sitemap, sm_malos, reglas, mayus,
-                 titulos, estado):
+                 titulos, estado, indexa):
     L = []
+    if indexa:
+        L.append(f"🕳 {len(indexa)} publicado(s) que Google NO ha indexado:")
+        L += [f"   {s[:32]} — {e} ({d} días)" if s else f"   ⚠ {e}"
+              for s, d, e in indexa[:6]]
     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]]
@@ -468,9 +537,13 @@ def build_report(roto, nocanon, prematuro, envejecido, n_sitemap, sm_malos, regl
 
 
 def fingerprint(roto, nocanon, prematuro, envejecido, sm_malos, reglas, mayus, titulos,
-                estado):
+                estado, indexa):
     return json.dumps({
         "estado": {k: sorted(v) for k, v in estado.items()},
+        # Sin los días A PROPÓSITO: la antigüedad sube cada mañana, así que
+        # meterla aquí haría cambiar la huella sola y el vigilante repetiría el
+        # mismo hallazgo a diario hasta que dejaras de leerlo.
+        "indexa": sorted(f"{s}|{e}" for s, _, e in indexa),
         "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),
@@ -508,16 +581,17 @@ def run_site(site):
     # programados). Van aquí y no en un timer aparte porque comparten lo que
     # hace útil a este vigilante: una sola huella por sitio y un solo aviso.
     estado = E.revisa(site, posts)
+    indexa = check_indexacion(posts, site)
 
     body = build_report(roto, nocanon, prematuro, envejecido, n_sm, sm_malos, reglas, mayus,
-                        titulos, estado)
+                        titulos, estado, indexa)
     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, mayus, titulos,
-                     estado)
+                     estado, indexa)
     hallazgos = bool(roto or nocanon or prematuro or envejecido or sm_malos or mayus
-                     or titulos or estado)
+                     or titulos or estado or indexa)
     return header + "\n" + body, fp, hallazgos
 
 

test_seo_watch.py

@@ -0,0 +1,163 @@
+#!/usr/bin/env python3
+"""Tests del chequeo de INDEXACIÓN de seo_watch.py.
+
+Cada uno es la cicatriz de algo que ya salió mal en producción una vez:
+
+  1. avisar demasiado pronto convierte el vigilante en ruido y se deja de leer
+  2. una huella que cambia sola reenvía hallazgos viejos como si fueran nuevos
+     (ver la memoria «orden no determinista y huellas»)
+  3. un chequeo que NO puede correr tiene que verse; si se traga la excepción,
+     el informe sale limpio justo el día en que estaba ciego
+
+    python3 -m pytest test_seo_watch.py -q
+    python3 test_seo_watch.py
+    for h in 1 2 3; do PYTHONHASHSEED=$h python3 test_seo_watch.py; done
+"""
+import datetime as dt
+import os
+import sys
+import types
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import seo_watch as W
+
+
+def _post(slug, dias, status="published"):
+    cuando = dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=dias, hours=1)
+    return {"slug": slug, "status": status,
+            "published_at": cuando.strftime("%Y-%m-%dT%H:%M:%S.000Z")}
+
+
+def _con_gsc(respuestas, credencial_rota=False):
+    """Inyecta un seo_gsc de mentira. `respuestas` es {slug: (icono, texto)}."""
+    fake = types.ModuleType("seo_gsc")
+    fake.consultados = []
+
+    def carga_credencial():
+        if credencial_rota:
+            raise FileNotFoundError("no hay service-account.json")
+        return {"client_email": "x"}
+
+    fake.carga_credencial = carga_credencial
+    fake.token = lambda sa: "tok"
+
+    def inspecciona_url(tok, site, slug):
+        fake.consultados.append(slug)
+        return {"slug": slug}
+
+    fake.inspecciona_url = inspecciona_url
+    fake.etiqueta_indexacion = lambda d: respuestas.get(d["slug"], ("✅", "indexado"))
+    sys.modules["seo_gsc"] = fake
+    return fake
+
+
+def test_no_avisa_de_un_post_recien_publicado():
+    """Google tarda días aunque todo vaya bien: avisar el día 1 es ruido."""
+    g = _con_gsc({"nuevo": ("⏳", "Google no lo conoce")})
+    fuera = W.check_indexacion([_post("nuevo", 1)], "en")
+    assert fuera == [], fuera
+    assert g.consultados == [], "ni siquiera debe gastar cuota de API"
+
+
+def test_avisa_pasado_el_periodo_de_gracia():
+    _con_gsc({"viejo": ("⏳", "Google no lo conoce")})
+    fuera = W.check_indexacion([_post("viejo", W.DIAS_GRACIA + 3)], "en")
+    assert len(fuera) == 1, fuera
+    assert fuera[0][0] == "viejo" and "no lo conoce" in fuera[0][2], fuera
+
+
+def test_calla_lo_que_si_esta_indexado():
+    """Solo habla de lo roto: un post sano no debe aparecer en el informe."""
+    _con_gsc({})          # todo devuelve ✅
+    assert W.check_indexacion([_post("sano", 20)], "en") == []
+
+
+def test_ignora_lo_muy_viejo_y_lo_no_publicado():
+    """Fuera de ventana ya no es un incidente que se arregle pidiendo indexación."""
+    g = _con_gsc({"antiguo": ("⏳", "x"), "borrador": ("⏳", "x")})
+    posts = [_post("antiguo", W.DIAS_VENTANA + 10),
+             _post("borrador", 30, status="draft")]
+    assert W.check_indexacion(posts, "en") == []
+    assert g.consultados == []
+
+
+def test_la_huella_NO_cambia_cuando_solo_pasa_un_dia():
+    """El fallo que mataría al vigilante: la antigüedad sube cada mañana, así
+    que si entrase en la huella repetiría el mismo hallazgo a diario."""
+    ayer = [("release-5", 21, "Google no lo conoce")]
+    hoy = [("release-5", 22, "Google no lo conoce")]
+    vacio = ([], [], [], [], [], {}, [], [], {})
+    a = W.fingerprint(*vacio, ayer)
+    b = W.fingerprint(*vacio, hoy)
+    assert a == b, "un día más no puede contar como hallazgo nuevo"
+
+
+def test_la_huella_SI_cambia_cuando_cambia_el_estado():
+    """Pasar de «no lo conoce» a «descubierto» es noticia: hubo movimiento."""
+    vacio = ([], [], [], [], [], {}, [], [], {})
+    a = W.fingerprint(*vacio, [("release-5", 21, "Google no lo conoce")])
+    b = W.fingerprint(*vacio, [("release-5", 21, "descubierto, sin rastrear")])
+    assert a != b
+
+
+def test_orden_estable_con_empate_de_dias():
+    """Dos posts del mismo día: sin desempate por slug el orden lo decide el
+    corpus y la huella cambia sola de una ejecución a otra."""
+    _con_gsc({"bbb": ("⏳", "x"), "aaa": ("⏳", "x")})
+    posts = [_post("bbb", 10), _post("aaa", 10)]
+    ida = [s for s, _, _ in W.check_indexacion(posts, "en")]
+    _con_gsc({"bbb": ("⏳", "x"), "aaa": ("⏳", "x")})
+    vuelta = [s for s, _, _ in W.check_indexacion(list(reversed(posts)), "en")]
+    assert ida == vuelta == ["aaa", "bbb"], (ida, vuelta)
+
+
+def test_el_mas_viejo_va_primero():
+    _con_gsc({"a": ("⏳", "x"), "b": ("⏳", "x")})
+    fuera = W.check_indexacion([_post("a", 8), _post("b", 30)], "en")
+    assert [s for s, _, _ in fuera] == ["b", "a"], fuera
+
+
+def test_si_no_hay_credencial_se_ve_en_el_informe():
+    """Un chequeo que no corre NO puede dejar el informe limpio."""
+    _con_gsc({}, credencial_rota=True)
+    fuera = W.check_indexacion([_post("x", 10)], "en")
+    assert len(fuera) == 1 and "no se pudo comprobar" in fuera[0][2], fuera
+    linea = W.build_report([], [], [], [], 1, [], {}, [], [], {}, fuera)
+    assert "no se pudo comprobar" in linea, linea
+
+
+def test_un_error_de_la_api_no_tumba_el_resto():
+    """Si una URL revienta, las demás se siguen comprobando."""
+    fake = _con_gsc({"malo": ("❓", "error 429"), "bueno": ("⏳", "no lo conoce")})
+
+    def revienta(tok, site, slug):
+        fake.consultados.append(slug)
+        if slug == "malo":
+            raise TimeoutError("timeout")
+        return {"slug": slug}
+
+    fake.inspecciona_url = revienta
+    fuera = W.check_indexacion([_post("malo", 12), _post("bueno", 11)], "en")
+    assert len(fuera) == 2, fuera
+    assert any("TimeoutError" in e for _, _, e in fuera), fuera
+
+
+def test_el_informe_saca_los_hallazgos_de_indexacion():
+    txt = W.build_report([], [], [], [], 1, [], {}, [], [], {},
+                         [("release-5", 22, "Google no lo conoce")])
+    assert "NO ha indexado" in txt and "release-5" in txt and "22 días" in txt, txt
+
+
+if __name__ == "__main__":
+    fallos = 0
+    for nombre, fn in sorted(globals().items()):
+        if not nombre.startswith("test_"):
+            continue
+        try:
+            fn()
+            print(f"  ✓ {nombre}")
+        except AssertionError as exc:
+            fallos += 1
+            print(f"  ✗ {nombre}: {exc}")
+    print(f"\n{'✓ todo pasa' if not fallos else f'✗ {fallos} fallo(s)'}")
+    sys.exit(1 if fallos else 0)