"""
Inspección de URLs y escritura del historial de cobertura.

El orden de las operaciones es lo que importa acá: se reserva cupo, se consulta,
se guarda. Si el cupo alcanza para menos de lo pedido, el lote termina en
`PARTIAL` y lo dice. Nunca en `COMPLETED`: quedaron URLs sin consultar, y
llamarlo «terminado» convertiría una foto parcial del sitio en una afirmación
sobre el sitio entero (principio I).

El historial se escribe **sólo cuando el estado cambia**. Un sitio de veinte mil
URLs consultadas a diario generaría siete millones de filas al año, casi todas
idénticas a la anterior. Lo que hay que poder responder es cuándo cambió algo,
no cuántas veces se preguntó.
"""

import csv
import logging
from collections.abc import Mapping
from dataclasses import dataclass, field
from datetime import timedelta

from django.db.models import (
    BooleanField,
    Count,
    ExpressionWrapper,
    F,
    Max,
    Min,
    OuterRef,
    Q,
    QuerySet,
    Subquery,
)
from django.utils import timezone

from apps.billing import limits
from apps.core.tables import Filter, TableSpec, applied_filters, apply_filters, from_slug
from apps.coverage.models import CoverageExport, CoverageRecord
from apps.coverage.translation import raw_fields, translate
from apps.domains.services import ensure_operational
from apps.jobs.budget import Origin, release, reserve
from apps.jobs.models import Batch, BatchKind, BatchOrigin, BatchState
from apps.sitemaps.models import CoverageState, Url

logger = logging.getLogger(__name__)


@dataclass
class InspectionSummary:
    queried: int = 0
    changed: int = 0
    unchanged: int = 0
    failed: int = 0
    pending: int = 0
    quota_granted: int = 0
    quota_requested: int = 0
    #: Los tres cambios que exigen una acción de la persona (T116).
    #:
    #: Se cuentan aparte de `changed` porque no todo cambio pide algo: una URL
    #: que pasa de «descubierta» a «rastreada» también cambió, y no hay nada que
    #: hacer con eso. Estos tres, en cambio, se resuelven en tres lugares
    #: distintos —el índice de Google, el servidor del sitio y su `robots.txt`—
    #: y por eso son tres cifras y no una.
    lost_indexing: int = 0
    #: Cuáles fueron, y no sólo cuántas.
    #:
    #: La cifra sola convierte el aviso en algo que no se puede accionar: dice
    #: que doce direcciones se cayeron del índice y no deja manera de saber
    #: cuáles. Buscarlas después en la tabla tampoco funciona, porque
    #: `record_inspection` pisa `coverage_state` con lo último que Google
    #: contestó: al día siguiente esa URL ya figura indexada otra vez y el
    #: recorrido que la vio caer no dejó rastro de cuál era.
    #:
    #: Por eso la dirección se guarda **congelada** —el texto, no una clave
    #: hacia `Url`—, igual que en `IndexingRequest`: un sitemap que deja de
    #: declararla la borra del inventario, y el aviso tiene que seguir
    #: explicándose después de eso.
    lost_indexing_urls: list[dict] = field(default_factory=list)
    #: URLs que pasaron a «error al descargarla»: Google no pudo obtener la
    #: página. Casi siempre es el servidor del sitio, no Google ni nosotros.
    fetch_errors: int = 0
    #: URLs que pasaron a «bloqueada por robots.txt». Puede ser deliberado, y
    #: por eso el aviso lo informa en vez de tratarlo como falla.
    blocked_robots: int = 0
    errors: list[str] = field(default_factory=list)

    def as_dict(self) -> dict:
        return {
            'queried': self.queried,
            'changed': self.changed,
            'unchanged': self.unchanged,
            'failed': self.failed,
            'pending': self.pending,
            'quota_granted': self.quota_granted,
            'quota_requested': self.quota_requested,
            'lost_indexing': self.lost_indexing,
            'lost_indexing_urls': self.lost_indexing_urls,
            'fetch_errors': self.fetch_errors,
            'blocked_robots': self.blocked_robots,
            'errors': self.errors,
        }


#: Cuánta prioridad tiene volver a preguntar por una URL, según lo último que
#: Google contestó sobre ella (T083).
#:
#: El criterio es una sola pregunta: **cuánto cambia la respuesta si volvemos a
#: preguntar hoy**. Un error de descarga se arregla y hay que enterarse; una
#: página rastreada y todavía sin indexar está en tránsito y su estado se mueve
#: solo; una excluida por `noindex` o bloqueada por `robots.txt` la decidió el
#: propio sitio y no va a cambiar hasta que el sitio cambie; una indexada es lo
#: más estable que hay.
#:
#: Los números no significan nada por sí solos: sólo importa su orden relativo.
PRIORITY_BY_STATE: dict[str, int] = {
    CoverageState.FETCH_ERROR: 40,
    CoverageState.CRAWLED_NOT_INDEXED: 30,
    CoverageState.DISCOVERED_NOT_INDEXED: 30,
    CoverageState.OTHER_NOT_INDEXED: 25,
    CoverageState.DUPLICATE_CANONICAL: 15,
    CoverageState.REDIRECT: 15,
    CoverageState.BLOCKED_ROBOTS: 10,
    CoverageState.EXCLUDED_NOINDEX: 10,
    CoverageState.INDEXED: 0,
    # Sin consultar no es un estado que Google haya informado: esas URLs van
    # primero por no tener fecha, no por su puntaje.
    CoverageState.UNKNOWN: 0,
}


def priority_score(state: str) -> int:
    """El puntaje que le corresponde a un estado. Lo desconocido no prioriza nada."""
    return PRIORITY_BY_STATE.get(state, 0)


