"""Le job `import_api` — un import de mots-clés tiré d'une API BYOK (spec Imports T8).

**Le même pipeline canonique que le CSV, sans le fichier.** Les écritures (`write_batch`,
`batches`) ne dépendent que de la cible, de la locale, de la source et du `fetched_at` : la
lecture seule change. Ce module ne réimplémente donc rien du réservoir — il fournit une autre
source de `CanonicalKeywordRow` au même moteur, ce qui garantit gratuitement l'idempotence
`(keyword_id, source, fetched_at)`, l'invariant `keyword_metrics.org_id = keywords.org_id` et
la reprise à mi-course déjà prouvés par T2.

**La clé n'est jamais dans le payload du job.** `jobs.payload` est du jsonb en clair, relu par
`GET /jobs/{id}` et lisible dans un dump : y ranger un secret annulerait le chiffrement de la
table `integrations`. Le job ne porte que l'`integration_id` ; le handler relit la ligne
(filtrée par `org_id`) et déchiffre au dernier moment.

**Tout échec du fournisseur fait échouer le job, proprement.** Clé refusée, quota atteint,
marché introuvable, format que l'adaptateur ne sait pas lire
(`mangools_api.parsing_ready()`) : l'exception remonte, le socle met le job en `failed` et
`on_fail` recopie le message dans la ligne `imports`, que l'écran affiche. Jamais un parse
hasardeux, jamais un import muet.
"""

import logging
import uuid
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from datetime import datetime
from typing import Any

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core import crypto
from app.integrations.models import Integration
from app.jobs import registry, service
from app.jobs.models import Job
from app.keywords import pipeline
from app.keywords.pipeline import ScanCounters, WriteCounters
from app.keywords.sources import mangools_api
from app.keywords.sources.base import CanonicalKeywordRow, RowError

logger = logging.getLogger(__name__)

JOB_TYPE = "import_api"

_REQUIRED_PAYLOAD = frozenset(
    {"import_id", "integration_id", "provider", "source", "site_target_id", "locale", "fetched_at"}
)


@dataclass(frozen=True, slots=True)
class ApiImportSpec:
    """Le contrat de `jobs.payload` pour `import_api`. **Aucun secret n'y figure.**"""

    import_id: uuid.UUID
    integration_id: uuid.UUID
    provider: str
    source: str
    site_target_id: uuid.UUID
    locale: str
    fetched_at: datetime
    #  Le **mode**, exclusif, miroir de `FetchSpec` : une liste enregistrée (`list_id`), un
    #  domaine (découverte) ou une liste nommée (enrichissement). Tous vides = payload d'une api
    #  antérieure au mode, que `fetch_rows` traite en découverte du domaine — d'où les défauts
    #  vides plutôt que des champs obligatoires, qui feraient échouer un job enfilé avant le
    #  déploiement.
    domain: str
    keywords: tuple[str, ...] = ()
    list_id: str = ""
    location_id: int | None = None
    limit: int = 1_000
    #  **La liste de destination, sans laquelle l'import serait invisible.** Le réservoir
    #  n'affiche un mot-clé que s'il appartient à une liste de la cible ou porte une attache
    #  d'URL sur elle (`app.keywords.service._BELONGS`) : un import qui n'écrit ni l'une ni
    #  l'autre remplit la base et laisse l'écran vide. `None` n'existe que pour les jobs
    #  enfilés avant ce correctif, qui se comporteront comme avant — mal, mais sans planter.
    keyword_list_id: uuid.UUID | None = None


def _country_of(locale: str) -> str | None:
    """Le pays ISO-2 d'une locale `langue-PAYS` (`fr-FR` → `FR`), ou `None`.

    `None` plutôt qu'une exception : une locale sans pays exploitable ne doit pas faire échouer
    l'import, elle doit seulement le priver de cloisonnement — c'est le même choix qu'un CSV
    dont la colonne pays est absente (`pipeline._country_filter`), et il est sans risque ici
    parce que `site_targets` impose déjà `^[A-Z]{2}$` à son pays.
    """
    _, _, tail = locale.partition("-")
    code = tail.strip().upper()
    return code if len(code) == 2 and code.isalpha() else None


