ChemaVX
f7be9ccb3e
Toda la caja miraba los enlaces en un solo sentido — a quién enlazo yo — y
esa es la pregunta del lector, no la de Google. Dos posts de EN publicados
el 21-jul seguían siendo «URL is unknown to Google» una semana después:
tenían enlaces entrantes, pero solo desde artículos con 8 y 12 impresiones
de por vida, páginas que Google casi no visita.
seo_semantic gana dos subcomandos: `padrinos <slug>`, que ordena candidatos
por tráfico REAL del origen (Search Console) en vez de por similitud, y
`descubrimiento`, que lista los publicados a los que no llega ni un enlace
desde el tercio fuerte del sitio. Salen 21 de 32 en EN y 21 de 31 en ES.
Ambos descartan los posts con canonical_url propio: Ghost los excluye del
sitemap por consolidados, así que apadrinar desde ahí no sirve de nada — el
de Grusch de mayo tiene 1.728 impresiones y habría encabezado la lista.
seo_gsc gana `inspecciona`, que pregunta el veredicto a Google en vez de
deducirlo. Es lo que separa «posiciona mal» de «no lo ha visto nunca», que
son problemas opuestos y se arreglan al revés.
Y el checklist del remate incorpora el paso 3b: el informe tiene que traer
el comando de enlace entrante listo para copiar. No lo aplica el agente
porque eso es editar otro post publicado, y eso sigue prohibido.
⚠️ Anotado en el docstring: contar enlaces internos desde el HTML público da
«cero huérfanos», porque el tema mete navegación y feed de relacionados que
parecen enlaces del cuerpo. La verdad está en el html del CLI de Ghost.
613 lines
27 KiB
Python
613 lines
27 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Tool H — el marcador: datos reales de Search Console para los dos blogs.
|
|
|
|
Por qué existe. El resto de la caja de herramientas mira HACIA DENTRO: enlaces
|
|
rotos, sitemap, metas que incumplen una regla, canibalización entre posts. Todo
|
|
eso dice si el sitio está bien construido, no si alguien lo encuentra ni si lo
|
|
pincha. Los únicos datos de resultado que había en la máquina eran una
|
|
exportación manual del 2026-06-17 — anterior a la optimización de metas del 23
|
|
de junio, o sea justo anterior al trabajo que pretendía mejorarlos. Cuarenta
|
|
días optimizando sin saber el resultado.
|
|
|
|
Qué NO hace: escribir en los blogs. Es de solo lectura de punta a punta.
|
|
|
|
Autenticación. Cuenta de servicio, sin OAuth: en esta máquina no hay navegador
|
|
y el flujo interactivo no funciona headless (misma piedra que con Gemini). No
|
|
usa las librerías de Google: se firma el JWT con PyJWT, se canjea por un token
|
|
y se llama al REST. Son treinta líneas y evita arrastrar media docena de
|
|
dependencias para tres peticiones.
|
|
|
|
Subcomandos:
|
|
|
|
seo_gsc.py sitios
|
|
Qué propiedades ve la cuenta de servicio. ES EL PRIMER PASO: hasta que
|
|
aparezcan aquí, el permiso en Search Console no está dado.
|
|
|
|
seo_gsc.py pull [--site en|es] [--desde F] [--hasta F]
|
|
Descarga y acumula en SQLite. Idempotente: reejecutar el mismo rango
|
|
no duplica. Sin --desde arranca donde se quedó (o 16 meses atrás la
|
|
primera vez, que es todo el historial que Google guarda).
|
|
|
|
seo_gsc.py informe [--site en|es] [--dias N] [--telegram]
|
|
El marcador. Primero COBERTURA (impresiones, páginas que reciben
|
|
alguna, consultas distintas, posición) y qué páginas empiezan o dejan
|
|
de aparecer; después conversión y las páginas que posicionan bien y no
|
|
se pinchan. Ese orden no es estético: medido el 2026-07-28, con ~180
|
|
impresiones semanales en EN el techo de optimizar titulares son unas
|
|
decenas de clics al mes. Lo que mueve la aguja es aparecer más.
|
|
|
|
seo_gsc.py antes-despues --corte AAAA-MM-DD [--dias N]
|
|
Compara la misma ventana antes y después de una fecha. Escrito para
|
|
responder de una vez si el remate de metas del 23-jun sirvió de algo.
|
|
|
|
seo_gsc.py inspecciona <slug…> --site en|es
|
|
El veredicto de Google sobre una URL: indexada, rastreada cuándo, y a
|
|
qué canoniza. Distingue «posiciona mal» de «no lo ha visto nunca», que
|
|
son problemas opuestos. Sale con código 1 si alguna no está indexada.
|
|
|
|
seo_gsc.py comprueba
|
|
Canario: recorre el camino REAL (credencial → firma → token → API) y
|
|
dice en qué eslabón exacto se rompe. No simula nada.
|
|
|
|
Gotchas codificados aquí:
|
|
* Los datos de GSC tardan 2-3 días en consolidarse: `dataState=final` y el
|
|
rango termina 3 días atrás. Pedir ayer devuelve cifras que luego cambian.
|
|
* La API pagina de 25.000 en 25.000 (startRow), y no avisa de que hay más.
|
|
* El nombre de la propiedad NO es el dominio: puede ser `sc-domain:x.com` o
|
|
`https://www.x.com/`, y con la barra final. Por eso existe `sitios`.
|
|
"""
|
|
import argparse
|
|
import datetime as dt
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import urllib.parse
|
|
|
|
SCOPE = "https://www.googleapis.com/auth/webmasters.readonly"
|
|
TOKEN_URL = "https://oauth2.googleapis.com/token"
|
|
API = "https://searchconsole.googleapis.com/webmasters/v3"
|
|
|
|
CRED = os.path.expanduser(os.environ.get("GSC_CRED", "~/.config/gsc/service-account.json"))
|
|
BD = os.path.expanduser(os.environ.get("GSC_DB", "~/.local/state/seo-gsc/gsc.db"))
|
|
TG_SECRET = ("monitoring", "grafana-telegram-infisical")
|
|
|
|
# La propiedad se confirma con `sitios`; esto es solo el intento por defecto.
|
|
# ⚠️ El nombre tiene que ser EXACTO al que devuelve `seo-gsc sitios`. Las dos
|
|
# propiedades no son del mismo tipo: EN es de dominio (cubre cualquier
|
|
# subdominio y esquema) y ES es de prefijo de URL, así que solo ve lo que
|
|
# cuelgue de https://zonadeexclusion.com/ — ni www ni http. Vale porque el
|
|
# canónico del ES es el apex y www redirige con 301, pero si algún día se
|
|
# publica bajo www, en este marcador no aparecería.
|
|
SITIOS = {
|
|
"en": {"host": "theexclusionzone.com", "propiedad": "sc-domain:theexclusionzone.com"},
|
|
"es": {"host": "zonadeexclusion.com", "propiedad": "https://zonadeexclusion.com/"},
|
|
}
|
|
|
|
RETRASO_DIAS = 3 # margen para que Google consolide
|
|
|
|
# ⚠️ Calibrado con datos reales el 2026-07-28, y el primer intento estaba mal.
|
|
# Con IMPRESIONES_MIN = 150 (razonable para un sitio con tráfico) NADA pasaba el
|
|
# filtro: el mejor caso de los dos blogs era la página de Varginha en EN con 125
|
|
# impresiones en 28 días. El vigilante se habría callado para siempre, que es la
|
|
# peor avería posible en algo que solo habla cuando hay noticias.
|
|
#
|
|
# 40 es el compromiso: por debajo, un solo clic mueve el CTR dos puntos y medio
|
|
# y cualquier conclusión es inventada. Entre 40 y 150 el dato es orientativo y
|
|
# el informe lo marca con «~» para que nadie lo lea como si fuese firme.
|
|
IMPRESIONES_MIN = 40
|
|
IMPRESIONES_FIRMES = 150 # a partir de aquí el CTR deja de ser orientativo
|
|
FRACCION_ALERTA = 0.45 # se avisa si el CTR no llega a esta parte del esperado
|
|
|
|
# Cuánto tiene que moverse la cobertura para merecer un aviso. A este volumen
|
|
# (~180 impresiones/semana en EN, ~40 en ES) el CTR no es la palanca: aunque
|
|
# convirtiéramos el 10% de TODAS las impresiones de EN serían 78 clics al mes.
|
|
# El techo lo pone cuánta gente nos ve, no el titular. Por eso el marcador vigila
|
|
# primero la cobertura y solo después el CTR.
|
|
UMBRAL_MOV = 0.25 # cambio relativo de impresiones que merece un aviso
|
|
UMBRAL_PAGINAS = 3 # o este salto en nº de páginas que reciben impresiones
|
|
|
|
# CTR esperado por posición. Es una curva de referencia del sector, no una
|
|
# medición nuestra: sirve para ORDENAR candidatas, no para prometer tráfico.
|
|
CTR_ESPERADO = {1: .270, 2: .150, 3: .110, 4: .080, 5: .062,
|
|
6: .050, 7: .042, 8: .035, 9: .030, 10: .026}
|
|
|
|
|
|
# ─────────────────────────────── credencial ───────────────────────────────
|
|
|
|
def carga_credencial():
|
|
if not os.path.exists(CRED):
|
|
sys.exit(
|
|
f"✗ No encuentro la credencial en {CRED}\n"
|
|
" Es una clave JSON de cuenta de servicio. Ver la cabecera de\n"
|
|
" MANUAL-GSC.md para los cuatro pasos de navegador que hacen falta."
|
|
)
|
|
if os.stat(CRED).st_mode & 0o077:
|
|
sys.exit(f"✗ {CRED} es legible por otros. `chmod 600` y repite.")
|
|
with open(CRED) as f:
|
|
sa = json.load(f)
|
|
for k in ("client_email", "private_key", "token_uri"):
|
|
if k not in sa:
|
|
sys.exit(f"✗ La credencial no parece de cuenta de servicio: falta '{k}'")
|
|
return sa
|
|
|
|
|
|
def token(sa):
|
|
"""Firma el JWT y lo canjea por un access token."""
|
|
import jwt as pyjwt
|
|
import requests
|
|
ahora = int(dt.datetime.now(dt.timezone.utc).timestamp())
|
|
afirmacion = pyjwt.encode(
|
|
{"iss": sa["client_email"], "scope": SCOPE, "aud": TOKEN_URL,
|
|
"iat": ahora, "exp": ahora + 3600},
|
|
sa["private_key"], algorithm="RS256",
|
|
)
|
|
r = requests.post(TOKEN_URL, timeout=30, data={
|
|
"grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
|
|
"assertion": afirmacion,
|
|
})
|
|
if r.status_code != 200:
|
|
sys.exit(f"✗ Google rechazó la credencial ({r.status_code}): {r.text[:300]}")
|
|
return r.json()["access_token"]
|
|
|
|
|
|
def api(tok, ruta, cuerpo=None):
|
|
import requests
|
|
url = f"{API}/{ruta}"
|
|
if cuerpo is None:
|
|
r = requests.get(url, headers={"Authorization": f"Bearer {tok}"}, timeout=60)
|
|
else:
|
|
r = requests.post(url, headers={"Authorization": f"Bearer {tok}"},
|
|
json=cuerpo, timeout=120)
|
|
if r.status_code == 403:
|
|
sys.exit("✗ 403: la cuenta de servicio no tiene permiso sobre esa propiedad.\n"
|
|
" En Search Console → Configuración → Usuarios y permisos,\n"
|
|
" añade el client_email de la credencial como usuario.")
|
|
if r.status_code != 200:
|
|
sys.exit(f"✗ API {r.status_code}: {r.text[:300]}")
|
|
return r.json()
|
|
|
|
|
|
# ──────────────────────────────── almacén ────────────────────────────────
|
|
|
|
ESQUEMA = """
|
|
CREATE TABLE IF NOT EXISTS filas (
|
|
sitio TEXT NOT NULL, dim TEXT NOT NULL, fecha TEXT NOT NULL, clave TEXT NOT NULL,
|
|
clics INTEGER NOT NULL, impresiones INTEGER NOT NULL, posicion REAL NOT NULL,
|
|
PRIMARY KEY (sitio, dim, fecha, clave));
|
|
CREATE INDEX IF NOT EXISTS idx_filas_fecha ON filas(sitio, dim, fecha);
|
|
CREATE TABLE IF NOT EXISTS descargas (
|
|
sitio TEXT, dim TEXT, hasta TEXT, cuando TEXT, filas INTEGER);
|
|
CREATE TABLE IF NOT EXISTS avisos (clave TEXT PRIMARY KEY, huella TEXT, cuando TEXT);
|
|
"""
|
|
|
|
|
|
def bd():
|
|
os.makedirs(os.path.dirname(BD), exist_ok=True)
|
|
c = sqlite3.connect(BD)
|
|
c.executescript(ESQUEMA)
|
|
return c
|
|
|
|
|
|
def ultima_fecha(c, sitio, dim):
|
|
r = c.execute("SELECT max(fecha) FROM filas WHERE sitio=? AND dim=?",
|
|
(sitio, dim)).fetchone()
|
|
return r[0] if r and r[0] else None
|
|
|
|
|
|
# ───────────────────────────────── descarga ─────────────────────────────────
|
|
|
|
def descarga(tok, propiedad, inicio, fin, dims):
|
|
"""Todas las filas del rango, paginando. La API no dice cuántas hay."""
|
|
ruta = f"sites/{urllib.parse.quote(propiedad, safe='')}/searchAnalytics/query"
|
|
filas, desde = [], 0
|
|
while True:
|
|
d = api(tok, ruta, {
|
|
"startDate": inicio, "endDate": fin, "dimensions": dims,
|
|
"rowLimit": 25000, "startRow": desde, "dataState": "final",
|
|
})
|
|
lote = d.get("rows", [])
|
|
filas.extend(lote)
|
|
if len(lote) < 25000:
|
|
return filas
|
|
desde += len(lote)
|
|
|
|
|
|
def cmd_pull(a):
|
|
sa = carga_credencial()
|
|
tok = token(sa)
|
|
c = bd()
|
|
hasta = a.hasta or (dt.date.today() - dt.timedelta(days=RETRASO_DIAS)).isoformat()
|
|
|
|
for sitio in (["en", "es"] if a.site == "todos" else [a.site]):
|
|
prop = SITIOS[sitio]["propiedad"]
|
|
for dim in ("page", "query"):
|
|
desde = a.desde or ultima_fecha(c, sitio, dim)
|
|
if desde:
|
|
# se resolapa un día: el último descargado pudo consolidarse después
|
|
desde = (dt.date.fromisoformat(desde) - dt.timedelta(days=1)).isoformat()
|
|
else:
|
|
desde = (dt.date.today() - dt.timedelta(days=480)).isoformat()
|
|
if desde > hasta:
|
|
print(f" [{sitio}/{dim}] al día (hasta {hasta})")
|
|
continue
|
|
filas = descarga(tok, prop, desde, hasta, ["date", dim])
|
|
c.executemany(
|
|
"INSERT OR REPLACE INTO filas VALUES (?,?,?,?,?,?,?)",
|
|
[(sitio, dim, f["keys"][0], f["keys"][1],
|
|
int(f["clicks"]), int(f["impressions"]), float(f["position"]))
|
|
for f in filas])
|
|
c.execute("INSERT INTO descargas VALUES (?,?,?,?,?)",
|
|
(sitio, dim, hasta, dt.datetime.now().isoformat(timespec="seconds"),
|
|
len(filas)))
|
|
c.commit()
|
|
print(f" [{sitio}/{dim}] {len(filas)} filas {desde} → {hasta}")
|
|
return 0
|
|
|
|
|
|
# ───────────────────────────────── informe ─────────────────────────────────
|
|
|
|
def esperado(pos):
|
|
if pos <= 1:
|
|
return CTR_ESPERADO[1]
|
|
if pos >= 10:
|
|
# más allá del 10 la curva se aplana; extrapolar promete lo que no hay
|
|
return CTR_ESPERADO[10]
|
|
bajo, alto = int(pos), int(pos) + 1
|
|
t = pos - bajo
|
|
return CTR_ESPERADO[bajo] + (CTR_ESPERADO[alto] - CTR_ESPERADO[bajo]) * t
|
|
|
|
|
|
def agrega(c, sitio, dim, desde, hasta):
|
|
"""Suma por clave en la ventana. La posición se pondera por impresiones:
|
|
promediar posiciones a pelo da igual peso a un día de 2 impresiones que a
|
|
uno de 500, y el número sale mentiroso."""
|
|
filas = c.execute(
|
|
"SELECT clave, sum(clics), sum(impresiones), "
|
|
" sum(posicion*impresiones)/nullif(sum(impresiones),0) "
|
|
"FROM filas WHERE sitio=? AND dim=? AND fecha BETWEEN ? AND ? "
|
|
"GROUP BY clave", (sitio, dim, desde, hasta)).fetchall()
|
|
return {k: {"clics": cl, "imp": im, "pos": po or 0.0} for k, cl, im, po in filas}
|
|
|
|
|
|
def totales(d):
|
|
cl = sum(v["clics"] for v in d.values())
|
|
im = sum(v["imp"] for v in d.values())
|
|
return cl, im, (cl / im * 100 if im else 0.0)
|
|
|
|
|
|
def pos_ponderada(d):
|
|
"""Posición media del sitio, ponderada por impresiones (nunca a pelo)."""
|
|
im = sum(v["imp"] for v in d.values())
|
|
return (sum(v["pos"] * v["imp"] for v in d.values()) / im) if im else 0.0
|
|
|
|
|
|
def desperdiciadas(paginas):
|
|
"""Páginas que rankean y no se pinchan, ordenadas por clics que faltan."""
|
|
out = []
|
|
for u, v in paginas.items():
|
|
if v["imp"] < IMPRESIONES_MIN or v["pos"] > 10.5:
|
|
continue
|
|
ctr = v["clics"] / v["imp"]
|
|
esp = esperado(v["pos"])
|
|
if ctr < esp * FRACCION_ALERTA:
|
|
out.append((u, v, ctr, esp, (esp - ctr) * v["imp"]))
|
|
return sorted(out, key=lambda x: (-x[4], x[0]))
|
|
|
|
|
|
def linea_mov(nombre, act, ant):
|
|
if ant is None:
|
|
return f"{nombre}: {act}"
|
|
d = act - ant
|
|
if isinstance(act, float):
|
|
return f"{nombre}: {act:.2f} ({d:+.2f})"
|
|
return f"{nombre}: {act} ({d:+d})"
|
|
|
|
|
|
def informe_sitio(c, sitio, dias):
|
|
hasta = dt.date.today() - dt.timedelta(days=RETRASO_DIAS)
|
|
desde = hasta - dt.timedelta(days=dias - 1)
|
|
p_ant, p_des = desde - dt.timedelta(days=dias), desde - dt.timedelta(days=1)
|
|
|
|
act = agrega(c, sitio, "page", desde.isoformat(), hasta.isoformat())
|
|
ant = agrega(c, sitio, "page", p_ant.isoformat(), p_des.isoformat())
|
|
if not act:
|
|
return f"[{sitio.upper()}] sin datos en la ventana {desde}…{hasta}", False
|
|
|
|
cl, im, ctr = totales(act)
|
|
cl0, im0, ctr0 = totales(ant) if ant else (None, None, None)
|
|
|
|
# Cobertura: cuántas páginas y cuántas consultas nos ven siquiera. Es la
|
|
# métrica principal, no el CTR — ver el comentario de UMBRAL_MOV.
|
|
q_act = agrega(c, sitio, "query", desde.isoformat(), hasta.isoformat())
|
|
q_ant = agrega(c, sitio, "query", p_ant.isoformat(), p_des.isoformat())
|
|
pos, pos0 = pos_ponderada(act), pos_ponderada(ant) if ant else None
|
|
|
|
L = [f"═══ [{sitio.upper()}] {SITIOS[sitio]['host']} ({desde} → {hasta}, {dias} d)",
|
|
" ── cobertura: a cuánta gente llegamos ──",
|
|
" " + linea_mov("impresiones", im, im0),
|
|
" " + linea_mov("páginas con impresiones", len(act), len(ant) if ant else None),
|
|
" " + linea_mov("consultas distintas", len(q_act), len(q_ant) if q_ant else None),
|
|
# ojo al signo: aquí subir es empeorar (posición 5 → 12 es caer)
|
|
" " + linea_mov("posición media (+ es peor)", round(pos, 1),
|
|
round(pos0, 1) if pos0 else None),
|
|
" ── conversión: cuántos pinchan ──",
|
|
" " + linea_mov("clics", cl, cl0),
|
|
" " + linea_mov("CTR %", round(ctr, 2), round(ctr0, 2) if ctr0 is not None else None)]
|
|
|
|
nuevas = sorted(set(act) - set(ant),
|
|
key=lambda u: (-act[u]["imp"], u))[:5] if ant else []
|
|
if nuevas:
|
|
L.append("\n ── páginas que EMPIEZAN a aparecer ──")
|
|
for u in nuevas:
|
|
L.append(f" {act[u]['imp']:5} imp pos {act[u]['pos']:4.1f} "
|
|
f"{u.rstrip('/').rsplit('/', 1)[-1][:48]}")
|
|
perdidas = sorted(set(ant) - set(act),
|
|
key=lambda u: (-ant[u]["imp"], u))[:5] if ant else []
|
|
if perdidas:
|
|
L.append("\n ── páginas que han DEJADO de aparecer ──")
|
|
for u in perdidas:
|
|
L.append(f" {ant[u]['imp']:5} imp antes {u.rstrip('/').rsplit('/', 1)[-1][:44]}")
|
|
|
|
mal = desperdiciadas(act)
|
|
if mal:
|
|
L.append(f"\n ── posicionan y no se pinchan ({len(mal)}) ──")
|
|
L.append(" clics que faltan si el CTR fuese el normal de su posición")
|
|
L.append(f" «~» = menos de {IMPRESIONES_FIRMES} impresiones: orientativo, no firme")
|
|
for u, v, ctr_r, esp, hueco in mal[:8]:
|
|
slug = u.rstrip("/").rsplit("/", 1)[-1][:44]
|
|
flojo = "~" if v["imp"] < IMPRESIONES_FIRMES else " "
|
|
L.append(f" {flojo}+{hueco:5.0f} clics {v['imp']:5} imp CTR {ctr_r*100:4.2f}% "
|
|
f"(normal {esp*100:4.2f}%) pos {v['pos']:4.1f} {slug}")
|
|
else:
|
|
L.append(f"\n ── ninguna página llega a {IMPRESIONES_MIN} impresiones ──")
|
|
|
|
# Aviso por movimiento de cobertura, no solo por CTR: en un sitio que aún no
|
|
# tiene tráfico, que aparezcan (o desaparezcan) páginas ES la noticia.
|
|
movida = False
|
|
if im0 and im0 >= 50:
|
|
movida = abs(im - im0) / im0 >= UMBRAL_MOV
|
|
if ant:
|
|
movida = movida or abs(len(act) - len(ant)) >= UMBRAL_PAGINAS
|
|
|
|
sin_clic = [(k, v) for k, v in q_act.items()
|
|
if v["clics"] == 0 and v["imp"] >= 10 and v["pos"] <= 15]
|
|
if sin_clic:
|
|
L.append(f"\n ── consultas donde salimos y no nos pinchan ({len(sin_clic)}) ──")
|
|
for k, v in sorted(sin_clic, key=lambda x: (-x[1]["imp"], x[0]))[:6]:
|
|
L.append(f" {v['imp']:5} imp pos {v['pos']:4.1f} {k[:58]}")
|
|
|
|
return "\n".join(L), bool(mal) or movida
|
|
|
|
|
|
def cmd_informe(a):
|
|
c = bd()
|
|
partes, hallazgos = [], False
|
|
for sitio in (["en", "es"] if a.site == "todos" else [a.site]):
|
|
t, h = informe_sitio(c, sitio, a.dias)
|
|
partes.append(t)
|
|
hallazgos = hallazgos or h
|
|
texto = "\n\n".join(partes)
|
|
print(texto)
|
|
|
|
if a.telegram:
|
|
import hashlib
|
|
huella = hashlib.sha256(texto.encode()).hexdigest()[:16]
|
|
prev = c.execute("SELECT huella FROM avisos WHERE clave='informe'").fetchone()
|
|
if not hallazgos:
|
|
print("\n(sin hallazgos: no aviso)")
|
|
elif prev and prev[0] == huella:
|
|
print("\n(mismo informe que la última vez: no aviso)")
|
|
else:
|
|
if telegram("📊 Marcador SEO\n\n" + texto):
|
|
c.execute("INSERT OR REPLACE INTO avisos VALUES ('informe',?,?)",
|
|
(huella, dt.datetime.now().isoformat(timespec="seconds")))
|
|
c.commit()
|
|
print("\n(avisado por Telegram)")
|
|
else:
|
|
print("\n✗ Telegram no aceptó el mensaje")
|
|
return 1
|
|
return 0
|
|
|
|
|
|
# ─────────────────────────────── antes/después ───────────────────────────────
|
|
|
|
def cmd_antes_despues(a):
|
|
c = bd()
|
|
corte = dt.date.fromisoformat(a.corte)
|
|
a0, a1 = corte - dt.timedelta(days=a.dias), corte - dt.timedelta(days=1)
|
|
d0 = corte
|
|
d1 = min(corte + dt.timedelta(days=a.dias - 1),
|
|
dt.date.today() - dt.timedelta(days=RETRASO_DIAS))
|
|
dias_reales = (d1 - d0).days + 1
|
|
|
|
for sitio in (["en", "es"] if a.site == "todos" else [a.site]):
|
|
antes = agrega(c, sitio, "page", a0.isoformat(), a1.isoformat())
|
|
desp = agrega(c, sitio, "page", d0.isoformat(), d1.isoformat())
|
|
print(f"\n═══ [{sitio.upper()}] corte {corte}")
|
|
if not antes or not desp:
|
|
print(" sin datos suficientes a un lado del corte "
|
|
f"(antes: {len(antes)} páginas, después: {len(desp)})")
|
|
continue
|
|
if dias_reales != a.dias:
|
|
print(f" ⚠ ventanas DESIGUALES: {a.dias} d antes vs {dias_reales} d después."
|
|
"\n Los totales no son comparables; mira el CTR y la posición.")
|
|
for nombre, f in (("clics", lambda t: t[0]), ("impresiones", lambda t: t[1]),
|
|
("CTR %", lambda t: round(t[2], 2))):
|
|
va, vd = f(totales(antes)), f(totales(desp))
|
|
print(f" {nombre:12} {va:>8} → {vd:>8}")
|
|
|
|
print("\n ── páginas que más movieron su CTR ──")
|
|
comunes = [(u, antes[u], desp[u]) for u in desp if u in antes
|
|
and antes[u]["imp"] >= 50 and desp[u]["imp"] >= 50]
|
|
movidos = []
|
|
for u, va, vd in comunes:
|
|
ca, cd = va["clics"] / va["imp"], vd["clics"] / vd["imp"]
|
|
movidos.append((cd - ca, u, ca, cd, va["pos"], vd["pos"]))
|
|
for delta, u, ca, cd, pa, pd in sorted(movidos, key=lambda x: (-abs(x[0]), x[1]))[:8]:
|
|
slug = u.rstrip("/").rsplit("/", 1)[-1][:40]
|
|
print(f" CTR {ca*100:5.2f}% → {cd*100:5.2f}% ({delta*100:+5.2f}) "
|
|
f"pos {pa:4.1f} → {pd:4.1f} {slug}")
|
|
return 0
|
|
|
|
|
|
# ──────────────────────────────── utilidades ────────────────────────────────
|
|
|
|
def cmd_inspecciona(a):
|
|
"""¿Conoce Google esta URL? Veredicto suyo, no deducción nuestra.
|
|
|
|
Es la única forma de distinguir «posiciona mal» de «no lo ha visto». El
|
|
2026-07-28 dos posts de EN publicados una semana antes salieron con «URL is
|
|
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]}")
|
|
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:
|
|
print(f" ⚠ Google canoniza a: {i['googleCanonical']}")
|
|
print(f" {url}")
|
|
return 1 if malas else 0
|
|
|
|
|
|
def cmd_sitios(a):
|
|
sa = carga_credencial()
|
|
print(f"cuenta de servicio: {sa['client_email']}")
|
|
d = api(token(sa), "sites")
|
|
sitios = d.get("siteEntry", [])
|
|
if not sitios:
|
|
print("\n✗ La cuenta de servicio no ve NINGUNA propiedad.\n"
|
|
" Falta darle permiso en Search Console (Configuración →\n"
|
|
" Usuarios y permisos → Añadir usuario → el correo de arriba).")
|
|
return 1
|
|
print(f"\n{len(sitios)} propiedad(es) visibles:")
|
|
conocidas = {v["propiedad"] for v in SITIOS.values()}
|
|
for s in sitios:
|
|
marca = " ← configurada" if s["siteUrl"] in conocidas else ""
|
|
print(f" {s['permissionLevel']:22} {s['siteUrl']}{marca}")
|
|
faltan = conocidas - {s["siteUrl"] for s in sitios}
|
|
if faltan:
|
|
print("\n⚠ configuradas en SITIOS pero NO visibles: " + ", ".join(sorted(faltan)))
|
|
print(" Corrige SITIOS con el nombre exacto de arriba, o da el permiso.")
|
|
return 1
|
|
return 0
|
|
|
|
|
|
def telegram(texto):
|
|
import base64
|
|
import subprocess
|
|
import urllib.request
|
|
|
|
def secreto(clave):
|
|
r = subprocess.run(
|
|
f"kubectl get secret -n {TG_SECRET[0]} {TG_SECRET[1]} "
|
|
f"-o jsonpath='{{.data.{clave}}}'", shell=True, capture_output=True, text=True)
|
|
return base64.b64decode(r.stdout).decode()
|
|
|
|
datos = urllib.parse.urlencode(
|
|
{"chat_id": secreto("TELEGRAM_CHAT_ID"), "text": texto[:3900]}).encode()
|
|
req = urllib.request.Request(
|
|
f"https://api.telegram.org/bot{secreto('TELEGRAM_BOT_TOKEN')}/sendMessage", data=datos)
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
return json.loads(resp.read()).get("ok", False)
|
|
|
|
|
|
def cmd_comprueba(a):
|
|
"""Canario: el camino real, eslabón por eslabón, sin simular nada."""
|
|
print("1. credencial en disco")
|
|
sa = carga_credencial()
|
|
print(f" ✓ {CRED} (600) — {sa['client_email']}")
|
|
|
|
print("2. firma del JWT y canje por token")
|
|
tok = token(sa)
|
|
print(f" ✓ token de {len(tok)} caracteres")
|
|
|
|
print("3. la API responde y la cuenta ve propiedades")
|
|
sitios = [s["siteUrl"] for s in api(tok, "sites").get("siteEntry", [])]
|
|
if not sitios:
|
|
sys.exit(" ✗ cero propiedades visibles: falta el permiso en Search Console")
|
|
print(f" ✓ {len(sitios)}: {', '.join(sitios)}")
|
|
|
|
print("4. cada sitio configurado devuelve datos")
|
|
fallos = 0
|
|
for k, v in SITIOS.items():
|
|
if v["propiedad"] not in sitios:
|
|
print(f" ✗ [{k}] {v['propiedad']} no visible")
|
|
fallos += 1
|
|
continue
|
|
hasta = (dt.date.today() - dt.timedelta(days=RETRASO_DIAS)).isoformat()
|
|
desde = (dt.date.today() - dt.timedelta(days=33)).isoformat()
|
|
filas = descarga(tok, v["propiedad"], desde, hasta, ["date"])
|
|
imp = sum(int(f["impressions"]) for f in filas)
|
|
print(f" {'✓' if filas else '✗'} [{k}] {len(filas)} días, {imp} impresiones")
|
|
fallos += 0 if filas else 1
|
|
|
|
print("5. almacén")
|
|
c = bd()
|
|
n = c.execute("SELECT count(*) FROM filas").fetchone()[0]
|
|
print(f" ✓ {BD} — {n} filas acumuladas")
|
|
|
|
print("\n" + ("✗ hay eslabones rotos" if fallos else "✓ camino completo"))
|
|
return 1 if fallos else 0
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(description="Marcador de Search Console (solo lectura)")
|
|
sub = ap.add_subparsers(dest="cmd", required=True)
|
|
|
|
p = sub.add_parser("sitios", help="propiedades visibles para la cuenta de servicio")
|
|
p.set_defaults(func=cmd_sitios)
|
|
|
|
p = sub.add_parser("pull", help="descarga y acumula en SQLite")
|
|
p.add_argument("--site", choices=("en", "es", "todos"), default="todos")
|
|
p.add_argument("--desde")
|
|
p.add_argument("--hasta")
|
|
p.set_defaults(func=cmd_pull)
|
|
|
|
p = sub.add_parser("informe", help="el marcador")
|
|
p.add_argument("--site", choices=("en", "es", "todos"), default="todos")
|
|
p.add_argument("--dias", type=int, default=28)
|
|
p.add_argument("--telegram", action="store_true")
|
|
p.set_defaults(func=cmd_informe)
|
|
|
|
p = sub.add_parser("antes-despues", help="misma ventana a los dos lados de una fecha")
|
|
p.add_argument("--corte", required=True)
|
|
p.add_argument("--dias", type=int, default=28)
|
|
p.add_argument("--site", choices=("en", "es", "todos"), default="todos")
|
|
p.set_defaults(func=cmd_antes_despues)
|
|
|
|
p = sub.add_parser("inspecciona", help="¿conoce Google estas URLs?")
|
|
p.add_argument("slugs", nargs="+", help="slugs o URLs completas")
|
|
p.add_argument("--site", choices=("en", "es"), required=True)
|
|
p.set_defaults(func=cmd_inspecciona)
|
|
|
|
p = sub.add_parser("comprueba", help="canario del camino real")
|
|
p.set_defaults(func=cmd_comprueba)
|
|
|
|
a = ap.parse_args()
|
|
return a.func(a) or 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|