def pending_urls(domain):
    """
    URLs por consultar, en el orden en que conviene hacerlo.

    Primero las que nunca se consultaron: una URL sin dato es la única sobre la
    que el producto no puede decir nada, y ése es exactamente el vacío que hay
    que cerrar. Después, las problemáticas: donde Google informó un error o una
    exclusión es donde la respuesta de mañana puede ser distinta, y donde
    enterarse tarde cuesta más. Y recién después, las más viejas.

    El puntaje va **antes** que la antigüedad a propósito, y tiene un costo que
    conviene tener presente: una URL con error de descarga vuelve a consultarse
    antes que una indexada más vieja, así que en un dominio con muchas
    problemáticas las estables se revisan más espaciadas. Es la decisión
    deliberada de T083: el ciclo sirve para descubrir cambios, y en las estables
    casi nunca los hay.

    Con el puntaje después de la fecha —como estaba— sólo desempataba entre
    URLs consultadas en el mismo instante exacto, que casi no ocurre: el
    criterio existía escrito y no se aplicaba nunca.
    """
    return (
        Url.objects.filter(domain=domain, in_sitemap=True)
        .annotate(
            never_checked=ExpressionWrapper(
                Q(last_checked_at__isnull=True), output_field=BooleanField()
            )
        )
        .order_by('-never_checked', '-priority_score', F('last_checked_at').asc(), 'loc')
    )


def queue_inspection(
    domain,
    *,
    limit: int | None = None,
    origin: str = BatchOrigin.SCHEDULED,
    url_ids: list[str] | None = None,
) -> Batch:
    """
    Deja el lote en cola y encola el trabajo. No consulta nada todavía.

    Es lo que usan la API y la pantalla. Antes las dos recorrían las URLs dentro
    de la petición y contestaban `202` igual: dos mil consultas a Google, una
    detrás de otra, con alguien esperando del otro lado de una respuesta que ya
    afirmaba haberlas encolado.

    El total de ítems queda en cero hasta que la tarea elija las URLs: cuántas
    entran depende del cupo que quede en ese momento, no del que había cuando se
    apretó el botón, y adelantar una cifra que después no se cumple es peor que
    no adelantar ninguna (RT-14).

    `url_ids` acota el recorrido a lo que alguien marcó en la tabla. Sin eso, la
    única forma de volver a preguntar por cuatro direcciones era consultar el
    sitio entero: doscientas cuarenta y dos llamadas del cupo del día para mirar
    cuatro, que es lo que llevaba a no mirarlas.
    """
    ensure_operational(domain)

    batch = Batch.objects.create(
        domain=domain,
        kind=BatchKind.URL_INSPECTION,
        origin=origin,
        state=BatchState.QUEUED,
    )

    # Importación diferida: el módulo de tareas importa este servicio.
    from apps.coverage.tasks import inspect_one_domain

    inspect_one_domain.delay(
        str(domain.id),
        limit=limit,
        origin=origin,
        batch_id=str(batch.id),
        url_ids=[str(value) for value in url_ids] if url_ids else None,
    )
    return batch


def inspect_domain(
    domain,
    *,
    limit: int | None = None,
    origin: str = BatchOrigin.SCHEDULED,
    urls=None,
    client=None,
    batch=None,
) -> Batch:
    """
    Consulta a Google el estado de las URLs pendientes, hasta donde alcance el cupo.

    `batch` llega cuando el trabajo se encoló antes: la petición ya creó el lote
    en `QUEUED` para poder devolverlo, y acá se lo toma en vez de crear un
    segundo. Sin ese parámetro, lo que la persona mira en pantalla y lo que el
    trabajador ejecuta serían dos lotes distintos.
    """
    ensure_operational(domain)

    manual = origin in (BatchOrigin.MANUAL, BatchOrigin.API)
    selection = list(urls) if urls is not None else list(pending_urls(domain)[: limit or 1000])

    if batch is None:
        batch = Batch.objects.create(
            domain=domain,
            kind=BatchKind.URL_INSPECTION,
            origin=origin,
            state=BatchState.RUNNING,
            started_at=timezone.now(),
            total_items=len(selection),
        )
    else:
        batch.state = BatchState.RUNNING
        batch.started_at = timezone.now()
        batch.total_items = len(selection)
        batch.save(update_fields=['state', 'started_at', 'total_items', 'updated_at'])

    summary = InspectionSummary(quota_requested=len(selection))

    if not selection:
        return _close(batch, summary)

    reservation = reserve(
        domain,
        len(selection),
        origin=Origin.MANUAL if manual else Origin.AUTOMATIC,
    )
    summary.quota_granted = reservation.granted

    if not reservation:
        summary.pending = len(selection)
        summary.errors.append(
            'No quedaba cupo diario para esta propiedad. Las URLs siguen en la cola del '
            'próximo ciclo.'
        )
        return _close(batch, summary)

    from apps.gsc.client import SearchConsoleClient
    from apps.gsc.errors import GoogleCallError, GoogleErrorCode

    gsc_client = client or SearchConsoleClient(domain.account)
    to_inspect = selection[: reservation.granted]
    summary.pending = len(selection) - len(to_inspect)
    spent = 0
    # Las que este recorrido dejó en un estado donde pedir la indexación puede
    # resolver algo. Se juntan acá y no se envían: el lote nace en borrador y
    # espera a una persona, como los otros dos caminos.
    candidates = []

    for done, url in enumerate(to_inspect, start=1):
        try:
            response = gsc_client.inspect_url(
                property_uri=domain.property_uri, url=url.loc, reservation=reservation
            )
        except GoogleCallError as exc:
            spent += 1
            summary.failed += 1
            batch.failed_items += 1
            summary.errors.append(f'{url.loc}: {exc.message}')

            if exc.code in (GoogleErrorCode.QUOTA_EXCEEDED, GoogleErrorCode.PERMISSION_DENIED):
                # Ante estos dos no tiene sentido seguir: el resto va a fallar
                # igual y cada intento gasta cupo que se podría usar mañana.
                summary.pending += len(to_inspect) - summary.queried - summary.failed
                break
            continue

        # Se mira antes de escribir: `record_inspection` pisa `coverage_state`
        # con lo que Google acaba de decir, así que después ya no hay contra qué
        # comparar. Es el único momento en que existen los dos estados a la vez.
        before = url.coverage_state

        spent += 1
        changed = record_inspection(url, response, batch=batch)
        summary.queried += 1
        batch.processed_items += 1
        if changed:
            summary.changed += 1
            _count_relevant_change(summary, url, before, url.coverage_state)
        else:
            summary.unchanged += 1

        from apps.indexing.rules import triggers_automatic_request

        if triggers_automatic_request(url.coverage_state):
            candidates.append(url)

        if done % PROGRESS_EVERY == 0:
            _note_progress(batch)

    # Lo reservado y no gastado vuelve al cupo: si no, un corte temprano dejaría
    # al usuario sin consultas por el resto del día sin que Google le hubiera
    # descontado ninguna.
    release(reservation, reservation.granted - spent)
    batch.quota_consumed = spent

    if spent:
        limits.check(domain.account, limits.Action.INSPECT_URLS, spent, domain=domain)

    batch = _close(batch, summary)
    _draft_indexing_batch(domain, candidates)
    return batch


