107 lines
4.0 KiB
Python
107 lines
4.0 KiB
Python
"""Extractor de Search Analytics de Google Search Console (GSC).
|
|
|
|
Consulta la Search Analytics API de Google Search Console y devuelve las filas
|
|
aplanadas (impresiones, clicks, CTR, posicion) por las dimensiones pedidas.
|
|
Es el extractor principal de datos SEO para alimentar un pipeline hacia
|
|
DuckDB/Postgres.
|
|
"""
|
|
|
|
from typing import Any
|
|
|
|
|
|
def pull_gsc_search_analytics(
|
|
service: object,
|
|
site_url: str,
|
|
start_date: str,
|
|
end_date: str,
|
|
dimensions: list = None,
|
|
row_limit: int = 25000,
|
|
max_total_rows: int = 0,
|
|
search_type: str = "web",
|
|
) -> list:
|
|
"""Extrae datos de Search Analytics de Google Search Console.
|
|
|
|
Llama a ``service.searchanalytics().query(...).execute()`` paginando los
|
|
resultados (la API devuelve como maximo ``row_limit`` filas por request,
|
|
con tope duro de 25000) y aplana cada fila a un dict donde el array ``keys``
|
|
se mapea posicionalmente a los nombres de ``dimensions``.
|
|
|
|
Args:
|
|
service: objeto service autenticado de la API de Search Console
|
|
(el que devuelve ``gsc_auth`` del registry). Se inyecta ya
|
|
construido; esta funcion NO lo crea.
|
|
site_url: propiedad de Search Console. ``sc-domain:ejemplo.com`` para
|
|
propiedad de dominio, o la URL completa ``https://ejemplo.com/``
|
|
para propiedad de prefijo.
|
|
start_date: fecha inicial inclusiva en formato ``YYYY-MM-DD``.
|
|
end_date: fecha final inclusiva en formato ``YYYY-MM-DD``. La API tiene
|
|
~2-3 dias de lag; el caller deberia pedir hasta hoy-3.
|
|
dimensions: lista de dimensiones a desglosar. Por defecto
|
|
``["query", "page"]``. Otras validas: ``date``, ``country``,
|
|
``device``, ``searchAppearance``.
|
|
row_limit: filas por request (1..25000). Tambien el tamaño de paso de
|
|
la paginacion. Por defecto 25000.
|
|
max_total_rows: tope total de filas acumuladas. ``0`` = sin tope (trae
|
|
todas las paginas disponibles).
|
|
search_type: tipo de busqueda. ``"web"`` | ``"image"`` | ``"video"`` |
|
|
``"news"`` | ``"discover"`` | ``"googleNews"``. Va en el body como
|
|
``"type"``.
|
|
|
|
Returns:
|
|
Lista de dicts aplanados. Cada dict tiene una clave por cada dimension
|
|
(con su nombre real, ej. ``query``, ``page``) mas ``clicks``,
|
|
``impressions``, ``ctr`` y ``position``. Lista vacia si la API no
|
|
devuelve filas.
|
|
|
|
Raises:
|
|
Exception: cualquier error de la API HTTP de Google se propaga
|
|
(autenticacion, permisos sobre la propiedad, rate limit, etc.).
|
|
"""
|
|
dims = list(dimensions) if dimensions else ["query", "page"]
|
|
# Clamp del row_limit al rango valido de la API (1..25000).
|
|
page_size = max(1, min(int(row_limit), 25000))
|
|
|
|
results: list = []
|
|
start_row = 0
|
|
|
|
while True:
|
|
body: dict[str, Any] = {
|
|
"startDate": start_date,
|
|
"endDate": end_date,
|
|
"dimensions": dims,
|
|
"type": search_type,
|
|
"rowLimit": page_size,
|
|
"startRow": start_row,
|
|
}
|
|
|
|
response = (
|
|
service.searchanalytics().query(siteUrl=site_url, body=body).execute()
|
|
)
|
|
|
|
rows = response.get("rows") if isinstance(response, dict) else None
|
|
if not rows:
|
|
# rows ausente o vacio => no hay mas datos.
|
|
break
|
|
|
|
for row in rows:
|
|
keys = row.get("keys", [])
|
|
flat: dict[str, Any] = {}
|
|
for i, dim in enumerate(dims):
|
|
flat[dim] = keys[i] if i < len(keys) else None
|
|
flat["clicks"] = row.get("clicks")
|
|
flat["impressions"] = row.get("impressions")
|
|
flat["ctr"] = row.get("ctr")
|
|
flat["position"] = row.get("position")
|
|
results.append(flat)
|
|
|
|
if max_total_rows > 0 and len(results) >= max_total_rows:
|
|
return results[:max_total_rows]
|
|
|
|
# Si la pagina trajo menos filas que el tope, no hay mas paginas.
|
|
if len(rows) < page_size:
|
|
break
|
|
|
|
start_row += page_size
|
|
|
|
return results
|