def parse_payload(payload: Any) -> ApiImportSpec:
    """Valide le payload. Un payload invalide est une erreur de programmation de l'api."""
    if not isinstance(payload, Mapping):
        raise ValueError("payload d'import_api invalide : objet attendu")
    if missing := _REQUIRED_PAYLOAD - payload.keys():
        raise ValueError(f"payload d'import_api incomplet : {', '.join(sorted(missing))}")
    fetched_at = datetime.fromisoformat(str(payload["fetched_at"]))
    if fetched_at.tzinfo is None:
        # Même raison qu'en CSV : sans fuseau, `(keyword_id, source, fetched_at)` cesse
        # d'identifier une observation.
        raise ValueError("fetched_at doit porter un fuseau horaire")
    location = payload.get("location_id")
    destination = payload.get("keyword_list_id")
    raw_keywords = payload.get("keywords") or ()
    if not isinstance(raw_keywords, list | tuple):
        raise ValueError("payload d'import_api invalide : `keywords` doit être une liste")
    return ApiImportSpec(
        import_id=uuid.UUID(str(payload["import_id"])),
        integration_id=uuid.UUID(str(payload["integration_id"])),
        provider=str(payload["provider"]),
        source=str(payload["source"]),
        site_target_id=uuid.UUID(str(payload["site_target_id"])),
        locale=str(payload["locale"]),
        fetched_at=fetched_at,
        domain=str(payload.get("domain") or ""),
        keywords=tuple(str(term) for term in raw_keywords if str(term).strip()),
        list_id=str(payload.get("list_id") or "").strip(),
        location_id=None if location is None else int(location),
        limit=int(payload.get("limit") or 1_000),
        keyword_list_id=None if destination is None else uuid.UUID(str(destination)),
    )


async def fetch_rows(
    session: AsyncSession, org_id: str, spec: ApiImportSpec
) -> Sequence[CanonicalKeywordRow]:
    """Relit l'intégration, déchiffre la clé, appelle le fournisseur. Rend les lignes canoniques.

    Le `SELECT` est filtré par `org_id` : un job dont le payload désignerait l'intégration
    d'une autre organisation ne peut pas en lire la clé, quelle que soit la façon dont il a
    été enfilé.
    """
    integration = await session.scalar(
        select(Integration).where(
            Integration.id == spec.integration_id, Integration.org_id == org_id
        )
    )
    if integration is None:
        raise ValueError("intégration inconnue ou supprimée : import impossible")
    if integration.provider != mangools_api.SOURCE_NAME:
        raise ValueError(f"fournisseur sans adaptateur d'import : {integration.provider}")
    if not spec.domain and not spec.keywords and not spec.list_id:
        #  Aucun mode : `FetchSpec` lèverait de toute façon à la construction, mais depuis le
        #  worker, sans personne devant. Le message dit ce qui manque et où le corriger —
        #  c'est tout ce qu'`imports.error` pourra afficher.
        raise ValueError(
            "import API sans mode : choisissez une liste KWFinder, ou renseignez le domaine du "
            "site (découverte) ou une liste de mots-clés (enrichissement)."
        )
    #  Dernier moment possible, et jamais logué : la clé ne vit que le temps de l'appel.
    api_key = crypto.decrypt(integration.credentials)
    client = mangools_api.MangoolsApi(api_key)
    #  **Exclusifs, et l'ordre est celui de l'explicite vers le repli** : une liste enregistrée
    #  d'abord (c'est le seul mode que l'interface produit), puis les mots-clés nommés, et le
    #  domaine en dernier — lui seul peut n'être qu'un repli sur celui du site. En pratique
    #  `enqueue_api_import` n'écrit jamais deux modes ; cette précédence n'existe que pour un
    #  payload malformé, qui doit se comporter de façon décidée plutôt que selon l'ordre des `if`.
    if spec.list_id:
        fetch = mangools_api.FetchSpec(list_id=spec.list_id, locale=spec.locale)
    elif spec.keywords:
        fetch = mangools_api.FetchSpec(
            keywords=spec.keywords,
            locale=spec.locale,
            location_id=spec.location_id,
            limit=spec.limit,
        )
    else:
        fetch = mangools_api.FetchSpec(
            domain=spec.domain,
            locale=spec.locale,
            location_id=spec.location_id,
            limit=spec.limit,
        )
    return await client.fetch(fetch)