def _draft_indexing_batch(domain, candidates) -> None:
    """
    Deja un borrador de indexación con lo que este recorrido encontró (camino 3).

    **Borrador y no envío.** Nada sale hacia Google sin que una persona lo
    confirme, así que esto termina donde terminan los otros dos caminos: en un
    lote esperando, con su aviso de nivel acción para que no se acumule
    invisible.

    Va después de cerrar el lote de inspección y nunca antes: si armar el
    borrador fallara, el recorrido que sí ocurrió tiene que quedar guardado
    igual. Por eso el fallo se anota y no se propaga — un problema al proponer
    trabajo futuro no puede borrar el trabajo ya hecho.
    """
    if not candidates:
        return

    from apps.indexing.rules import ONCE_ONLY_STATES
    from apps.indexing.services import create_batch, has_been_requested, pending_request_exists

    selection = []
    for url in candidates:
        # Los de `ONCE_ONLY_STATES` entran una sola vez por URL. Si el fallo de
        # descarga era pasajero el pedido ya sirvió; si sigue igual, el problema
        # está en el sitio y no en el índice de Google, y volver a pedirlo sólo
        # quema cupo.
        if url.coverage_state in ONCE_ONLY_STATES and has_been_requested(url):
            continue
        # Y ninguna que ya esté esperando en otro lote: dos borradores con la
        # misma dirección la mandarían dos veces y gastarían dos del techo real.
        if pending_request_exists(url):
            continue
        selection.append(url)

    if not selection:
        return

    from apps.core.errors import ApiError

    try:
        create_batch(domain, selection, origin=BatchOrigin.SCHEDULED)
    except ApiError as exc:
        logger.warning(
            'El recorrido de %s no pudo dejar el borrador de indexación: %s',
            domain.hostname,
            exc.message,
        )


#: Cuántas direcciones caídas del índice entran en el resumen de un lote.
#:
#: Mismo criterio que los motivos de fallo y que la ficha de indexación:
#: doscientas alcanzan para reconocer qué pasó, y el total va aparte para que el
#: recorte no se lea como «éstas fueron todas». El resumen vive en una columna
#: JSON que se lee entera en cada visita a la pantalla del lote, así que un sitio
#: al que se le caen cinco mil direcciones no puede escribirlas todas ahí.
MAX_LOST_INDEXING_URLS = 200


def _count_relevant_change(
    summary: InspectionSummary, url: Url, before: str, after: str
) -> None:
    """
    Anota los cambios de estado que exigen una acción de la persona (T116).

    Se cuentan **transiciones**, no estados: lo que importa avisar es que algo
    pasó a estar así en este recorrido. Una URL que ya venía con error de
    descarga desde la semana pasada no es una novedad, y contarla otra vez cada
    día convertiría el aviso en un número que sube solo y que nadie mira.

    Una misma URL puede caer en dos cuentas —de indexada a error de descarga es
    las dos cosas— y está bien: son dos hechos distintos, uno dice qué se perdió
    y el otro dónde mirar, y se resuelven en lugares distintos.

    De la pérdida de indexación se guarda además **cuál**: es el único de los
    tres que se avisa por notificación, y una notificación que no dice sobre qué
    dirección habla no se puede atender.
    """
    if before == CoverageState.INDEXED and after != CoverageState.INDEXED:
        summary.lost_indexing += 1
        # El contador sigue subiendo aunque la lista ya esté llena: la cifra es
        # el total y la lista es una muestra, y la pantalla compara las dos para
        # poder decir cuántas quedaron afuera. Igualarlas escondería el recorte.
        if len(summary.lost_indexing_urls) < MAX_LOST_INDEXING_URLS:
            summary.lost_indexing_urls.append({'loc': url.loc, 'state': after})

    if after == CoverageState.FETCH_ERROR:
        summary.fetch_errors += 1

    if after == CoverageState.BLOCKED_ROBOTS:
        summary.blocked_robots += 1


#: Cada cuántas URLs el lote anota lo que lleva hecho (T075).
#:
#: La pantalla consulta el estado cada cinco segundos (RT-13) y una inspección
#: tarda cerca de un segundo, así que veinticinco URLs son unos veinte segundos:
#: la barra avanza a saltos visibles sin agregar una escritura por consulta.
#: Anotar al final —como estaba— hacía saltar la barra de cero al total, que no
#: distingue un trabajo que avanza de uno que se colgó.
PROGRESS_EVERY = 25


def _note_progress(batch: Batch) -> None:
    """Deja escrito el avance, sin tocar nada más del lote."""
    batch.save(update_fields=['processed_items', 'failed_items', 'updated_at'])


