ChemaVX
33b24f33c7
Estreno con credencial. Dos cosas que solo se ven con la API delante: La propiedad ES no es de dominio sino de prefijo de URL (https://zonadeexclusion.com/), así que sc-domain:zonadeexclusion.com no existía y el canario lo cazó en el paso 4. Queda anotado que ese tipo de propiedad no ve www ni http. Y el histórico no es de 16 meses: las dos propiedades se dieron de alta en mayo de 2026 y el dato más antiguo es del 2026-05-09. El manual prometía comparar contra el año pasado y eso no va a existir nunca.
502 lines
21 KiB
Python
502 lines
21 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: totales, movimiento contra el periodo anterior y la lista
|
|
de páginas que POSICIONAN BIEN Y NO SE PINCHAN, que es donde está el
|
|
dinero sin escribir nada nuevo.
|
|
|
|
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 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
|
|
IMPRESIONES_MIN = 150 # por debajo, el CTR es ruido estadístico
|
|
FRACCION_ALERTA = 0.45 # se avisa si el CTR no llega a esta parte del esperado
|
|
|
|
# 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 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])
|
|
|
|
|
|
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)
|
|
L = [f"═══ [{sitio.upper()}] {SITIOS[sitio]['host']} ({desde} → {hasta}, {dias} d)",
|
|
" " + linea_mov("clics", cl, cl0),
|
|
" " + linea_mov("impresiones", im, im0),
|
|
" " + linea_mov("CTR %", round(ctr, 2), round(ctr0, 2) if ctr0 is not None else None)]
|
|
|
|
mal = desperdiciadas(act)
|
|
if mal:
|
|
L.append(f"\n ── posicionan y no se pinchan ({len(mal)}) ──")
|
|
L.append(" cada línea: clics que faltan si el CTR fuese el normal de su posición")
|
|
for u, v, ctr_r, esp, hueco in mal[:8]:
|
|
slug = u.rstrip("/").rsplit("/", 1)[-1][:44]
|
|
L.append(f" +{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("\n ── sin páginas desperdiciadas por encima del umbral ──")
|
|
|
|
consultas = agrega(c, sitio, "query", desde.isoformat(), hasta.isoformat())
|
|
sin_clic = [(k, v) for k, v in consultas.items()
|
|
if v["clics"] == 0 and v["imp"] >= 30 and v["pos"] <= 15]
|
|
if sin_clic:
|
|
L.append(f"\n ── consultas con impresiones y CERO clics ({len(sin_clic)}) ──")
|
|
for k, v in sorted(sin_clic, key=lambda x: -x[1]["imp"])[:6]:
|
|
L.append(f" {v['imp']:5} imp pos {v['pos']:4.1f} {k[:58]}")
|
|
|
|
return "\n".join(L), bool(mal)
|
|
|
|
|
|
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]))[: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_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("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())
|