async def run_import_api(session: AsyncSession, job: Job) -> Mapping[str, Any]:
    """Handler du job `import_api`. Rend le rapport final, recopié dans `imports`."""
    spec = parse_payload(job.payload)
    #  `ImportSpec` sans `file_sha256` : le moteur d'écriture ne connaît que la cible, la
    #  locale, la source et l'horodatage — voir la docstring d'`ImportSpec`.
    write_spec = pipeline.ImportSpec(
        import_id=spec.import_id,
        file_sha256=None,
        source=spec.source,
        site_target_id=spec.site_target_id,
        locale=spec.locale,
        fetched_at=spec.fetched_at,
        #  Le même moteur d'écriture que le CSV, et donc la même appartenance : c'est ce champ
        #  qui alimente `keyword_list_items`, seule chose qui rende un mot-clé visible dans le
        #  réservoir de la cible. Il manquait — d'où `list_items_added: 0` sur un import réussi
        #  et un écran « Réservoir vide » (correctif 2026-09-02).
        keyword_list_id=spec.keyword_list_id,
    )

    await pipeline.set_import_status(session, spec.import_id, job.org_id, "running")
    await session.commit()

    rows = await fetch_rows(session, job.org_id, spec)

    #  **Le cloisonnement pays, et il ne concerne qu'un mode sur trois.** Une liste enregistrée
    #  se lit telle qu'elle a été constituée, marchés mélangés compris ; les deux autres modes
    #  ont passé un `location_id` en entrée et n'ont donc rien à trier. Ce qui ne correspond pas
    #  à la cible est écarté et **compté** (`rows_off_target`), exactement comme sur un CSV
    #  KWFinder à onze pays — jamais importé en silence, jamais compté comme une erreur.
    wanted = pipeline.api_country_filter(_country_of(spec.locale)) if spec.list_id else None

    scan = ScanCounters()
    written = WriteCounters.from_dict(
        job.progress.get("counters") if isinstance(job.progress, Mapping) else None
    )
    errors: list[RowError] = []
    #  Pas de reprise par curseur ici : un fetch API n'est pas rejouable ligne à ligne (la
    #  réponse du fournisseur n'est pas un fichier stable, et la rejouer coûterait du quota).
    #  L'idempotence métier suffit : un job repris refait le fetch et réécrit exactement les
    #  mêmes observations `(keyword_id, source, fetched_at)`, donc n'en duplique aucune.
    for index, batch in enumerate(pipeline.batches(rows, scan, errors, wanted)):
        await pipeline.write_batch(session, job, write_spec, batch, written)
        alive = await service.save_progress(
            session,
            job,
            {"batch": index + 1, "rows_read": scan.rows_read, "counters": written.as_dict()},
        )
        if not alive:
            logger.warning("import API %s : job repris par un autre worker, arrêt", spec.import_id)
            return {"dispossessed": True, **pipeline.build_report(scan, written, errors, None)}

    report = pipeline.build_report(scan, written, errors, None)
    await pipeline.set_import_status(
        session, spec.import_id, job.org_id, "done", report=dict(report)
    )
    await session.commit()
    return report


registry.register(
    registry.JobType(
        JOB_TYPE,
        run_import_api,
        #  `'active'`, comme `import_csv` : un import est long et exclusif. La clé
        #  `import_api:<intégration>:<cible>:<période>` n'est relâchée qu'à `done`/`failed`,
        #  donc le bouton « Importer maintenant » cliqué deux fois ne lance qu'un import, et
        #  un tick de planificateur qui doublerait ne produit rien de plus.
        dedup_scope="active",
        #  Un fetch API échoue surtout pour des raisons **déterministes** (clé refusée, format
        #  non capturé) : une seule tentative, un message clair, et l'utilisateur décide.
        #  Rejouer trois fois brûlerait du quota pour le même échec.
        max_attempts=1,
        on_fail=pipeline.mark_import_failed,
    )
)