def record_inspection(url: Url, response: dict, *, batch=None) -> bool:
    """
    Guarda el resultado y devuelve si el estado cambió.

    `fetched_at` se toma del momento de la respuesta y es obligatorio. Un estado
    de indexación sin la fecha en que Google lo dijo no es un dato verificable, y
    todo lo que este producto afirma se apoya en esa fecha.
    """
    index_status = (response or {}).get('inspectionResult', {}).get('indexStatusResult', {})
    state = translate(index_status)
    now = timezone.now()

    changed = url.coverage_state != state

    # Se escribe si cambió o si es la primera lectura. La primera va siempre,
    # aunque coincida con el valor por defecto: sin ella no habría ninguna fila
    # que pruebe cuándo se supo por primera vez.
    if changed or url.last_checked_at is None:
        CoverageRecord.objects.create(
            url=url,
            batch=batch,
            state=state,
            fetched_at=now,
            last_crawl_time=_parse_date(index_status.get('lastCrawlTime')),
            **raw_fields(index_status),
        )

    url.coverage_state = state
    url.last_checked_at = now
    # El puntaje se recalcula en cada lectura y no se acumula: describe el estado
    # de hoy, no la historia de la URL. Guardarlo acá —y no al ordenar— es lo que
    # permite que el ciclo elija con un índice en vez de traducir diez estados a
    # números en cada corrida sobre cien mil filas.
    url.priority_score = priority_score(state)
    url.save(update_fields=['coverage_state', 'last_checked_at', 'priority_score', 'updated_at'])

    return changed


def coverage_summary(domain) -> dict:
    """
    Reparto de estados del dominio, con «sin consultar» contado aparte.

    Las URLs sin dato quedan **fuera** del porcentaje y de cualquier barra
    apilada. Meterlas adentro convertiría «todavía no preguntamos» en «no está
    indexada», que es precisamente la mentira que el producto no puede cometer.
    Y todo porcentaje viaja con su denominador: un 62 % sin decir sobre cuántas
    no significa nada.
    """
    urls = Url.objects.filter(domain=domain, in_sitemap=True)

    by_state = {state: 0 for state, _ in CoverageState.choices}
    for row in urls.values('coverage_state').annotate(total=Count('id')):
        by_state[row['coverage_state']] = row['total']

    without_data = by_state.pop(CoverageState.UNKNOWN, 0)
    with_data = sum(by_state.values())
    indexed = by_state.get(CoverageState.INDEXED, 0)

    return {
        'total': with_data + without_data,
        'with_data': with_data,
        'without_data': without_data,
        'indexed': indexed,
        # Nulo, no cero: sin ninguna URL con dato no hay porcentaje que informar,
        # y un cero se leería como «ninguna está indexada».
        'indexed_percentage': round(indexed * 100 / with_data, 1) if with_data else None,
        'by_state': by_state,
        'last_checked_at': (
            urls.exclude(last_checked_at=None)
            .order_by('-last_checked_at')
            .values_list('last_checked_at', flat=True)
            .first()
        ),
    }


def coverage_annotations() -> dict:
    """
    El mismo resumen que `coverage_summary()`, pero para una lista de dominios (FR-062).

    El listado necesita el reparto de cada fila y llamar a `coverage_summary()`
    una vez por dominio serían dos consultas por renglón: con el paginado en
    cincuenta filas, cien viajes a la base para dibujar una pantalla. Acá las
    cuatro cifras salen como agregados condicionales sobre el mismo join, o sea
    en la consulta que ya trae los dominios.

    Se devuelven armadas en el momento y no como constante del módulo para que
    ordenar por cobertura sea posible: las claves son alias reales del queryset,
    y `order_by('coverage_known')` las usa igual que a un campo.
    """
    in_sitemap = Q(urls__in_sitemap=True)

    return {
        'coverage_total': Count('urls', filter=in_sitemap),
        # `known` excluye `UNKNOWN` porque es el **único** denominador honesto:
        # una URL sin consultar no es una URL sin indexar, y meterla en el
        # reparto convierte «todavía no preguntamos» en una afirmación sobre
        # Google que no podemos hacer (R-B).
        'coverage_known': Count(
            'urls', filter=in_sitemap & ~Q(urls__coverage_state=CoverageState.UNKNOWN)
        ),
        'coverage_indexed': Count(
            'urls', filter=in_sitemap & Q(urls__coverage_state=CoverageState.INDEXED)
        ),
        'coverage_last_cycle': Max('urls__last_checked_at', filter=in_sitemap),
    }


def coverage_row(domain) -> dict | None:
    """
    Traduce las anotaciones de `coverage_annotations()` a lo que muestra una fila.

    Devuelve **nulo** y no un montón de ceros cuando el dominio todavía no tiene
    ninguna URL descubierta: un cero en la columna de cobertura se lee como
    «ninguna indexada», que es una afirmación sobre Google, mientras que lo que
    pasa de verdad es que todavía no leímos ningún sitemap.

    Lee los atributos sin `getattr` de cortesía a propósito: si alguien lista
    dominios sin anotarlos, tiene que romper acá y no devolver en silencio un
    resumen vacío para toda la tabla.
    """
    total = domain.coverage_total
    if not total:
        return None

    known = domain.coverage_known
    last_cycle = domain.coverage_last_cycle

    return {
        'total_urls': total,
        'known_urls': known,
        'indexed': domain.coverage_indexed,
        'not_indexed': known - domain.coverage_indexed,
        # Nulo mientras no haya corrido ningún ciclo. La fila lo necesita para
        # decir «sin datos todavía» en vez de mostrar el reparto en cero (R-A).
        'last_cycle_at': last_cycle.isoformat() if last_cycle else None,
    }


