""" Traer el rendimiento de Google y contestar las preguntas que la interfaz hace. Tres ideas gobiernan este archivo, y las tres son decisiones tomadas y no detalles de implementación: 1. **«Hoy» no existe.** Search Analytics publica con dos o tres días de atraso, así que el presente del módulo es `last_closed_date` y no la fecha del calendario. Todo —el rango de la próxima corrida y los cuatro comparadores— se ancla ahí. 2. **Se pregunta dos veces por día.** Una sin la dimensión de página, para los totales; otra con ella, para el reparto. Las impresiones de la segunda no se suman: varias páginas pueden venir de una sola búsqueda. 3. **No se inventan ceros.** Un día que Google no devuelve es un día sin dato, no un día en cero. La diferencia importa cuando se promedia. """ import calendar import logging from dataclasses import dataclass from datetime import date, timedelta from django.db import transaction from django.utils import timezone from apps.core.dates import today from apps.credentials.models import Module from apps.credentials.services import CredentialNotReady, resolve_shared_credential from apps.gsc.client import SEARCH_ANALYTICS_ROW_LIMIT, SearchConsoleClient from apps.gsc.errors import PERMANENT_CODES, GoogleCallError, GoogleErrorCode from apps.seo.models import ( CannibalizationSeverity, Engagement, KeywordDaily, KeywordPageDaily, KeywordSource, ModuleStatus, PageRotation, PositionBand, RunKind, RunOrigin, RunState, SyncRun, SyncState, TrackedKeyword, ) logger = logging.getLogger(__name__) #: Cuántos días atrás está el último dato que Google da por cerrado. #: #: La documentación promete dos o tres; se toman tres para caer del lado seguro. #: El margen es lo que permite no reconsultar: si se pidiera el día de ayer #: habría que volver a pedirlo después, porque todavía se está completando. LATENCY_DAYS = 3 #: Impresiones mínimas que necesita un extremo para que su variación se calcule. #: #: La posición de Search Console es un promedio, y con dos o tres impresiones ese #: promedio describe a quien buscó y no la posición del sitio: estando en el #: puesto 7, una sola impresión en el 30 mueve el número once puestos con dos #: impresiones y dos puestos con diez. Diez es donde el promedio empieza a #: hablar del sitio. Es un valor elegido y ajustable, no una ley. MIN_IMPRESSIONS = 10 #: Las cuatro ventanas de comparación, en días. #: #: La de un día compara dos días sueltos, que es lo que se le pide: avisar de lo #: que acaba de pasar. Las demás comparan **promedios de rango contra rango**, #: porque su trabajo es la tendencia y un día atípico no puede decidirla: todas #: las columnas parten del mismo `last_closed_date`, así que un solo día raro #: torcería las cuatro a la vez. DELTA_WINDOWS: tuple[int, ...] = (1, 3, 7, 15) #: Meses que el modal ofrece importar. El tope no lo pone el peso —doce meses son #: unas decenas de miles de filas— sino que más atrás empieza otro sitio. BACKFILL_MONTHS: tuple[int, ...] = (1, 3, 12) TOTALS_DIMENSIONS = ['date', 'query'] PAGES_DIMENSIONS = ['date', 'query', 'page'] #: El tramo donde una consulta está a un empujón de la primera página. #: #: Del 11 al 20 es la segunda página de Google, adonde casi nadie llega. Subir #: tres puestos ahí vale mucho más que subir tres en el puesto 60. NEAR_FIRST_PAGE = (10, 20) #: Hasta dónde llega la primera página. Diez resultados, desde que existe Google. FIRST_PAGE_LAST_POSITION = 10 #: Dónde termina el tramo de arriba. Los tres primeros se llevan la mayoría de #: los clics, y por eso se separan del resto de la primera página. TOP_POSITIONS = 3 #: Impresiones desde las cuales «apareciste y no te clickearon» es un dato y no #: una casualidad. Es el mismo piso que usan las variaciones, por el mismo #: motivo: por debajo, la muestra no sostiene ninguna afirmación. REACH_MINIMUM_IMPRESSIONS = MIN_IMPRESSIONS def classify_band(position: float) -> str: """ En qué tramo del listado cae una posición promedio. **Es el único lugar donde se corta.** El indicador del inicio y el filtro de la tabla salen los dos de acá, así que no pueden discrepar: antes el indicador cortaba con una comparación suelta y el filtro habría cortado con otra, escrita en TypeScript, y la primera vez que alguien tocara una de las dos la tarjeta y la tabla habrían dejado de coincidir sin que nada fallara. Los cortes son **por arriba**: la posición 10,0 es primera página y la 10,1 ya es la segunda. Es como los cuenta Google y como los contaba `near_first_page`, que usaba `10 < position <= 20`. """ if position <= TOP_POSITIONS: return PositionBand.TOP_3 if position <= FIRST_PAGE_LAST_POSITION: return PositionBand.FIRST_PAGE if position <= NEAR_FIRST_PAGE[1]: return PositionBand.NEAR_FIRST_PAGE return PositionBand.BEYOND def classify_engagement(clicks: int, impressions: int) -> str: """ Si la consulta trajo visitas, y si no, si apareció lo bastante como para que eso signifique algo. Mismo criterio que `classify_band`: acá se corta una sola vez. El piso de impresiones es parte de la definición y no un adorno —sin él, «sin clics» se llena de consultas que aparecieron tres veces— y por eso vive adentro de la clasificación en vez de en quien la consume. """ if clicks > 0: return Engagement.WITH_CLICKS if impressions >= REACH_MINIMUM_IMPRESSIONS: return Engagement.NO_CLICKS return Engagement.LOW_REACH class BackfillNotAllowed(Exception): """Se pidió una importación que las reglas del módulo no admiten.""" # --- El calendario del módulo --------------------------------------------- def last_available_date() -> date: """El día más reciente que tiene sentido pedirle a Google.""" return today() - timedelta(days=LATENCY_DAYS) def months_before(day: date, months: int) -> date: """ El mismo día, tantos meses atrás. Se calcula a mano en vez de restar treinta días por mes: un backfill de doce meses acumularía cinco días de error, y el día de arranque de la historia es justamente lo que la pantalla promete. """ month = day.month - months year = day.year while month <= 0: month += 12 year -= 1 # El 31 no existe en todos los meses. Se recorta al último día real en vez de # desbordar al mes siguiente. last_day = calendar.monthrange(year, month)[1] return date(year, month, min(day.day, last_day)) def sync_range(state: SyncState) -> tuple[date, date] | None: """ Qué le toca traer a la próxima corrida diaria. Sale del **estado guardado** y no del calendario, y ahí está lo que importa: si el trabajador no corre un día —un despliegue, Redis caído, un error de red— la corrida siguiente trae sola lo que faltó. Un «pedí el día de hace tres» escrito contra la fecha de hoy dejaría ese día como un agujero permanente en la serie. Devuelve `None` cuando no hay nada que hacer: o nunca se importó historia —y entonces lo que corresponde es un backfill, no seguir desde la nada— o el módulo ya está al día. """ if state.last_closed_date is None: return None start = state.last_closed_date + timedelta(days=1) end = last_available_date() if start > end: return None return start, end # --- La conversación con Google ------------------------------------------- def _client() -> SearchConsoleClient: return SearchConsoleClient(credential=resolve_shared_credential(Module.SEARCH_CONSOLE)) def _fetch(client, property_uri: str, start: date, end: date, dimensions: list[str]) -> list[dict]: """ Todas las filas del rango, paginando hasta agotarlas. Se avanza con `startRow` hasta que Google devuelve menos filas que el tope. Una respuesta exacta al tope no significa que no haya más: significa que hay que preguntar otra vez. """ rows: list[dict] = [] start_row = 0 while True: payload = client.search_analytics( property_uri=property_uri, start_date=start.isoformat(), end_date=end.isoformat(), dimensions=dimensions, start_row=start_row, ) batch = payload.get('rows', []) rows.extend(batch) if len(batch) < SEARCH_ANALYTICS_ROW_LIMIT: return rows start_row += len(batch) def check_module(domain) -> tuple[str, str]: """ Comprueba que el módulo puede trabajar sobre este sitio. Devuelve el estado del módulo y, cuando lo hay, el `GoogleErrorCode` que lo explica. **No inventa un vocabulario de errores**: los ocho códigos de `apps.gsc.errors` ya distinguen lo que hace falta —incluido `API_NOT_ENABLED`, que es de los fallos más frecuentes al conectar y que un vocabulario propio habría metido en la bolsa de «error temporal», que es lo contrario de lo que es—. Lo único propio son los cinco estados de sincronización, que describen otra cosa: no una llamada, sino en qué punto está el módulo. """ end = last_available_date() start = end - timedelta(days=1) try: payload = _fetch(_client(), domain.property_uri, start, end, TOTALS_DIMENSIONS) except CredentialNotReady: return ModuleStatus.ACTION_REQUIRED, GoogleErrorCode.INVALID_KEY except GoogleCallError as error: # Reintentar un permiso denegado gasta tiempo para volver al mismo lugar; # reintentar un 503 sirve. Esa partición ya está escrita. status = ( ModuleStatus.ACTION_REQUIRED if error.code in PERMANENT_CODES else ModuleStatus.ERROR ) return status, error.code # Sin filas la integración está sana y el sitio no aparece por ninguna # búsqueda en ese par de días. No es un error, y confundirlo con uno mandaría # a revisar permisos que están bien. return (ModuleStatus.READY if payload else ModuleStatus.NO_DATA), '' # --- Guardar lo que llegó ------------------------------------------------- def _row_date(row: dict) -> date: """ La fecha tal como la nombró Google. Search Console fecha en horario del Pacífico. **No se traslada** a la zona de la cuenta: correrla un día haría que nuestra serie no coincida con la que la persona ve en Search Console, y la comparación entre las dos es lo único con lo que se puede comprobar que este módulo dice la verdad. """ return date.fromisoformat(row['keys'][0]) def _store_totals(domain, rows: list[dict]) -> int: stamp = timezone.now() records = [ KeywordDaily( domain=domain, date=_row_date(row), query=row['keys'][1][:500], clicks=int(row.get('clicks', 0)), impressions=int(row.get('impressions', 0)), ctr=float(row.get('ctr', 0.0)), position=float(row.get('position', 0.0)), updated_at=stamp, ) for row in rows ] KeywordDaily.objects.bulk_create( records, update_conflicts=True, update_fields=['clicks', 'impressions', 'ctr', 'position', 'updated_at'], unique_fields=['domain', 'date', 'query'], ) return len(records) def _store_pages(domain, rows: list[dict]) -> int: stamp = timezone.now() records = [ KeywordPageDaily( domain=domain, date=_row_date(row), query=row['keys'][1][:500], page=row['keys'][2][:2000], clicks=int(row.get('clicks', 0)), impressions=int(row.get('impressions', 0)), ctr=float(row.get('ctr', 0.0)), position=float(row.get('position', 0.0)), updated_at=stamp, ) for row in rows ] KeywordPageDaily.objects.bulk_create( records, update_conflicts=True, update_fields=['clicks', 'impressions', 'ctr', 'position', 'updated_at'], unique_fields=['domain', 'date', 'query', 'page'], ) return len(records) def import_range(domain, start: date, end: date) -> dict: """ Trae un rango y lo guarda en las dos tablas. **Dos consultas y no una.** La de totales va sin la dimensión de página porque es el único modo de que las impresiones sean las que Google cuenta: cuando varias páginas del sitio aparecen en la misma búsqueda, cada una registra su impresión pero la búsqueda fue una sola. Medido contra un sitio real, sumar por página inflaba una de cada cuatro claves y en el peor caso multiplicaba por seis. Es idempotente: las dos escrituras son `upsert` sobre su clave natural, así que repetir un rango corrige lo que haya cambiado y no duplica nada. """ client = _client() totals = _fetch(client, domain.property_uri, start, end, TOTALS_DIMENSIONS) pages = _fetch(client, domain.property_uri, start, end, PAGES_DIMENSIONS) with transaction.atomic(): stored_totals = _store_totals(domain, totals) stored_pages = _store_pages(domain, pages) return { 'api_requests': 2, 'keyword_rows': stored_totals, 'keyword_page_rows': stored_pages, } # --- Las corridas --------------------------------------------------------- def state_for(domain) -> SyncState: state, _created = SyncState.objects.get_or_create(domain=domain) return state def queue_backfill(domain, months: int, *, origin: str = RunOrigin.MANUAL) -> SyncRun: """ Registra la importación de historia y la deja lista para el trabajador. No la ejecuta: encolarla es lo que permite que la pantalla conteste enseguida y que quien la pidió pueda seguir usando el resto de la plataforma, que es lo que la especificación pide durante la primera importación. """ if months not in BACKFILL_MONTHS: raise BackfillNotAllowed(f'{months} is not one of {BACKFILL_MONTHS}.') end = last_available_date() start = months_before(end, months) run = SyncRun.objects.create( domain=domain, kind=RunKind.BACKFILL, origin=origin, requested_start=start, requested_end=end, total_items=(end - start).days + 1, ) state = state_for(domain) state.status = ModuleStatus.SYNCING state.save(update_fields=['status', 'updated_at']) return run def queue_daily_sync(domain) -> SyncRun | None: """La corrida de todos los días. Devuelve `None` cuando no hay nada que traer.""" window = sync_range(state_for(domain)) if window is None: return None start, end = window return SyncRun.objects.create( domain=domain, kind=RunKind.DAILY_SYNC, origin=RunOrigin.SCHEDULED, requested_start=start, requested_end=end, total_items=(end - start).days + 1, ) def execute(run: SyncRun) -> SyncRun: """ Corre una importación ya registrada, sea de historia o del día. Las dos hacen exactamente lo mismo con distinto rango, y por eso hay una sola función: separarlas garantizaría que algún día difieran sin que nadie lo haya decidido. """ run.state = RunState.RUNNING run.started_at = timezone.now() run.save(update_fields=['state', 'started_at', 'updated_at']) state = state_for(run.domain) try: summary = import_range(run.domain, run.requested_start, run.requested_end) except (GoogleCallError, CredentialNotReady) as error: code = getattr(error, 'code', GoogleErrorCode.INVALID_KEY) run.state = RunState.FAILED run.failed_items = run.total_items run.error_code = code run.finished_at = timezone.now() run.save( update_fields=['state', 'failed_items', 'error_code', 'finished_at', 'updated_at'] ) # Una corrida fallida **no destruye el último conjunto válido**: se marca # el estado y se deja la serie como estaba. Vaciar el módulo ante un 503 # convertiría un problema pasajero en una pantalla sin nada. state.status = ( ModuleStatus.ACTION_REQUIRED if code in PERMANENT_CODES else ModuleStatus.ERROR ) state.last_error_code = code state.save(update_fields=['status', 'last_error_code', 'updated_at']) raise run.state = RunState.COMPLETED run.processed_items = run.total_items run.summary = summary run.finished_at = timezone.now() run.save( update_fields=['state', 'processed_items', 'summary', 'finished_at', 'updated_at'] ) _advance_state(state, run, summary) return run def _advance_state(state: SyncState, run: SyncRun, summary: dict) -> None: """ Deja escrito hasta dónde llega la historia después de una corrida. `last_closed_date` sólo avanza si algo llegó. Un rango vacío no es un rango cubierto: moverlo igual saltearía esos días para siempre, porque la corrida siguiente arranca justo después de esta marca. """ fields = ['status', 'last_sync_at', 'last_error_code', 'updated_at'] if summary['keyword_rows'] or summary['keyword_page_rows']: state.last_closed_date = run.requested_end fields.append('last_closed_date') if state.coverage_start is None or run.requested_start < state.coverage_start: state.coverage_start = run.requested_start fields.append('coverage_start') state.status = ModuleStatus.READY else: state.status = ModuleStatus.NO_DATA state.last_sync_at = timezone.now() state.last_error_code = '' state.save(update_fields=fields) def reset_and_repopulate(domain, months: int) -> SyncRun: """ Borra lo importado de este sitio y vuelve a traerlo. Es la herramienta de desarrollo: si al mirar los datos aparece un defecto, se corrige y se relanza limpio en vez de convivir con una serie a medio arreglar. Sólo la alcanza un perfil de administración total, y por eso la comprobación del rol vive en la vista y no acá: este servicio hace lo que se le pide. **Borra únicamente las tablas de este módulo y de este sitio.** La cobertura, los sitemaps y los lotes de la otra herramienta no se tocan. No hay identificadores ni secuencias que reiniciar: todo hereda de `BaseModel`, que usa UUID como clave primaria. """ with transaction.atomic(): KeywordDaily.objects.filter(domain=domain).delete() KeywordPageDaily.objects.filter(domain=domain).delete() SyncState.objects.filter(domain=domain).update( last_closed_date=None, coverage_start=None, status=ModuleStatus.SYNCING, last_error_code='', ) return queue_backfill(domain, months) # --- Las variaciones ------------------------------------------------------ @dataclass(frozen=True) class Point: """Un extremo de una comparación, con el respaldo que lo sostiene.""" position: float impressions: int def window_average( series: dict[date, tuple[int, float]], end: date, days: int ) -> Point | None: """ La posición promedio de una ventana, ponderada por impresiones. Ponderada y no simple: un día con cinco impresiones y uno con quinientas no describen lo mismo, y promediarlos en partes iguales le daría al día flojo el poder de mover la tendencia. Es además cómo agrega Google. Los días sin dato **se saltean, no cuentan como cero**. Google omite las consultas poco frecuentes para proteger la privacidad, así que un día ausente casi nunca significa «no apareciste»: significa «no te lo puedo decir». Rellenarlo con un cero inventaría una caída. Devuelve `None` cuando la ventana no junta `MIN_IMPRESSIONS`, que es la forma de decir «no hay con qué comparar». La regla se aplica **al total del rango** y no día por día: los rangos largos acumulan muestra —una consulta con una impresión diaria junta una en la ventana de un día y quince en la de quince—, así que en la práctica el filtro lo siente la ventana corta y casi nunca las largas, que es lo que se busca. Escrito como excepción sólo para la de un día quedaría afuera la consulta casi muerta, donde ni quince días llegan a diez impresiones y dibujar una tendencia sería igual de falso. """ impressions = 0 weighted = 0.0 for offset in range(days): entry = series.get(end - timedelta(days=offset)) if entry is None: continue day_impressions, day_position = entry impressions += day_impressions weighted += day_position * day_impressions if impressions < MIN_IMPRESSIONS: return None return Point(position=weighted / impressions, impressions=impressions) def deltas( series: dict[date, tuple[int, float]], last: date ) -> dict[int, dict | None]: """ Las cuatro variaciones de una consulta, ancladas al último día cerrado. El signo sigue la convención de la especificación: **positivo es mejora**. Como en Google una posición más baja es mejor, se resta la actual de la anterior —de la 12 a la 7 son `+5`—, que es al revés de lo que la intuición sugiere y por eso se dice acá. Cada ventana compara contra la ventana inmediatamente anterior del mismo largo. La de un día es la única que compara dos días sueltos. """ result: dict[int, dict | None] = {} for days in DELTA_WINDOWS: current = window_average(series, last, days) previous = window_average(series, last - timedelta(days=days), days) if current is None or previous is None: result[days] = None continue result[days] = { 'value': round(previous.position - current.position, 1), 'current': round(current.position, 1), 'previous': round(previous.position, 1), } return result # --- Lo que leen las pantallas -------------------------------------------- #: Ventana por defecto de la tabla, en días. Veintiocho es lo que muestra el #: propio informe de Rendimiento, y coincidir con él permite contrastar. DEFAULT_WINDOW_DAYS = 28 #: Cuántos días hay que leer para poder calcular la variación más larga: la #: ventana de quince contra los quince anteriores. DELTA_SPAN_DAYS = max(DELTA_WINDOWS) * 2 def _aggregate(entries) -> dict: """ Junta varios días de una consulta en una sola fila. El CTR **sí** se calcula acá, al revés de lo que se hace con el dato diario. No es una contradicción: para un día se guarda el que manda Google porque es su número, pero un promedio de CTR de varios días no significa nada —hay que dividir los clics totales por las impresiones totales—, y la posición se pondera por impresiones por el mismo motivo. """ clicks = sum(entry.clicks for entry in entries) impressions = sum(entry.impressions for entry in entries) weighted = sum(entry.position * entry.impressions for entry in entries) return { 'clicks': clicks, 'impressions': impressions, 'ctr': (clicks / impressions) if impressions else 0.0, 'position': (weighted / impressions) if impressions else 0.0, } def keyword_rows(domain, *, days: int = DEFAULT_WINDOW_DAYS) -> list[dict]: """ La tabla de consultas: totales del período y las cuatro variaciones. Sale **entera** de `KeywordDaily`. La tabla de páginas no se toca ni para contar: sus impresiones están repartidas entre las URLs que aparecieron en cada búsqueda, y sumarlas devolvería totales inflados. **Cuántas páginas tiene cada consulta ya no viaja acá.** Estaba, contada sobre todo el período, y esa cuenta no significaba lo que la columna decía: una página que rankeó la semana pasada y otra que rankea esta no compiten, se reemplazaron. La competencia real se mide en `cannibalized_keywords`, que cuenta páginas **del mismo día**, y vive en su propio indicador. """ state = state_for(domain) if state.last_closed_date is None: return [] last = state.last_closed_date window_start = last - timedelta(days=days - 1) # Se lee el rango más largo de los dos: el que la tabla muestra y el que las # variaciones necesitan. Una sola consulta para las dos cosas. since = min(window_start, last - timedelta(days=DELTA_SPAN_DAYS)) entries = KeywordDaily.objects.filter(domain=domain, date__gte=since, date__lte=last) grouped: dict[str, list] = {} series: dict[str, dict[date, tuple[int, float]]] = {} for entry in entries: series.setdefault(entry.query, {})[entry.date] = (entry.impressions, entry.position) if entry.date >= window_start: grouped.setdefault(entry.query, []).append(entry) tracked = set( TrackedKeyword.objects.filter(domain=domain).values_list('keyword', flat=True) ) rows = [] for query, entries_of_query in grouped.items(): totals = _aggregate(entries_of_query) rows.append( { 'query': query, **totals, 'deltas': deltas(series[query], last), 'tracked': query in tracked, # Los dos ejes por los que se filtra la tabla viajan **calculados # por el servidor**. Es lo que garantiza que el largo de la tabla # filtrada sea igual a la cifra de la tarjeta que llevó hasta # ella: los dos salen de la misma función, no de la misma regla # escrita dos veces. 'band': classify_band(totals['position']), 'engagement': classify_engagement(totals['clicks'], totals['impressions']), } ) rows.sort(key=lambda row: row['impressions'], reverse=True) return rows #: Cuántas keywords entran en el gráfico del inicio. #: #: Diez es mucho para un gráfico —diez trazos no se siguen con la vista— y es el #: mínimo que la pregunta pide: «por qué me encuentran» no se contesta con tres. #: Lo que lo hace legible no es el color sino el resaltado, que muestra una por #: vez. TOP_KEYWORDS = 10 def top_keywords_series( domain, *, days: int = DEFAULT_WINDOW_DAYS, limit: int = TOP_KEYWORDS ) -> dict | None: """ Las consultas con más impresiones, con su recorrido diario. Las diez salen de `keyword_rows()` recortado, que ya viene ordenado por impresiones, y **la misma lista alimenta el gráfico y la tabla de abajo**. Si cada uno hiciera su propia consulta, un empate en impresiones podría dejarlos mostrando conjuntos distintos, y eso no se ve roto: se ven dos listas de diez. Las tres métricas viajan juntas. Son diez consultas por veintiocho días, o sea nada, y mandarlas juntas es lo que permite que cambiar de métrica no vaya a la red —que es lo único que justifica que sea un interruptor y no un filtro—. Devuelve `None` sin historia importada: un gráfico vacío se lee como un error de carga, y lo que corresponde es no dibujarlo. """ state = state_for(domain) if state.last_closed_date is None: return None rows = keyword_rows(domain, days=days)[:limit] if not rows: return None last = state.last_closed_date start = last - timedelta(days=days - 1) queries = [row['query'] for row in rows] entries = KeywordDaily.objects.filter( domain=domain, query__in=queries, date__gte=start, date__lte=last ).order_by('date') points: dict[str, list] = {query: [] for query in queries} for entry in entries: points[entry.query].append( { 'date': entry.date.isoformat(), 'impressions': entry.impressions, 'clicks': entry.clicks, 'position': round(entry.position, 1), } ) return { # El eje entero de la ventana, **sin trasladar de zona**: Google fecha en # horario del Pacífico y éstos son sus días, así que correrlos a la zona # de la cuenta haría que el rango no coincida con el que muestra Search # Console, que es contra lo que se lo quiere contrastar. 'dates': [(start + timedelta(days=offset)).isoformat() for offset in range(days)], 'keywords': rows, # Los días que Google no reportó **no están**. El hueco lo arma la # pantalla cruzando contra `dates`: mandar doscientos ochenta nulos para # no decir nada es peso puro. 'series': [{'query': query, 'points': points[query]} for query in queries], } def keyword_detail(domain, query: str, *, days: int = DEFAULT_WINDOW_DAYS) -> dict | None: """ Una consulta: su evolución, sus variaciones y qué páginas rankean por ella. Devuelve `None` sólo si la consulta no está seguida **y** no tiene un solo dato. Una consulta agregada a mano que Google todavía no reporta sí devuelve ficha: con la serie vacía y el aviso de que no hay datos, que es precisamente para lo que sirve poder agregarlas. """ state = state_for(domain) last = state.last_closed_date tracked = TrackedKeyword.objects.filter(domain=domain, keyword=query).first() if last is None: return None if tracked is None else _empty_detail(query, tracked) since = last - timedelta(days=max(days, DELTA_SPAN_DAYS)) entries = list( KeywordDaily.objects.filter( domain=domain, query=query, date__gte=since, date__lte=last ).order_by('date') ) if not entries and tracked is None: return None if not entries: return _empty_detail(query, tracked) window_start = last - timedelta(days=days - 1) series = {entry.date: (entry.impressions, entry.position) for entry in entries} return { 'query': query, 'tracked': tracked is not None, 'source': tracked.source if tracked else None, 'has_data': True, **_aggregate([entry for entry in entries if entry.date >= window_start]), 'deltas': deltas(series, last), 'history': [ { 'date': entry.date.isoformat(), 'clicks': entry.clicks, 'impressions': entry.impressions, 'ctr': entry.ctr, 'position': round(entry.position, 1), } for entry in entries if entry.date >= window_start ], 'pages': _pages_for_query(domain, query, window_start, last), } def _empty_detail(query: str, tracked) -> dict: return { 'query': query, 'tracked': True, 'source': tracked.source, 'has_data': False, 'clicks': 0, 'impressions': 0, 'ctr': 0.0, 'position': 0.0, 'deltas': dict.fromkeys(DELTA_WINDOWS), 'history': [], 'pages': [], } def _pages_for_query(domain, query: str, start: date, end: date) -> list[dict]: """ Las URLs que aparecen para una consulta, con cuánto aporta cada una. El reparto se calcula **entre estas filas**, que es de donde salen: dice qué porción del rendimiento observado en esta tabla se lleva cada página. No se compara contra el total de `KeywordDaily`, porque los dos números cuentan cosas distintas y su cociente no significaría nada. """ entries = KeywordPageDaily.objects.filter( domain=domain, query=query, date__gte=start, date__lte=end ) grouped: dict[str, list] = {} for entry in entries: grouped.setdefault(entry.page, []).append(entry) pages = [{'page': page, **_aggregate(rows)} for page, rows in grouped.items()] total = sum(page['impressions'] for page in pages) for page in pages: page['share'] = (page['impressions'] / total) if total else 0.0 pages.sort(key=lambda page: page['impressions'], reverse=True) return pages def page_rows(domain, *, days: int = DEFAULT_WINDOW_DAYS) -> list[dict]: """Qué páginas del sitio traen tráfico, en el período.""" state = state_for(domain) if state.last_closed_date is None: return [] last = state.last_closed_date start = last - timedelta(days=days - 1) entries = KeywordPageDaily.objects.filter(domain=domain, date__gte=start, date__lte=last) grouped: dict[str, list] = {} queries: dict[str, set] = {} for entry in entries: grouped.setdefault(entry.page, []).append(entry) queries.setdefault(entry.page, set()).add(entry.query) rows = [ {'page': page, **_aggregate(rows_of_page), 'queries': len(queries[page])} for page, rows_of_page in grouped.items() ] rows.sort(key=lambda row: row['impressions'], reverse=True) return rows def page_detail(domain, url: str, *, days: int = DEFAULT_WINDOW_DAYS) -> dict | None: """Por qué consultas aparece una URL: la pregunta inversa de la ficha de consulta.""" state = state_for(domain) if state.last_closed_date is None: return None last = state.last_closed_date start = last - timedelta(days=days - 1) entries = KeywordPageDaily.objects.filter( domain=domain, page=url, date__gte=start, date__lte=last ) grouped: dict[str, list] = {} for entry in entries: grouped.setdefault(entry.query, []).append(entry) if not grouped: return None keywords = [{'query': query, **_aggregate(rows)} for query, rows in grouped.items()] keywords.sort(key=lambda row: row['impressions'], reverse=True) return { 'page': url, **_aggregate(list(entries)), 'keywords': keywords, } # --- Los indicadores del inicio ------------------------------------------- #: Desde cuántas páginas compitiendo **el mismo día** una consulta cuenta como #: canibalizada. #: #: Tres y no dos: que dos páginas aparezcan por la misma búsqueda es corriente y #: muchas veces inofensivo —una de categoría y una de detalle—, así que empezar #: en dos convertiría a una de cada cuatro consultas en «problema» y el #: indicador dejaría de ser una señal. Medido sobre el sitio real, tres deja 36 #: de 470. CANNIBALIZATION_THRESHOLD = 3 #: Desde cuántas páginas el problema deja de ser un par de URLs parecidas y pasa #: a ser una sección entera compitiendo consigo misma. SEVERE_COMPETING = 6 #: Desde cuántas deja de ser el caso corriente —una de categoría y una de #: detalle— y ya no se explica por la estructura del sitio. HIGH_COMPETING = 4 def classify_severity(competing: int) -> str: """ Cuánto pesa una canibalización. **Único lugar donde se corta.** Mismo criterio que `classify_band`: el filtro de la tabla lee esta etiqueta en vez de volver a comparar en el navegador, así que no pueden discrepar. """ if competing >= SEVERE_COMPETING: return CannibalizationSeverity.SEVERE if competing >= HIGH_COMPETING: return CannibalizationSeverity.HIGH return CannibalizationSeverity.MODERATE def classify_rotation(competing: int, pages_in_period: int) -> str: """ Si el conjunto que compite es siempre el mismo o va cambiando. Que aparezcan más páginas en el período que en el peor día significa que la URL que rankea se está turnando. Es un problema distinto y se arregla distinto, por eso es un eje aparte y no un número más. """ if pages_in_period > competing: return PageRotation.ROTATING return PageRotation.STABLE def cannibalized_keywords(domain, *, days: int = DEFAULT_WINDOW_DAYS) -> list[dict]: """ Consultas por las que compiten varias páginas del sitio **el mismo día**. La distinción con «páginas distintas en el período» no es un detalle: contar el período mide casi el doble. Si la página A rankeó la semana pasada y la B rankea esta, no compitieron —se reemplazaron—, y eso es un cambio de URL dominante, que a veces es exactamente lo que se buscaba. Se informa el **máximo de páginas en un mismo día**, que es el peor momento observado. **No devuelve las URLs.** Las pide la fila cuando alguien la abre, con `cannibalization_pages()`. Mandarlas acá era mandar, en la carga de la pantalla, la lista completa de páginas de cada consulta afectada —decenas de filas por consulta— para dibujar una sola: las que se abren son las pocas que alguien mira. Y dejarlas de este lado además de pedirlas después sería pagar las dos veces, que es exactamente lo que la carga bajo demanda viene a evitar. """ state = state_for(domain) if state.last_closed_date is None: return [] last = state.last_closed_date start = last - timedelta(days=days - 1) entries = KeywordPageDaily.objects.filter( domain=domain, date__gte=start, date__lte=last ).values_list('query', 'date', 'page', 'impressions') per_day: dict[str, dict[date, set]] = {} per_page: dict[str, set] = {} impressions: dict[str, int] = {} for query, day, page, page_impressions in entries: per_day.setdefault(query, {}).setdefault(day, set()).add(page) per_page.setdefault(query, set()).add(page) impressions[query] = impressions.get(query, 0) + page_impressions rows = [] for query, days_of_query in per_day.items(): competing = max(len(pages) for pages in days_of_query.values()) if competing < CANNIBALIZATION_THRESHOLD: continue #: Cuántas distintas aparecieron en todo el período. Va al lado de #: `competing` a propósito: la diferencia entre las dos es cuánto rotó la #: URL que rankea, que es otra historia. in_period = len(per_page[query]) rows.append( { 'query': query, #: Cuántas se pisaron en el peor día observado. 'competing': competing, 'pages_in_period': in_period, #: Sumadas **entre páginas**, que es lo correcto acá y sólo acá: #: la pregunta es cuánto pesa el problema en el reparto por URL, #: no cuántas veces apareció la consulta. Ese total sale de #: `KeywordDaily` y es otro número (**O8**). 'impressions': impressions[query], #: Los dos ejes por los que se filtra la tabla, calculados acá #: por lo mismo que `band` y `engagement` en `keyword_rows`: el #: corte se escribe una vez, o la tabla filtrada deja de #: coincidir con lo que la etiqueta promete. 'severity': classify_severity(competing), 'rotation': classify_rotation(competing, in_period), } ) rows.sort(key=lambda row: (row['competing'], row['impressions']), reverse=True) return rows def cannibalization_pages( domain, query: str, *, days: int = DEFAULT_WINDOW_DAYS ) -> list[dict] | None: """ Qué URLs del sitio se pisan por una consulta, con lo que trajo cada una. Se pide por consulta y bajo demanda: es lo que hay que mirar para decidir cuál conservar y cuál redirigir, y de una pantalla con decenas de consultas afectadas se abren dos o tres. Devuelve `None` cuando la consulta no tiene ninguna página registrada en el período, que no es lo mismo que una lista vacía: una es «esa consulta no existe acá» y la otra sería «existe y no compite con nada», que no puede pasar —si está en la lista, compite—. """ state = state_for(domain) if state.last_closed_date is None: return None last = state.last_closed_date start = last - timedelta(days=days - 1) entries = KeywordPageDaily.objects.filter( domain=domain, query=query, date__gte=start, date__lte=last ) grouped: dict[str, list] = {} for entry in entries: grouped.setdefault(entry.page, []).append(entry) if not grouped: return None pages = [{'page': page, **_aggregate(rows)} for page, rows in grouped.items()] # Por CTR y no por impresiones: la pregunta de esta lista es **cuál # conservar**, y la que más se clickea es la que los buscadores prefieren # cuando se la ofrecen. Las impresiones desempatan, que es lo que evita que # una página con dos apariciones y un clic se ponga arriba de la que se lleva # el tráfico de verdad. pages.sort(key=lambda page: (page['ctr'], page['impressions']), reverse=True) return pages def near_first_page(domain, *, days: int = DEFAULT_WINDOW_DAYS) -> list[dict]: """ Consultas en la segunda página: cerca de donde está el tráfico. **No vuelve a cortar**: lee el tramo que `keyword_rows` ya escribió en cada fila. Es lo que hace que la tarjeta del inicio y la tabla filtrada por el mismo tramo devuelvan exactamente el mismo conjunto, que es la condición bajo la cual esa tarjeta puede permitirse enlazar en vez de abrir un modal. """ return [ row for row in keyword_rows(domain, days=days) if row['band'] == PositionBand.NEAR_FIRST_PAGE ] def reach_without_clicks(domain, *, days: int = DEFAULT_WINDOW_DAYS) -> list[dict]: """ Consultas por las que el sitio aparece y nadie entra. Casi nunca es un problema de posición: es el título o la descripción del resultado los que no invitan a hacer clic. Por eso va como indicador aparte y no mezclado con lo que se resuelve subiendo puestos. El piso de impresiones está adentro de `classify_engagement`, no acá: es parte de qué significa el valor, y afuera se podía olvidar. """ return [ row for row in keyword_rows(domain, days=days) if row['engagement'] == Engagement.NO_CLICKS ] def period_performance(domain, *, days: int = DEFAULT_WINDOW_DAYS) -> dict | None: """ Lo que trajo el período, contra el período anterior del mismo largo. La comparación es contra el tramo inmediatamente anterior y no contra un número suelto: «2.146 impresiones» no dice nada sin con qué compararlo, y un porcentaje sin denominador dice todavía menos. Devuelve `None` cuando no hay período anterior con datos: prometer una variación sin nada detrás sería inventarla. """ state = state_for(domain) if state.last_closed_date is None: return None last = state.last_closed_date current_start = last - timedelta(days=days - 1) previous_start = current_start - timedelta(days=days) entries = list( KeywordDaily.objects.filter(domain=domain, date__gte=previous_start, date__lte=last) ) current = [entry for entry in entries if entry.date >= current_start] previous = [entry for entry in entries if entry.date < current_start] # El tramo anterior sólo cuenta si el histórico lo cubre **entero**. # # Con una importación que arranca a mitad de ese tramo, lo que hay son unos # pocos días y no un período: compararlos contra veintiocho y presentarlo # como «contra el período anterior» inventa una caída del tamaño de lo que # falta importar. Es el mismo criterio del mínimo de impresiones —no se # compara contra lo que no se tiene—, aplicado a la escala en vez de a la # muestra. covered = state.coverage_start is not None and state.coverage_start <= previous_start return { 'current': _aggregate(current) if current else None, 'previous': _aggregate(previous) if previous and covered else None, 'keywords': len({entry.query for entry in current}), # Los cuatro extremos viajan aunque la pantalla los muestre chiquitos. # Una comparación contra «el período anterior» sin decir de cuándo a # cuándo deja la cifra sin escala: no se sabe si son siete días o # noventa, y con eso no se puede juzgar si la diferencia es mucha. 'current_start': current_start.isoformat(), 'current_end': last.isoformat(), 'previous_start': previous_start.isoformat(), 'previous_end': (current_start - timedelta(days=1)).isoformat(), } def home_metrics(domain, *, days: int = DEFAULT_WINDOW_DAYS) -> dict: """ Las tarjetas del inicio: **una cifra cada una, sin lista adentro**. Antes cada tarjeta viajaba con diez filas de muestra, porque el botón abría un modal: la regla del producto dice que una cifra se abre donde se la lee, y ninguna pantalla contestaba entonces la misma pregunta con el mismo conjunto —la tabla de consultas las mostraba todas—. Ahora sí lo hacen: la tabla se abre recortada por el mismo tramo y el mismo alcance que `keyword_rows()` escribió en cada fila, que son los que producen estas cifras. Con el destino contestando exactamente lo mismo, el modal sobra, y las muestras con él: eran diez filas que ya no dibuja nadie. """ return { 'performance': period_performance(domain, days=days), 'cannibalization': { 'total': len(cannibalized_keywords(domain, days=days)), 'threshold': CANNIBALIZATION_THRESHOLD, }, 'near_first_page': { 'total': len(near_first_page(domain, days=days)), 'range': list(NEAR_FIRST_PAGE), }, 'reach_without_clicks': { 'total': len(reach_without_clicks(domain, days=days)), 'minimum': REACH_MINIMUM_IMPRESSIONS, }, } def module_overview(domain) -> dict: """ Lo que muestra el inicio del módulo: **actividad, no consumo**. No lleva medidor de cuota, y es a propósito. Un medidor necesita un denominador contra el cual llenarse, y el de Search Analytics son treinta millones de llamadas diarias por proyecto contra las pocas que hace este módulo: marcaría cero para siempre. Un medidor que nunca se mueve no informa, ocupa el lugar de algo que sí y entrena a no mirar ese rincón. La cuota aparece cuando molesta, como estado, no como barra. """ state = state_for(domain) last_run = SyncRun.objects.filter(domain=domain).first() return { 'status': state.status, 'error_code': state.last_error_code or None, 'last_closed_date': state.last_closed_date.isoformat() if state.last_closed_date else None, 'coverage_start': state.coverage_start.isoformat() if state.coverage_start else None, 'last_sync_at': state.last_sync_at.isoformat() if state.last_sync_at else None, 'keywords': KeywordDaily.objects.filter(domain=domain) .values('query') .distinct() .count(), 'rows': KeywordDaily.objects.filter(domain=domain).count(), 'last_run': _run_props(last_run) if last_run else None, } def _run_props(run: SyncRun) -> dict: """ La forma que esperan las piezas del cliente que se reutilizan. `BatchProgress` y `BatchStateBadge` vienen de la otra herramienta y leen `state`, `total_items` y `processed_items`: mientras la corrida los publique con esos nombres, se dibujan sin tocarlas. Se reutilizan **las piezas**, no las rutas ni las pantallas. """ return { 'id': str(run.id), 'kind': run.kind, 'origin': run.origin, 'state': run.state, 'total_items': run.total_items, 'processed_items': run.processed_items, 'failed_items': run.failed_items, 'requested_start': run.requested_start.isoformat(), 'requested_end': run.requested_end.isoformat(), 'started_at': run.started_at.isoformat() if run.started_at else None, 'finished_at': run.finished_at.isoformat() if run.finished_at else None, 'error_code': run.error_code or None, 'summary': run.summary, } def run_history(domain, limit: int = 20) -> list[dict]: """ Las últimas corridas, para la pantalla de conexión. Viven ahí y no en una pantalla propia: con un backfill que tarda segundos y una corrida diaria de un día, un listado con filtros sería una entrada más de menú para algo que casi nunca se mira. **El recorte se dice en pantalla.** Un tope silencioso se lee como «esto es todo lo que hay». """ runs = SyncRun.objects.filter(domain=domain)[:limit] return [_run_props(run) for run in runs] # --- Objetivos ------------------------------------------------------------ def track_keyword(domain, keyword: str) -> TrackedKeyword: """ Empieza a seguir una consulta. El origen se decide por lo que ya hay guardado: si Google la viene reportando, la descubrió él; si no, la escribió una persona. La distinción no es cosmética —la ficha dice «sin datos de Google» en vez de «no hay nada»— y deducirla acá evita que la pantalla tenga que declararla. """ keyword = keyword.strip()[:500] discovered = KeywordDaily.objects.filter(domain=domain, query=keyword).exists() tracked, _created = TrackedKeyword.objects.get_or_create( domain=domain, keyword=keyword, defaults={ 'source': ( KeywordSource.GOOGLE_DISCOVERED if discovered else KeywordSource.CUSTOM ) }, ) return tracked def untrack_keyword(domain, keyword: str) -> None: """ Deja de seguirla. **No borra su historia.** Los días que Google ya reportó siguen siendo ciertos, y volver a seguirla más adelante tiene que encontrar su serie entera. """ TrackedKeyword.objects.filter(domain=domain, keyword=keyword.strip()).delete()