ChemaVXandClaude Opus 5
66540c12e6
`page` y `query` se descargaban por separado, así que no había forma de saber
de qué búsquedas salían las impresiones de una URL. Con eso faltando, el dato
más llamativo del marcador (uss-nimitz: 1.254 impresiones en posición 6,7 y
CERO clics) admitía tres explicaciones incompatibles y no se podía elegir.
Ya se puede. El 91% de las impresiones del EN son ANÓNIMAS: consultas tan raras
que Google no las nombra. En el Nimitz, de 1.254 impresiones solo 29 tienen
consulta conocida. No es un problema de titular — reescribirlo no tocaría el
99% del volumen. En el ES el anónimo es el 66%, y el ES convierte 3x mejor.
Decisiones:
- tabla filas_pq aparte, no un `dim` más de `filas`: son DOS dimensiones y
con la clave concatenada cualquier agregado que olvidase filtrar por dim
contaría el mismo clic dos veces
- backfill troceado en ventanas de 90 días: con tres dimensiones las filas
se multiplican y un tirón de 16 meses puede pasarse del tope de paginación
- el informe enseña el HUECO lo primero, no lo esconde. Una página cuyo
volumen es casi todo anónimo no se arregla por el titular, y el informe lo
dice en voz alta al pasar del 70%
Verificado: filas_pq suma exactamente lo mismo que dim='query' en la misma
ventana (EN 405 imp, ES 144), que es la comprobación de que no se cuelan ni se
pierden filas por el camino.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
766 lines
35 KiB
Python
766 lines
35 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 consultas [--site en|es] [--pagina SLUG] [--dias N] [--min N]
|
||
Página × consulta: de qué búsquedas salen las impresiones de cada URL.
|
||
Responde la pregunta que las otras vistas NO pueden responder, porque
|
||
`page` y `query` por separado no se cruzan: si una página tiene 2.449
|
||
impresiones en posición 6 y cero clics, esto dice si es que sale en
|
||
consultas sin intención de clic o es que el titular no vende.
|
||
|
||
Lo primero que imprime es el HUECO: impresiones de la página menos las
|
||
que Google atribuye a alguna consulta. Ver el gotcha de la anonimización
|
||
más abajo — ese hueco no es un fallo, es el dato.
|
||
|
||
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.
|
||
* ⚠️ Google ANONIMIZA las consultas poco frecuentes: no las devuelve en
|
||
ninguna vista que incluya la dimensión `query`. Consecuencia práctica: la
|
||
suma por `page` NO cuadra con la suma por `page`×`query`, y la diferencia
|
||
puede ser la mayor parte. No es un error de descarga ni de la ventana; es
|
||
volumen de cola larga que Google se niega a nombrar. Por eso `consultas`
|
||
enseña el hueco en primer lugar en vez de esconderlo: una página cuyas
|
||
impresiones son casi todas anónimas no tiene un problema de titular.
|
||
* 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);
|
||
-- Página × consulta. Tabla aparte y no un `dim` más de `filas` a propósito:
|
||
-- son DOS dimensiones, y meterlas ahí con la clave concatenada haría que
|
||
-- cualquier agregado que olvidase filtrar por dim contase el mismo clic dos
|
||
-- veces. Separadas, ese error es imposible.
|
||
CREATE TABLE IF NOT EXISTS filas_pq (
|
||
sitio TEXT NOT NULL, fecha TEXT NOT NULL, pagina TEXT NOT NULL, consulta TEXT NOT NULL,
|
||
clics INTEGER NOT NULL, impresiones INTEGER NOT NULL, posicion REAL NOT NULL,
|
||
PRIMARY KEY (sitio, fecha, pagina, consulta));
|
||
CREATE INDEX IF NOT EXISTS idx_pq_fecha ON filas_pq(sitio, 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
|
||
|
||
|
||
def ultima_fecha_pq(c, sitio):
|
||
r = c.execute("SELECT max(fecha) FROM filas_pq WHERE sitio=?", (sitio,)).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}")
|
||
|
||
# ── página × consulta ──
|
||
# Se trocea en ventanas de 90 días. Con tres dimensiones las filas se
|
||
# multiplican y un backfill de 16 meses de un tirón puede pasarse del
|
||
# tope de paginación de la API; troceado no depende de eso, y además va
|
||
# informando en vez de quedarse mudo cinco minutos.
|
||
desde = a.desde or ultima_fecha_pq(c, sitio)
|
||
if desde:
|
||
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}/page×query] al día (hasta {hasta})")
|
||
continue
|
||
d0, dfin, total = dt.date.fromisoformat(desde), dt.date.fromisoformat(hasta), 0
|
||
while d0 <= dfin:
|
||
d1 = min(d0 + dt.timedelta(days=89), dfin)
|
||
filas = descarga(tok, prop, d0.isoformat(), d1.isoformat(),
|
||
["date", "page", "query"])
|
||
c.executemany(
|
||
"INSERT OR REPLACE INTO filas_pq VALUES (?,?,?,?,?,?,?)",
|
||
[(sitio, f["keys"][0], f["keys"][1], f["keys"][2],
|
||
int(f["clicks"]), int(f["impressions"]), float(f["position"]))
|
||
for f in filas])
|
||
c.commit()
|
||
total += len(filas)
|
||
if d1 < dfin:
|
||
print(f" …{d0} → {d1}: {len(filas)} filas")
|
||
d0 = d1 + dt.timedelta(days=1)
|
||
c.execute("INSERT INTO descargas VALUES (?,?,?,?,?)",
|
||
(sitio, "page+query", hasta,
|
||
dt.datetime.now().isoformat(timespec="seconds"), total))
|
||
c.commit()
|
||
print(f" [{sitio}/page×query] {total} 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
|
||
|
||
|
||
# ─────────────────────────── página × consulta ───────────────────────────
|
||
|
||
def agrega_pq(c, sitio, desde, hasta, filtro=None):
|
||
"""[(pagina, consulta, clics, imp, pos)] sumado en la ventana."""
|
||
sql = ("SELECT pagina, consulta, sum(clics), sum(impresiones), "
|
||
" sum(posicion*impresiones)/nullif(sum(impresiones),0) "
|
||
"FROM filas_pq WHERE sitio=? AND fecha BETWEEN ? AND ? ")
|
||
par = [sitio, desde, hasta]
|
||
if filtro:
|
||
sql += "AND pagina LIKE ? "
|
||
par.append(f"%{filtro}%")
|
||
return c.execute(sql + "GROUP BY pagina, consulta", par).fetchall()
|
||
|
||
|
||
def consultas_sitio(c, sitio, dias, filtro, minimo):
|
||
hasta = dt.date.today() - dt.timedelta(days=RETRASO_DIAS)
|
||
desde = hasta - dt.timedelta(days=dias - 1)
|
||
filas = agrega_pq(c, sitio, desde.isoformat(), hasta.isoformat(), filtro)
|
||
paginas = agrega(c, sitio, "page", desde.isoformat(), hasta.isoformat())
|
||
if filtro:
|
||
paginas = {u: v for u, v in paginas.items() if filtro in u}
|
||
|
||
L = [f"═══ [{sitio.upper()}] {SITIOS[sitio]['host']} ({desde} → {hasta}, {dias} d)"]
|
||
if not paginas:
|
||
return "\n".join(L + [" sin datos de páginas en la ventana"])
|
||
if not filas:
|
||
return "\n".join(L + [
|
||
" ⚠ no hay datos de página×consulta en esta ventana.",
|
||
" Si nunca has hecho `pull` desde que existe esta vista, hazlo:",
|
||
" la tabla filas_pq se rellena aparte y arranca vacía."])
|
||
|
||
# Por página: qué parte de sus impresiones tiene nombre de consulta.
|
||
por_pag = {}
|
||
for pag, cons, cl, im, pos in filas:
|
||
d = por_pag.setdefault(pag, {"clics": 0, "imp": 0, "q": []})
|
||
d["clics"] += cl
|
||
d["imp"] += im
|
||
d["q"].append((cons, cl, im, pos or 0.0))
|
||
|
||
im_tot = sum(v["imp"] for v in paginas.values())
|
||
im_nom = sum(v["imp"] for v in por_pag.values())
|
||
anon = im_tot - im_nom
|
||
L += [" ── de dónde salen las impresiones ──",
|
||
f" impresiones de las páginas {im_tot:7}",
|
||
f" atribuidas a una consulta {im_nom:7} ({im_nom*100//max(im_tot,1):3}%)",
|
||
f" ANÓNIMAS (Google no las nombra) {anon:7} ({anon*100//max(im_tot,1):3}%)"]
|
||
if im_tot and anon / im_tot > 0.7:
|
||
L.append(" ⚠ la mayor parte del volumen es cola larga sin nombre: en esas")
|
||
L.append(" páginas el CTR bajo NO se arregla reescribiendo el titular.")
|
||
|
||
L.append("\n ── por página ──")
|
||
L.append(" imp c/nombre anón pos clics página")
|
||
orden = sorted(paginas.items(), key=lambda kv: (-kv[1]["imp"], kv[0]))
|
||
for u, v in orden:
|
||
if v["imp"] < minimo:
|
||
continue
|
||
nom = por_pag.get(u, {}).get("imp", 0)
|
||
L.append(f" {v['imp']:5} {nom:7} {v['imp']-nom:5} {v['pos']:4.1f} "
|
||
f"{v['clics']:5} {u.rstrip('/').rsplit('/', 1)[-1][:44]}")
|
||
|
||
# Las consultas con nombre, que es lo único sobre lo que se puede actuar.
|
||
cuantas = 25 if filtro else 12
|
||
todas = sorted(((c_, cl, im, po, pag) for pag, d in por_pag.items()
|
||
for c_, cl, im, po in d["q"]), key=lambda x: (-x[2], x[0]))
|
||
L.append(f"\n ── consultas con nombre (top {cuantas} de {len(todas)}) ──")
|
||
for c_, cl, im, po, pag in todas[:cuantas]:
|
||
L.append(f" {im:5} imp {cl:3} clics pos {po:4.1f} {c_[:52]}"
|
||
+ ("" if filtro else f" → {pag.rstrip('/').rsplit('/', 1)[-1][:26]}"))
|
||
return "\n".join(L)
|
||
|
||
|
||
def cmd_consultas(a):
|
||
c = bd()
|
||
for sitio in (["en", "es"] if a.site == "todos" else [a.site]):
|
||
print(consultas_sitio(c, sitio, a.dias, a.pagina, a.min))
|
||
print()
|
||
return 0
|
||
|
||
|
||
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("consultas", help="página × consulta: de qué búsquedas sale cada URL")
|
||
p.add_argument("--site", choices=("en", "es", "todos"), default="todos")
|
||
p.add_argument("--pagina", help="filtra por subcadena del slug/URL")
|
||
p.add_argument("--dias", type=int, default=28)
|
||
p.add_argument("--min", type=int, default=10,
|
||
help="impresiones mínimas para listar una página")
|
||
p.set_defaults(func=cmd_consultas)
|
||
|
||
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())
|