def cycle_estimate(domain) -> dict:
    """
    Cuánto tarda una vuelta completa al ritmo del cupo diario.

    Es una estimación de nuestra capacidad de consulta, no una predicción sobre
    Google: dice cada cuánto podemos volver a mirar cada URL, no cuándo Google
    va a rastrearlas.
    """
    pending = Url.objects.filter(domain=domain, in_sitemap=True).count()
    per_day = domain.automatic_budget

    return {
        'monitored_urls': pending,
        'queries_per_day': per_day,
        'days_per_cycle': (pending + per_day - 1) // per_day if per_day else None,
    }


def _parse_date(value):
    if not value:
        return None
    from django.utils.dateparse import parse_datetime

    return parse_datetime(value)


def _close(batch: Batch, summary: InspectionSummary) -> Batch:
    if summary.pending or (summary.errors and summary.queried):
        batch.state = BatchState.PARTIAL
    elif summary.errors and not summary.queried:
        batch.state = BatchState.FAILED
    else:
        batch.state = BatchState.COMPLETED

    batch.finished_at = timezone.now()
    batch.summary = summary.as_dict()
    batch.save(
        update_fields=[
            'state',
            'finished_at',
            'summary',
            'processed_items',
            'failed_items',
            'quota_consumed',
            'updated_at',
        ]
    )

    # Los avisos se dejan acá, en el único lugar donde un lote de inspección se
    # cierra, y no en cada quien lo invoca: así valen igual para el ciclo
    # automático, para el botón de la pantalla y para la API. La importación es
    # diferida porque los avisos leen el modelo de lotes.
    from apps.notifications import services as notifications

    notifications.batch_finished(batch)

    # Los tres cambios que exigen una acción, cada uno con su aviso (T116). Son
    # lo que nadie va a descubrir mirando el tablero: la cifra global se mueve
    # poco cuando caen doce URLs de cinco mil, y las tres se resuelven en
    # lugares distintos. `notify()` no escribe dos veces el mismo hecho, así que
    # un lote reintentado no vuelve a avisar.
    notifications.coverage_drop(batch.domain, count=summary.lost_indexing, batch=batch)
    notifications.fetch_errors(batch.domain, count=summary.fetch_errors, batch=batch)
    notifications.crawl_blocked(batch.domain, count=summary.blocked_robots, batch=batch)
    return batch


def urls_for_display(domain, *, state: str = '', page: int = 1, per_page: int = 50) -> dict:
    """
    Página de URLs para la tabla, con su total.

    El total viaja siempre: una tabla que muestra cincuenta filas sin decir
    cuántas hay en total no permite saber si se está viendo el principio de algo
    grande o todo lo que existe.

    `state` llega en su forma pública —la que viaja en la dirección— igual que en
    la pantalla. Es lo que hace que el mismo enlace sirva para las dos.
    """
    queryset = coverage_queryset(domain)

    stored = from_slug(state, CoverageState.values) if state else ''
    if stored:
        queryset = queryset.filter(coverage_state=stored)

    total = queryset.count()
    start = max(page - 1, 0) * per_page

    return {
        'items': list(queryset[start : start + per_page]),
        'count': total,
        'page': page,
        'per_page': per_page,
        'pages': (total + per_page - 1) // per_page if total else 1,
    }


# --- La tabla de cobertura --------------------------------------------------


#: Filas a partir de las cuales la exportación deja de resolverse en el acto
#: (FR-058). Por debajo el archivo sale en la misma respuesta; por encima hay
#: que avisar antes de que alguien confirme, porque la descarga tarda.
EXPORT_THRESHOLD = 5000

#: Los tres recortes que la pantalla ofrece como un solo clic.
#:
#: `not_indexed` **excluye** `UNKNOWN` a propósito, y es la razón de que este
#: diccionario exista en vez de armarse en el navegador: meter las sin consultar
#: adentro de «no indexadas» convierte «todavía no preguntamos» en «Google dijo
#: que no», que es la afirmación falsa que el producto no puede cometer (R-B).
STATE_GROUPS: dict[str, tuple[str, ...]] = {
    'with_data': tuple(s for s in CoverageState.values if s != CoverageState.UNKNOWN),
    'not_indexed': tuple(
        s for s in CoverageState.values if s not in (CoverageState.UNKNOWN, CoverageState.INDEXED)
    ),
    'without_data': (CoverageState.UNKNOWN,),
}


#: Qué se puede ordenar y filtrar en la tabla de cobertura (RT-09).
#:
#: El orden por omisión es la dirección y **no** el estado. Ordenar por estado de
#: entrada dejaría las «sin consultar» amontonadas en la última página, invisibles
#: para quien nunca pasa de la primera; y son justamente las URLs sobre las que el
#: producto no puede afirmar nada.
#:
#: Sólo se declara ordenable lo que la base puede resolver con un índice. El
#: motivo crudo, la canónica y el último rastreo llegan por subconsulta al
#: historial: ofrecer un orden por esas columnas sería ofrecer un recorrido
#: completo de la tabla cada vez que alguien toca un encabezado.
COVERAGE_TABLE = TableSpec(
    sortable={
        'url': ('loc',),
        'state': ('coverage_state', 'loc'),
        'fetched': ('last_checked_at', 'loc'),
        'sitemap': ('in_sitemap', 'loc'),
    },
    default_sort='url',
    filters=(
        Filter(param='state', lookup='coverage_state', choices=tuple(CoverageState.values)),
        Filter(
            param='group',
            lookup='coverage_state__in',
            choices=tuple(STATE_GROUPS),
            cast=lambda key: STATE_GROUPS[key],
        ),
        # «yes»/«no» y no «true»/«false»: es lo que se lee en la dirección de la
        # vista, que es una dirección que la gente copia y pega.
        Filter(
            param='sitemap',
            lookup='in_sitemap',
            choices=('yes', 'no'),
            cast=lambda value: value == 'yes',
        ),
        Filter(param='q', lookup='loc__icontains'),
    ),
)


def coverage_queryset(domain) -> QuerySet:
    """
    Las URLs que la pantalla de cobertura puede mostrar.

    Entran las que hoy declara un sitemap y las que alguna vez tuvieron dato. Una
    URL que salió del sitemap y nunca se consultó queda afuera: no hay nada que
    decir de ella, y contarla inflaría el total de la tabla por encima del
    denominador del resumen sin que nada explicara la diferencia.
    """
    return Url.objects.filter(domain=domain).filter(
        Q(in_sitemap=True) | Q(last_checked_at__isnull=False)
    )


def with_latest_record(queryset: QuerySet) -> QuerySet:
    """
    Agrega a cada URL lo que Google contestó la última vez que cambió de estado.

    El motivo crudo, las dos canónicas y el último rastreo viven en
    `CoverageRecord`, no en `Url`. Se traen por subconsulta y no por `join`
    porque el `join` devolvería una fila por registro del historial y habría que
    agrupar; la subconsulta, en cambio, se evalúa sólo para las filas que la
    página llegó a pedir.
    """

    def latest(column: str) -> Subquery:
        return Subquery(
            CoverageRecord.objects.filter(url=OuterRef('pk'))
            .order_by('-fetched_at')
            .values(column)[:1]
        )

    return queryset.annotate(
        raw_coverage_state=latest('raw_coverage_state'),
        google_canonical=latest('google_canonical'),
        user_canonical=latest('user_canonical'),
        last_crawl_time=latest('last_crawl_time'),
    )


def filtered_urls(domain, params: Mapping[str, str]) -> QuerySet:
    """
    Las URLs que sobreviven a los filtros vigentes, sin paginar.

    La exportación tiene que llevarse **exactamente** el recorte que está en
    pantalla (FR-058): un archivo con otro filtro que el que se estaba mirando
    es peor que no exportar nada, porque nadie lo vuelve a revisar. Por eso el
    recorte sale de la misma declaración que usa la tabla —`COVERAGE_TABLE`—, que
    sigue siendo la única fuente de qué filtros existen y qué valores aceptan.
    """
    return apply_filters(coverage_queryset(domain), COVERAGE_TABLE, params)


def retired_count(domain) -> int:
    """
    Cuántas URLs mira la tabla que el resumen no cuenta.

    Son las que salieron de todos los sitemaps y conservan su último dato. La
    tabla las muestra —hay algo que decir de ellas— y el resumen no las cuenta
    —ya no se monitorean—, así que los dos denominadores de la pantalla son
    distintos a propósito.

    El número existe para poder decirlo. Dos cifras que no cierran, sin una
    frase que explique por qué, hacen dudar del resto del tablero: es el
    material del que están hechas las mentiras que esta vista trata de evitar, y
    da igual que la diferencia sea legítima si nadie la puede reconstruir.
    """
    return Url.objects.filter(domain=domain, in_sitemap=False).exclude(last_checked_at=None).count()


def state_counts(domain) -> dict[str, int]:
    """
    Cuántas URLs hay en cada estado, sobre el mismo conjunto que lista la tabla.

    Se cuenta acá y no desde `coverage_summary()` porque los dos conjuntos no son
    el mismo: el resumen mide lo que hoy se monitorea —lo que declara un sitemap—
    y la tabla muestra además lo que dejó de declararse pero tiene historial. Un
    filtro que dijera «412» y después mostrara otra cantidad de filas haría dudar
    de todo lo demás que hay en la pantalla.
    """
    counts = {state: 0 for state in CoverageState.values}
    for row in coverage_queryset(domain).values('coverage_state').annotate(total=Count('id')):
        counts[row['coverage_state']] = row['total']
    return counts


def coverage_window(domain) -> dict:
    """
    Desde cuándo y hasta cuándo van los datos que la pantalla está mostrando.

    Sin los dos extremos, «1.240 URLs tienen dato» no dice si ese dato es de esta
    mañana o de un recorrido que arrancó hace tres semanas y todavía no terminó.
    """
    bounds = (
        Url.objects.filter(domain=domain, in_sitemap=True)
        .exclude(last_checked_at=None)
        .aggregate(first=Min('last_checked_at'), last=Max('last_checked_at'))
    )
    return {
        'start': bounds['first'].isoformat() if bounds['first'] else None,
        'end': bounds['last'].isoformat() if bounds['last'] else None,
    }


def url_history(domain, loc: str) -> dict | None:
    """
    La línea de tiempo de una URL: cada cambio de estado con su fecha (FR-060).

    Devuelve nada si la URL no pertenece al dominio, para que la vista pueda
    decir «esa dirección no está en este dominio» en vez de abrir un panel vacío.

    Dos fechas distintas viajan a propósito. `fetched_at` de cada entrada es
    cuándo el estado **cambió**; `last_checked_at` de la URL es cuándo se lo
    confirmó por última vez. Mostrar sólo la primera haría creer que desde
    entonces nadie volvió a mirar, y sólo la segunda borraría cuándo empezó a ser
    cierto lo que hoy se afirma.
    """
    url = Url.objects.filter(domain=domain, loc=loc).first()
    if url is None:
        return None

    records = list(
        CoverageRecord.objects.filter(url=url).select_related('batch').order_by('-fetched_at')
    )
    oldest = len(records) - 1

    return {
        'url': url.loc,
        'in_sitemap': url.in_sitemap,
        'current': {
            'state': url.coverage_state,
            # Nulo sólo cuando nunca se consultó. Un estado con dato y sin esta
            # fecha es un defecto, y la pantalla lo muestra como tal (RT-02).
            'fetched_at': url.last_checked_at.isoformat() if url.last_checked_at else None,
        },
        'entries': [
            {
                'id': str(record.id),
                'state': record.state,
                'fetched_at': record.fetched_at.isoformat(),
                'raw_coverage_state': record.raw_coverage_state,
                'google_canonical': record.google_canonical,
                'user_canonical': record.user_canonical,
                'last_crawl_time': (
                    record.last_crawl_time.isoformat() if record.last_crawl_time else None
                ),
                # El más viejo es el primer dato que tuvimos, no un cambio: antes
                # de él no había un estado anterior del que haber cambiado.
                'first': index == oldest,
                'batch': _record_batch(record),
            }
            for index, record in enumerate(records)
        ],
    }


def _record_batch(record: CoverageRecord) -> dict | None:
    """
    El lote que produjo la lectura, si todavía existe.

    Admite nada porque los lotes se borran y el registro no: la fecha de
    obtención sigue siendo cierta aunque ya no se pueda decir en qué corrida se
    obtuvo.
    """
    batch = record.batch
    if batch is None:
        return None

    return {
        'id': str(batch.id),
        'state': batch.state,
        'origin': batch.origin,
    }


# --- La exportación ---------------------------------------------------------


#: Las columnas del archivo, en su orden. Son las mismas que la tabla de la
#: pantalla: una columna que se ve y no se exporta convierte al CSV en un
#: resumen de otra cosa. Están publicadas en `contracts/openapi.yaml`, así que
#: agregar o mover una es un cambio de contrato.
#:
#: **Van en inglés y no se traducen, en ningún idioma.** El archivo lo lee una
#: planilla o un script, no una persona: una cabecera que cambia con el idioma de
#: quien exportó rompe cualquier fórmula, cualquier `pandas.read_csv` y cualquier
#: importación armada contra la exportación anterior. Lo que sí se traduce es la
#: columna que se lee: `state_label` sale del vocabulario del producto, en el
#: idioma de la cuenta.
EXPORT_HEADER = (
    'url',
    'state',
    'state_label',
    'fetched_at',
    'google_reason',
    'google_canonical',
    'last_crawl',
    'in_sitemap',
    'display_timezone',
)

#: Cada cuántas filas escritas el lote anota lo que lleva.
#:
#: Coincide con el tamaño de trozo de la consulta: anotar más seguido serían
#: escrituras que no cambian lo que se ve —la pantalla consulta cada cinco
#: segundos (RT-13)— y anotar menos dejaría la barra quieta durante minutos, que
#: es indistinguible de un trabajo colgado.
EXPORT_PROGRESS_EVERY = 2000


class _Buffer:
    """Escritor que devuelve la línea en vez de guardarla, para poder transmitir."""

    def write(self, value):
        return value


def export_rows(domain, filters: Mapping[str, str]) -> QuerySet:
    """Las filas del archivo: el mismo recorte de la pantalla, con su dato crudo."""
    return with_latest_record(filtered_urls(domain, filters)).values_list(
        'loc',
        'coverage_state',
        'last_checked_at',
        'raw_coverage_state',
        'google_canonical',
        'last_crawl_time',
        'in_sitemap',
    )


def export_csv(domain, filters: Mapping[str, str], *, on_progress=None, language: str = ''):
    """
    Emite el CSV línea por línea, para poder transmitirlo o escribirlo a disco.

    La fecha va como columna y no en una nota al pie porque el archivo se abre
    en otro lado, meses después, y un estado sin la fecha en que Google lo dijo
    deja de significar algo (R-A).

    `language` decide en qué idioma sale la **única** columna que se lee con los
    ojos, `state_label`. Sin él se usa el de la cuenta dueña del dominio: la
    exportación también corre en un lote, donde no hay petición de la cual
    deducir nada.

    `on_progress` recibe cuántas filas de datos se llevan escritas. Lo usa el
    lote de exportación para poder informar avance; la descarga inmediata no lo
    necesita y no lo pasa.
    """
    from django.conf import settings

    from apps.core.catalog import text
    from apps.core.dates import display_timezone

    writer = csv.writer(_Buffer())
    yield writer.writerow(list(EXPORT_HEADER))

    tz = display_timezone()
    reading_language = language or domain.account.language
    written = 0

    for loc, state, fetched_at, reason, canonical, crawled_at, in_sitemap in export_rows(
        domain, filters
    ).iterator(chunk_size=EXPORT_PROGRESS_EVERY):
        yield writer.writerow(
            [
                loc,
                # El código y su etiqueta, en columnas separadas. El código es el
                # que sirve para filtrar en una planilla o cruzar contra otra
                # exportación, y no cambia con el idioma; la etiqueta es la que
                # se lee, y por eso sí lo hace.
                state or '',
                text(f'coverage.{state}.label', reading_language) if state else '',
                _in_timezone(fetched_at, tz),
                # El texto original de Google, sin traducir. Es la prueba de lo
                # que dijo, y es lo que permite entender un «sin indexar, otro
                # motivo» sin volver a consultarle.
                reason or '',
                canonical or '',
                # Cuándo Google visitó la página, que no es cuándo se lo
                # preguntamos: van en columnas distintas porque confundirlas
                # envejece o rejuvenece el dato en la dirección equivocada.
                _in_timezone(crawled_at, tz),
                # `true`/`false` y no «sí»/«no»: la columna la lee una planilla y
                # un valor que cambia de idioma rompe cualquier fórmula armada
                # contra la exportación anterior.
                'true' if in_sitemap else 'false',
                settings.DISPLAY_TIMEZONE,
            ]
        )

        written += 1
        if on_progress is not None and written % EXPORT_PROGRESS_EVERY == 0:
            on_progress(written)

    if on_progress is not None:
        on_progress(written)


def _in_timezone(moment, tz) -> str:
    """
    La fecha, expresada en la zona que declara la última columna.

    Sin convertir, el archivo diría «America/Argentina/Buenos_Aires» al lado de
    instantes en UTC: dos afirmaciones sobre la misma fecha que no coinciden, y
    quien abre la planilla meses después no tiene cómo saber cuál manda.
    """
    if moment is None:
        return ''
    if timezone.is_aware(moment):
        moment = moment.astimezone(tz)
    return moment.isoformat()


def export_path(export: CoverageExport):
    """
    Dónde vive el archivo de una exportación.

    Fuera de `STATIC_ROOT` y sin dirección pública: el CSV lleva las URLs del
    sitio de una cuenta, y se sirve por una vista que comprueba de quién es.
    """
    from django.conf import settings

    return settings.EXPORT_ROOT / export.file_name


def queue_export(domain, filters: Mapping[str, str], *, origin: str = BatchOrigin.MANUAL) -> Batch:
    """
    Deja el lote de exportación en cola y encola el trabajo (FR-058).

    No exige que el dominio esté operativo, a diferencia de la sincronización y
    la inspección: exportar no le pregunta nada a Google, lee lo que ya está
    guardado. Un dominio que perdió el acceso conserva su historial, y negarle
    el archivo justo cuando alguien necesita revisar qué pasó sería quitarle el
    dato por un motivo que no tiene que ver con el dato.
    """
    applied = applied_filters(COVERAGE_TABLE, filters)

    batch = Batch.objects.create(
        domain=domain,
        kind=BatchKind.COVERAGE_EXPORT,
        origin=origin,
        state=BatchState.QUEUED,
        total_items=filtered_urls(domain, applied).count(),
    )
    export = CoverageExport.objects.create(batch=batch, filters=applied)

    # Importación diferida: el módulo de tareas importa este servicio.
    from apps.coverage.tasks import build_export

    build_export.delay(str(export.id))
    return batch


def write_export(export: CoverageExport) -> CoverageExport:
    """
    Escribe el archivo de una exportación encolada y cierra su lote.

    El progreso se anota mientras se escribe y no al final: un archivo de cien
    mil filas tarda minutos, y una barra que salta de cero al total no distingue
    un trabajo que avanza de uno que se colgó (RT-14).
    """
    from django.conf import settings

    batch = export.batch
    batch.state = BatchState.RUNNING
    batch.started_at = timezone.now()
    batch.save(update_fields=['state', 'started_at', 'updated_at'])

    path = export_path(export)
    written = 0

    def note(rows: int) -> None:
        nonlocal written
        written = rows
        batch.processed_items = rows
        batch.save(update_fields=['processed_items', 'updated_at'])

    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        with path.open('w', encoding='utf-8', newline='') as handle:
            for line in export_csv(batch.domain, export.filters, on_progress=note):
                handle.write(line)
    except OSError as exc:
        logger.exception('No se pudo escribir la exportación %s', export.id)
        # El archivo a medio escribir no sirve para nada y sí puede confundir:
        # se descarta, y lo que queda es el lote fallido con su motivo.
        path.unlink(missing_ok=True)
        batch.state = BatchState.FAILED
        batch.finished_at = timezone.now()
        batch.summary = {'rows': 0, 'filters': export.filters, 'errors': [str(exc)]}
        batch.save(update_fields=['state', 'finished_at', 'summary', 'updated_at'])
        return export

    export.row_count = written
    export.size_bytes = path.stat().st_size
    export.expires_at = timezone.now() + timedelta(days=settings.EXPORT_RETENTION_DAYS)
    export.save(update_fields=['row_count', 'size_bytes', 'expires_at', 'updated_at'])

    batch.state = BatchState.COMPLETED
    batch.finished_at = timezone.now()
    batch.processed_items = written
    # El total se corrige con lo que el archivo tiene de verdad. Entre el pedido
    # y la escritura puede haber corrido un ciclo que descubrió URLs nuevas, y
    # un lote que dice «12.000 de 11.940» se lee como que algo salió mal.
    batch.total_items = written
    batch.summary = {'rows': written, 'filters': export.filters, 'errors': []}
    batch.save(
        update_fields=[
            'state',
            'finished_at',
            'processed_items',
            'total_items',
            'summary',
            'updated_at',
        ]
    )

    from apps.notifications import services as notifications

    notifications.export_ready(export)
    return export


def export_props(export: CoverageExport) -> dict:
    """
    Lo que una pantalla necesita para hablar de una exportación.

    El tamaño viaja porque un archivo de cien megabytes es una decisión antes de
    tocarlo, sobre todo desde una conexión medida. Y la fecha de vencimiento
    porque un botón que deja de funcionar sin haber avisado se lee como un error
    de la plataforma.

    Vive acá y no en una de las dos vistas que lo usan —la cobertura y la ficha
    del lote— porque las dos tienen que decir lo mismo del mismo archivo.
    """
    from django.urls import reverse

    errors = (export.batch.summary or {}).get('errors') or []

    return {
        'id': str(export.id),
        # El motivo del fallo viaja resuelto porque es lo único del resumen que
        # hace falta para decir qué pasó con el archivo.
        'error': errors[0] if errors else None,
        'filters': export.filters,
        'row_count': export.row_count,
        'size_bytes': export.size_bytes,
        'expires_at': export.expires_at.isoformat() if export.expires_at else None,
        'expired': export.is_expired,
        'download_path': reverse('export.download', kwargs={'export_id': export.id}),
        'requested_at': export.created_at.isoformat(),
    }


def purge_expired_exports() -> int:
    """
    Borra los archivos vencidos y devuelve cuántos.

    La fila se conserva. El archivo es lo que ocupa lugar y lo que contiene los
    datos; la constancia de qué se pidió y con qué recorte pesa nada y permite
    que la pantalla diga «venció el 20 ago» en vez de dar un 404 sin explicación.
    """
    removed = 0

    for export in CoverageExport.objects.filter(expires_at__lte=timezone.now()):
        path = export_path(export)
        if path.exists():
            path.unlink()
            removed += 1

    return removed