"""Intégrations BYOK : CRUD, déclenchement d'import et sweep des synchros dues (spec T8).

Trois propriétés portent ce module :

1. **La clé ne sort jamais.** Elle entre chiffrée (`crypto.encrypt`), n'est relue que par le
   job et la vérification, et rien ici ne la met dans une réponse, un log ou un payload de job.
2. **Un import est publié d'un seul commit** — ligne `imports` + job — comme le CSV
   (`imports/service.create_import`) : un arrêt entre deux commits laisserait sinon un job qui
   peuple le réservoir et *rien* dans l'onglet Imports.
3. **Le sweep ne double jamais un import.** Deux verrous concentriques : la ligne `schedule`
   est prise `FOR UPDATE SKIP LOCKED` et son `next_run_at` avance dans la même transaction que
   l'enfilage (deux workers ne peuvent pas prendre le même rendez-vous), et la `dedup_key`
   `import_api:<intégration>:<cible>:<période>` est l'arbitre final côté file — celui que la
   spec nomme. Un tick raté se rattrape au tick suivant parce que le sweep ne regarde que
   `next_run_at <= now`, jamais l'horloge d'un cron.
"""

import asyncio
import logging
import time
import uuid
from collections.abc import Mapping, Sequence
from datetime import UTC, datetime
from typing import Any

from sqlalchemy import func, select
from sqlalchemy import inspect as sa_inspect
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from app.core import crypto
from app.core.errors import AppError
from app.imports.schemas import CountryCount
from app.imports.service import get_target, locale_of
from app.integrations import schedule as cadence
from app.integrations.models import ImportSchedule, Integration
from app.integrations.pipeline import JOB_TYPE
from app.integrations.schemas import (
    IntegrationCreate,
    IntegrationUpdate,
    ListPreview,
    ScheduleUpsert,
)
from app.jobs import registry
from app.jobs import service as jobs_service
from app.jobs.models import Job
from app.keywords import countries
from app.keywords.models import Import, KeywordList
from app.keywords.sources import mangools_api
from app.sites.models import Site, SiteTarget

logger = logging.getLogger(__name__)

# Les fournisseurs dont l'api sait vérifier la clé et lancer un import. `provider` est déjà
# contraint en base ; ce dictionnaire dit *ce qu'on sait en faire*.
_CLIENTS = {mangools_api.SOURCE_NAME: mangools_api.MangoolsApi}

# Plafond de l'appel de vérification, **volontairement sous les 10 s d'`apiFetch`**
# (`web/lib/api.ts`). C'est un invariant de frontière, et il vit ici parce que c'est ici qu'on
# choisit combien de temps attendre un tiers : si le fournisseur traîne plus longtemps que le
# web n'attend l'api, le `fetch` de la Server Action avorte sur un `TimeoutError` — qui n'est
# pas une `ApiError`, donc que `relay` relance, donc qui fait tomber tout l'écran Intégrations
# dans son `error.tsx`. L'utilisateur verrait un crash là où la promesse de `check_integration`
# est un badge « En erreur ». Le défaut de 30 s du client reste bon pour un import, qui tourne
# dans le worker et n'a pas de navigateur au bout.
CHECK_TIMEOUT_SECONDS = 6.0

# Plafond des lectures de listes, **mesuré et non deviné** : `GET /kwfinder/lists/{id}` a mis
# 17,68 s pour la plus grosse liste du compte relevé (2106 mots-clés, 1,9 Mo, le 2026-09-02).
# Le timeout de vérification (6 s) coupait donc cette liste-là avant la fin, et l'utilisateur ne
# pouvait pas l'importer du tout — un 502 « injoignable » sur sa liste principale. On garde
# deux fois et demie la mesure : ces appels ne coûtent aucun quota, et l'appelant web les attend
# déjà jusqu'à 60 s (`kwfinderListsPreviewAction`), sous le plafond d'exécution du segment.
LIST_TIMEOUT_SECONDS = 45.0

# Durée de vie du recensement des listes d'un compte, **par processus**. `GET /kwfinder/lists`
# est gratuit mais il compte dans le débit (3 requêtes par courte période) : le refaire à chaque
# cible d'un lancement multi-cibles brûlait le budget avant même le premier import. Le contenu
# ne bouge que quand l'utilisateur touche à son compte KWFinder ; cinq minutes de mémoire ne
# peuvent que servir un nom déjà juste.
# Échéance **globale** du recensement d'une sélection, et elle est distincte du plafond d'une
# lecture : `LIST_TIMEOUT_SECONDS` borne *une* liste, or `preview_lists` en enchaîne jusqu'à
# `LIST_IDS_MAX`, espacées du débit du fournisseur. Quatre listes de la taille mesurée font déjà
# ~81 s — plus que les 60 s qu'attend la Server Action, qui abandonnait alors sur un « réessayez »
# que l'utilisateur ne pouvait pas suivre : le réessai reprenait des 429. On répond donc **avant**
# le web, et on dit quoi faire (cocher moins de listes) plutôt que de le laisser deviner.
PREVIEW_BUDGET_SECONDS = 50.0
_LISTS_TTL_SECONDS = 300.0
_LISTS_CACHE: dict[uuid.UUID, tuple[float, list[mangools_api.KeywordListSummary]]] = {}


# --- CRUD ---------------------------------------------------------------------------------


async def get_integration(
    session: AsyncSession, org_id: str, integration_id: uuid.UUID
) -> Integration:
    found = await session.scalar(
        select(Integration).where(Integration.id == integration_id, Integration.org_id == org_id)
    )
    if found is None:
        # 404 et non 403 : une intégration d'une autre org n'existe pas (architecture §9).
        raise AppError(404, "Not Found", "Intégration inconnue")
    return found


async def list_integrations(session: AsyncSession, org_id: str) -> list[Integration]:
    #  Pas de pagination : une organisation a autant d'intégrations que de fournisseurs
    #  supportés, c'est-à-dire un, bientôt deux. Un curseur serait du décor.
    return list(
        (
            await session.scalars(
                select(Integration)
                .where(Integration.org_id == org_id)
                .order_by(Integration.provider)
            )
        ).all()
    )


async def create_integration(
    session: AsyncSession, org_id: str, payload: IntegrationCreate
) -> Integration:
    key = payload.api_key.strip()
    if not key:
        raise AppError(422, "Validation error", "La clé est vide")
    integration = Integration(
        org_id=org_id,
        provider=payload.provider,
        credentials=crypto.encrypt(key),
        credentials_hint=crypto.hint(key),
        config=dict(payload.config),
        status="unverified",
    )
    session.add(integration)
    try:
        await session.commit()
    except IntegrityError as exc:
        #  L'UNIQUE (org_id, provider) est l'arbitre, pas un pré-check : deux onglets qui
        #  enregistrent en même temps ne doivent pas créer deux clés pour le même fournisseur.
        await session.rollback()
        raise AppError(
            409,
            "Conflict",
            f"Une intégration « {payload.provider} » existe déjà : modifiez-la plutôt.",
        ) from exc
    return integration


async def update_integration(
    session: AsyncSession, org_id: str, integration_id: uuid.UUID, payload: IntegrationUpdate
) -> Integration:
    integration = await get_integration(session, org_id, integration_id)
    if payload.api_key is not None:
        key = payload.api_key.strip()
        if not key:
            raise AppError(422, "Validation error", "La clé est vide")
        integration.credentials = crypto.encrypt(key)
        integration.credentials_hint = crypto.hint(key)
        #  Nouvelle clé = statut remis à zéro : l'ancien « connectée » parlait d'une autre clé
        #  et laisserait croire que celle-ci a été vérifiée.
        integration.status = "unverified"
        integration.status_error = None
        integration.checked_at = None
    if payload.config is not None:
        integration.config = dict(payload.config)
    await session.commit()
    #  La clé a pu changer : les listes mémorisées sont celles d'un autre compte.
    forget_account_lists(integration_id)
    return integration


async def delete_integration(session: AsyncSession, org_id: str, integration_id: uuid.UUID) -> None:
    integration = await get_integration(session, org_id, integration_id)
    #  CASCADE emporte les plannings ; les imports déjà faits et le réservoir restent
    #  (rétention org, décision n°4) — débrancher une clé n'efface pas des données payées.
    await session.delete(integration)
    await session.commit()
    forget_account_lists(integration_id)


async def check_integration(
    session: AsyncSession, org_id: str, integration_id: uuid.UUID
) -> Integration:
    """Confronte la clé au fournisseur et enregistre le verdict (§1.11 « statut »).

    L'appel de quotas est le seul qui ne demande aucune connaissance du format de réponse :
    il donne un statut honnête bien avant que l'adaptateur de fetch sache lire quoi que ce
    soit. Un refus n'est pas une erreur HTTP de *notre* api — c'est un état de l'intégration :
    on rend 200 avec `status='error'` et le motif, que l'écran affiche. **Un fournisseur qui ne
    répond pas en est un aussi** : d'où `CHECK_TIMEOUT_SECONDS`, qui garantit qu'on rend
    toujours ce 200 avant que le navigateur n'ait renoncé.
    """
    integration = await get_integration(session, org_id, integration_id)
    client_class = _CLIENTS.get(integration.provider)
    if client_class is None:
        raise AppError(
            422, "Validation error", f"Fournisseur non vérifiable : {integration.provider}"
        )
    client = client_class(crypto.decrypt(integration.credentials), timeout=CHECK_TIMEOUT_SECONDS)
    try:
        await client.check()
    except mangools_api.MangoolsApiError as exc:
        integration.status = "error"
        integration.status_error = str(exc)
    else:
        integration.status = "connected"
        integration.status_error = None
    integration.checked_at = datetime.now(UTC)
    await session.commit()
    return integration


async def read_quota(
    session: AsyncSession, org_id: str, integration_id: uuid.UUID
) -> tuple[int | None, int | None, str | None]:
    """`(plafond, restant, motif d'échec)` du quota du fournisseur. **Ne lève pas** pour un tiers.

    Le même appel que `check_integration` — `GET /kwfinder/limits`, gratuit — mais **sans écrire
    quoi que ce soit** : c'est le corps d'un `GET`, et un `GET` qui repositionnerait le statut de
    l'intégration ferait de l'ouverture d'un récapitulatif une modification. Le statut a son
    endpoint, celui-ci n'a que le quota.

    Un fournisseur qui refuse ou se tait rend un motif, pas une erreur HTTP : le récapitulatif
    doit pouvoir dire « quota indisponible » et laisser lancer l'import quand même — la clé peut
    très bien marcher pour l'import et l'appel de quotas avoir raté sur un aléa réseau.
    """
    integration = await get_integration(session, org_id, integration_id)
    client_class = _CLIENTS.get(integration.provider)
    if client_class is None:
        return None, None, f"Fournisseur sans lecture de quota : {integration.provider}"
    client = client_class(crypto.decrypt(integration.credentials), timeout=CHECK_TIMEOUT_SECONDS)
    try:
        payload = await client.check()
    except mangools_api.MangoolsApiError as exc:
        return None, None, str(exc)
    limit, remaining = mangools_api.resource_quota(payload)
    return limit, remaining, None


# --- listes enregistrées du compte KWFinder --------------------------------------------------


async def _client(
    session: AsyncSession,
    org_id: str,
    integration_id: uuid.UUID,
    *,
    #  `timeout_seconds` et non `timeout` : ce n'est pas un délai d'annulation de la coroutine
    #  (ce que `ASYNC109` cherche) mais le plafond du client HTTP construit ici.
    timeout_seconds: float = LIST_TIMEOUT_SECONDS,
) -> Any:
    """Le client du fournisseur d'une intégration, clé déchiffrée au dernier moment.

    Le plafond par défaut est celui d'une **lecture de liste**, pas celui de la vérification de
    clé : une grosse liste met une vingtaine de secondes à revenir (voir `LIST_TIMEOUT_SECONDS`),
    et la couper à six ferait échouer précisément les listes que l'utilisateur veut importer.
    L'appelant peut le raccourcir quand son propre appelant, lui, n'attend pas si longtemps.
    """
    integration = await get_integration(session, org_id, integration_id)
    client_class = _CLIENTS.get(integration.provider)
    if client_class is None:
        raise AppError(422, "Validation error", f"Fournisseur sans listes : {integration.provider}")
    return client_class(crypto.decrypt(integration.credentials), timeout=timeout_seconds)


async def _account_lists(
    session: AsyncSession, org_id: str, integration_id: uuid.UUID
) -> list[mangools_api.KeywordListSummary]:
    """Les listes du compte, **mémorisées quelques minutes** (voir `_LISTS_TTL_SECONDS`).

    L'appartenance de l'intégration à l'organisation est vérifiée **avant** de consulter la
    mémoire (`_client` appelle `get_integration`) : un identifiant d'une autre organisation
    reste un 404, mémoire chaude ou pas.
    """
    #  Le **listing** du compte est court (des identifiants et des noms) et il est appelé depuis
    #  le rendu d'une page : son plafond reste celui de la vérification, sous les dix secondes
    #  d'`apiFetch`. Seule la lecture d'une liste entière a besoin des 45 s.
    client = await _client(session, org_id, integration_id, timeout_seconds=CHECK_TIMEOUT_SECONDS)
    cached = _LISTS_CACHE.get(integration_id)
    if cached is not None and time.monotonic() - cached[0] < _LISTS_TTL_SECONDS:
        return cached[1]
    try:
        items = await client.lists()
    except mangools_api.MangoolsApiError as exc:
        raise _as_app_error(exc) from exc
    _LISTS_CACHE[integration_id] = (time.monotonic(), items)
    return items


def forget_account_lists(integration_id: uuid.UUID) -> None:
    """Oublie les listes mémorisées d'une intégration — clé remplacée, intégration supprimée."""
    _LISTS_CACHE.pop(integration_id, None)


def _as_app_error(exc: mangools_api.MangoolsApiError) -> AppError:
    """Un refus du fournisseur en Problem Details lisible. **Jamais un 500.**

    Une clé refusée ou un quota atteint ne sont pas des pannes de notre api : ce sont des
    états que l'écran doit pouvoir nommer et que l'utilisateur peut corriger.
    """
    return AppError(502, "Bad Gateway", str(exc))


async def list_keyword_lists(
    session: AsyncSession, org_id: str, integration_id: uuid.UUID
) -> list[mangools_api.KeywordListSummary]:
    """Les listes déjà enregistrées dans le compte du fournisseur. **Aucun quota consommé.**"""
    return await _account_lists(session, org_id, integration_id)


async def preview_lists(
    session: AsyncSession,
    org_id: str,
    integration_id: uuid.UUID,
    list_ids: Sequence[str],
    site_target_id: uuid.UUID | None,
) -> tuple[list[ListPreview], list[CountryCount]]:
    """Le recensement des marchés d'une sélection de listes (§4.2 du chemin CSV, transposé).

    **Toute la sélection en un seul aller-retour**, et c'est la raison d'être de cet endpoint :
    l'espacement des requêtes (`_NEXT_REQUEST_AT`) est tenu côté api, donc un recensement
    découpé en N appels depuis le navigateur n'aurait rien gagné et aurait multiplié les
    latences. Une grosse liste met une vingtaine de secondes : dix listes cochées font une
    attente qui frôle le plafond d'exécution du segment web — voir `LIST_IDS_MAX`.

    Les identifiants inconnus sont refusés **ici**, une fois, plutôt que de partir en job : le
    listing du compte est gratuit, et un 404 découvert dans le worker n'aurait plus personne
    devant pour le lire.

    L'enchaînement est borné par `PREVIEW_BUDGET_SECONDS` : au-delà, l'api rend un 504 qui
    **dit quoi faire**, au lieu de laisser la Server Action abandonner sur un motif générique
    pendant que l'api continue à consommer le débit du fournisseur.
    """
    client = await _client(session, org_id, integration_id)
    target = None if site_target_id is None else await get_target(session, org_id, site_target_id)
    known = {
        summary.id: summary for summary in await _account_lists(session, org_id, integration_id)
    }
    unknown = [identifier for identifier in list_ids if identifier not in known]
    if unknown:
        raise AppError(
            404,
            "Not Found",
            f"Liste KWFinder inconnue sur ce compte : {', '.join(unknown[:5])}",
        )
    previews: list[ListPreview] = []
    #  Un dictionnaire par libellé **brut**, agrégé sur toute la sélection : deux listes qui
    #  portent « United States » comptent leurs lignes ensemble.
    merged: dict[str, int] = {}
    try:
        async with asyncio.timeout(PREVIEW_BUDGET_SECONDS):
            for identifier in list_ids:
                payloads = await client.list_keywords(identifier)
                counted: dict[str, int] = {}
                for payload in payloads:
                    label = mangools_api.row_country_label(payload)
                    if not label:
                        #  Pas de libellé : rien à nommer à l'écran ni à proposer comme cible.
                        #  La ligne sera comptée `rows_country_unresolved` par le rapport, qui
                        #  compte des lignes et non des libellés — même règle qu'en CSV.
                        continue
                    counted[label] = counted.get(label, 0) + 1
                    merged[label] = merged.get(label, 0) + 1
                previews.append(
                    ListPreview(
                        list_id=identifier,
                        name=known[identifier].name,
                        keywords_total=len(payloads),
                        countries=await _census_countries(session, org_id, target, counted),
                    )
                )
    except TimeoutError as exc:
        #  `504` et non `502` : le tiers n'a rien refusé, c'est **notre** budget qui est atteint.
        raise AppError(
            504,
            "Gateway Timeout",
            f"Recensement trop long ({len(previews)} liste(s) sur {len(list_ids)} lues) : "
            "cochez moins de listes.",
        ) from exc
    except mangools_api.MangoolsApiError as exc:
        raise _as_app_error(exc) from exc
    return previews, await _census_countries(session, org_id, target, merged)


async def _census_countries(
    session: AsyncSession,
    org_id: str,
    target: SiteTarget | None,
    counted: Mapping[str, int],
) -> list[CountryCount]:
    """`{libellé brut: lignes}` → le recensement de l'écran, trié par lignes décroissantes.

    Le pendant du `_census` du chemin CSV (`app.imports.service`), sur un décompte déjà fait :
    la résolution ISO, le repli de libellé et les cibles existantes suivent exactement les
    mêmes règles, donc l'écran réutilise le même composant et le même miroir Zod.
    """
    if not counted:
        return []
    existing: dict[str, list[uuid.UUID]] = {}
    if target is not None:
        rows = await session.execute(
            select(SiteTarget.id, SiteTarget.country).where(
                SiteTarget.site_id == target.site_id, SiteTarget.org_id == org_id
            )
        )
        for target_id, country in rows:
            existing.setdefault(country, []).append(target_id)
    census: list[CountryCount] = []
    for raw, rows_count in sorted(counted.items(), key=lambda item: (-item[1], item[0])):
        code = countries.resolve(raw)
        census.append(
            CountryCount(
                raw=raw,
                code=code,
                #  Jamais nul, par contrat avec le miroir Zod : le libellé brut fait office de
                #  nom quand il n'y a pas de pays à nommer (« New York »).
                label=raw if code is None else countries.label(code),
                rows=rows_count,
                suggested_language=None if code is None else countries.suggested_language(code),
                target_ids=[] if code is None else existing.get(code, []),
            )
        )
    return census


# --- plannings ----------------------------------------------------------------------------


async def list_schedules(
    session: AsyncSession, org_id: str, integration_id: uuid.UUID
) -> list[ImportSchedule]:
    await get_integration(session, org_id, integration_id)
    return list(
        (
            await session.scalars(
                select(ImportSchedule)
                .where(
                    ImportSchedule.integration_id == integration_id,
                    ImportSchedule.org_id == org_id,
                )
                .order_by(ImportSchedule.id)
            )
        ).all()
    )


async def upsert_schedule(
    session: AsyncSession, org_id: str, integration_id: uuid.UUID, payload: ScheduleUpsert
) -> ImportSchedule:
    """Pose ou change le mode de synchro d'une cible. Idempotent.

    Le `next_run_at` n'est **recalculé que si la cadence change** : rebasculer `enabled` ou
    ajuster la configuration ne doit pas repousser une échéance déjà prise — sinon un
    utilisateur qui ouvre son écran chaque semaine repousserait sa synchro à l'infini sans
    jamais rien voir arriver.
    """
    integration = await get_integration(session, org_id, integration_id)
    target = await get_target(session, org_id, payload.site_target_id)
    now = datetime.now(UTC)

    existing = await session.scalar(
        select(ImportSchedule).where(
            ImportSchedule.integration_id == integration.id,
            ImportSchedule.site_target_id == target.id,
            ImportSchedule.org_id == org_id,
        )
    )
    if existing is None:
        record = ImportSchedule(
            org_id=org_id,
            integration_id=integration.id,
            site_target_id=target.id,
            sync_mode=payload.sync_mode,
            next_run_at=cadence.first_run_at(payload.sync_mode, now),
            #  Champ absent du corps = défaut du modèle, pas `None` : la colonne est NOT NULL.
            enabled=True if payload.enabled is None else payload.enabled,
            config=dict(payload.config or {}),
        )
        session.add(record)
        try:
            await session.commit()
        except IntegrityError as exc:
            await session.rollback()
            raise AppError(
                409, "Conflict", "Un mode de synchro existe déjà pour cette cible"
            ) from exc
        return record

    if existing.sync_mode != payload.sync_mode:
        existing.sync_mode = payload.sync_mode
        existing.next_run_at = cadence.first_run_at(payload.sync_mode, now)
    #  Champ absent = inchangé (voir `ScheduleUpsert`) : changer de cadence ne remet à zéro ni
    #  l'interrupteur, ni les réglages de la cible.
    if payload.enabled is not None:
        existing.enabled = payload.enabled
    if payload.config is not None:
        existing.config = dict(payload.config)
    await session.commit()
    return existing


async def delete_schedule(
    session: AsyncSession, org_id: str, integration_id: uuid.UUID, schedule_id: uuid.UUID
) -> None:
    record = await session.scalar(
        select(ImportSchedule).where(
            ImportSchedule.id == schedule_id,
            ImportSchedule.integration_id == integration_id,
            ImportSchedule.org_id == org_id,
        )
    )
    if record is None:
        raise AppError(404, "Not Found", "Planning inconnu")
    await session.delete(record)
    await session.commit()


async def last_imports(
    session: AsyncSession, org_id: str, target_ids: list[uuid.UUID] | None, source: str
) -> dict[uuid.UUID, Import]:
    """Le dernier import API par cible — le « dernier rapport » de l'écran (§1.11).

    Une seule requête pour toutes les cibles, `DISTINCT ON` : l'écran en affiche autant qu'il
    y a de cibles, et une requête par ligne ferait un N+1 sur le chemin le plus visité de la
    page.

    `target_ids=None` veut dire « toutes les cibles de l'organisation », et c'est ce dont
    l'écran a besoin depuis que le dernier rapport ne passe plus par les plannings : une cible
    qui n'a **jamais** eu de planning a très bien pu recevoir un import manuel, et son rapport
    doit s'afficher. La liste explicite reste pour l'appelant qui sait déjà quelles cibles il
    montre. Dans les deux cas le filtre `org_id` est là, seul et suffisant.
    """
    if target_ids is not None and not target_ids:
        return {}
    query = select(Import).where(
        Import.org_id == org_id,
        Import.kind == "api",
        Import.source == source,
    )
    if target_ids is not None:
        query = query.where(Import.site_target_id.in_(target_ids))
    rows = (
        await session.scalars(
            query.distinct(Import.site_target_id)
            #  `id` est un uuidv7 : décroissant, c'est « le plus récent » sans colonne de tri
            #  supplémentaire, et c'est l'ordre qu'exige `DISTINCT ON`.
            .order_by(Import.site_target_id, Import.id.desc())
        )
    ).all()
    return {row.site_target_id: row for row in rows}


# --- enfilage d'un import API ---------------------------------------------------------------


def _location_id(settings: Mapping[str, Any]) -> int | None:
    """`config.location_id` en entier, ou un 422 explicite. **Jamais un `int()` nu.**

    `config` est du jsonb : il a pu être écrit avant que `validate_config` n'existe, ou par une
    restauration, ou à la main. Un `int("abc")` ici lèverait un `ValueError` **hors** de tout
    contexte utilisateur — dans le sweep du worker, où il empoisonnait le rendez-vous fautif et
    affamait tous les suivants, toutes organisations confondues.

    Un `AppError` 422 dit la même chose des deux côtés : sur le déclenchement manuel c'est un
    message que l'écran affiche ; dans le sweep c'est une exception attrapée par la garde
    d'itération, qui loge le rendez-vous fautif, avance son aiguille et passe au suivant.
    """
    raw = settings.get("location_id")
    if raw is None:
        return None
    if isinstance(raw, bool) or not isinstance(raw, int):
        raise AppError(
            422,
            "Validation error",
            f"config.location_id doit être un entier, reçu {raw!r} — corrigez le réglage de "
            "cette cible.",
        )
    return raw


async def _target_domain(session: AsyncSession, target: SiteTarget) -> str:
    domain = await session.scalar(select(Site.domain).where(Site.id == target.site_id))
    if domain is None:  # pragma: no cover - la FK l'interdit
        raise AppError(404, "Not Found", "Site inconnu")
    return domain


async def _destination_list(
    session: AsyncSession, org_id: str, target_id: uuid.UUID, provider: str, name: str
) -> uuid.UUID:
    """La liste de destination d'un import API : **retrouvée par son nom, ou créée**.

    ## Pourquoi elle est obligatoire

    Le réservoir n'affiche un mot-clé que s'il est membre d'une liste de la cible ou porteur
    d'une attache d'URL sur elle (`app.keywords.service._BELONGS`). Le chemin API n'en créait
    aucune : il écrivait 89 mots-clés et 89 métriques, rendait un rapport parfait, et l'écran
    disait « Réservoir vide » — des données payées, en base, invisibles pour toujours. Aucun
    import ne doit pouvoir écrire sans appartenance.

    ## Pourquoi par le nom, et pourquoi une seule par nom

    La dette notée au pre-prod disait le risque : une liste neuve par exécution ferait
    cinquante-deux listes identiques par an. Réponse actée le 2026-09-02 : **une liste Mangools
    est une liste chez nous**, portant son nom sur la cible choisie, et un réimport la
    *rafraîchit* au lieu d'en fabriquer une seconde. L'`UNIQUE (site_target_id, name)` en fait
    la règle plutôt qu'un usage : deux imports partis ensemble ne peuvent pas créer deux listes.

    Le rattachement par identifiant Mangools serait plus stable qu'un nom (une liste renommée
    chez le fournisseur devient une seconde liste chez nous), mais il demanderait une colonne,
    donc une migration : le nom fait le travail sans elle, et une liste renommée reste un cas
    que l'utilisateur voit et peut corriger.
    """
    picked = select(KeywordList.id).where(
        KeywordList.site_target_id == target_id,
        KeywordList.org_id == org_id,
        KeywordList.name == name,
    )
    found = await session.scalar(picked)
    if found is not None:
        return found
    created = KeywordList(org_id=org_id, site_target_id=target_id, name=name, source=provider)
    try:
        #  **Un point de sauvegarde, pas un rollback.** La transaction en cours porte déjà
        #  l'avancement du rendez-vous quand c'est le sweep qui appelle : la faire tomber pour
        #  une course sur le nom laisserait le planning éternellement dû.
        async with session.begin_nested():
            session.add(created)
            await session.flush()
    except IntegrityError:
        again = await session.scalar(picked)
        if again is None:  # pragma: no cover - l'UNIQUE est la seule cause possible
            raise
        return again
    return created.id


async def enqueue_api_import(
    session: AsyncSession,
    org_id: str,
    integration: Integration,
    target: SiteTarget,
    *,
    period: str,
    config: dict[str, Any] | None = None,
    domain: str | None = None,
    keywords: Sequence[str] | None = None,
    list_id: str | None = None,
    list_name: str | None = None,
) -> tuple[Import, Job, bool] | None:
    """Crée la ligne `imports` et le job `import_api`, publiés **par un seul commit**.

    Rend `None` quand la `dedup_key` est déjà prise : un import de la même période pour la
    même cible **et la même liste** est déjà en vol, et le doubler ne produirait que des
    réécritures des mêmes observations. Ce n'est pas une erreur — c'est la garantie demandée.

    `list_id`, `domain` et `keywords` portent le **mode** (voir `schemas.ImportTrigger`) et sont
    exclusifs. Aucun des trois : découverte du domaine du site, ce que faisait cette fonction
    avant qu'ils n'existent — le sweep des plannings n'a donc rien à changer.

    **Tout mode a une liste de destination, ou il est refusé** (`_destination_list`) : une liste
    KWFinder porte son nom, une découverte porte le domaine interrogé, et l'enrichissement par
    mots-clés — qui n'a pas de nom stable — est refusé plutôt que d'écrire dans le vide.
    """
    #  **Tout est capturé en variables locales avant le commit.** Un `rollback()` expire les
    #  attributs des instances ORM, et les relire ensuite déclencherait un chargement paresseux
    #  impossible en asynchrone (`MissingGreenlet`) — jusque dans le `logger.info` du chemin
    #  d'échec, qui planterait alors à la place du no-op qu'il documente.
    integration_id = integration.id
    provider = integration.provider
    target_id = target.id
    target_country = target.country
    locale = locale_of(target)
    settings = dict(integration.config)
    settings.update(config or {})
    wanted = tuple(keywords or ())
    chosen_list = (list_id or "").strip()
    #  **Ni la liste ni l'enrichissement n'ont de domaine, et n'en fabriquent pas.** Le repli sur
    #  le domaine du site n'est calculé que dans la dernière branche : sinon le payload porterait
    #  deux modes, et `fetch_rows` choisirait par l'ordre de ses `if` plutôt que par ce qui a été
    #  demandé — exactement l'ambiguïté que `ImportTrigger` refuse à l'entrée.
    chosen_domain = (
        ""
        if wanted or chosen_list
        else str(domain or settings.get("domain") or await _target_domain(session, target))
    )
    #  Lu **avant** toute écriture : un réglage illisible doit refuser l'import, pas le laisser
    #  s'enfiler à moitié construit.
    location_id = _location_id(settings)
    #  **Le nom de la liste de destination, décidé ici, une fois pour toutes.** Il vaut aussi
    #  garde de mode : ce qui n'a pas de destination nommable n'est pas enfilé du tout, plutôt
    #  que de partir en job et d'échouer dans le worker, loin de qui a cliqué.
    if chosen_list:
        list_label = (list_name or "").strip() or f"Liste {chosen_list}"
    elif wanted:
        raise AppError(
            422,
            "Validation error",
            "L'enrichissement par mots-clés n'a pas de liste de destination : importez une "
            "liste KWFinder.",
        )
    elif chosen_domain:
        list_label = chosen_domain
    else:
        raise AppError(
            422,
            "Validation error",
            "Import API sans mode : choisissez une liste KWFinder, ou renseignez le domaine du "
            "site.",
        )
    destination_id = await _destination_list(session, org_id, target_id, provider, list_label)
    #  **La liste entre dans la clé de dédup**, et il le faut : « un import par (liste, cible) »
    #  veut dire N imports simultanés sur la même cible et la même période. Sans elle, la
    #  deuxième liste cochée retomberait sur l'import de la première et l'utilisateur verrait un
    #  seul rapport pour trois listes, sans un mot. La forme sans liste est laissée intacte pour
    #  les deux modes dormants et pour le sweep, dont la clé est celle que nomme la spec.
    dedup_key = (
        f"{JOB_TYPE}:{integration_id}:{target_id}:{period}"
        if not chosen_list
        else f"{JOB_TYPE}:{integration_id}:{target_id}:list:{chosen_list}:{period}"
    )
    fetched_at = datetime.now(UTC)

    #  L'identifiant est demandé à Postgres, pas tiré en Python : `uuidv7()` est
    #  temporellement croissant, ce dont dépend l'ordre du listing des imports.
    raw_id = await session.scalar(select(func.uuidv7()))
    assert raw_id is not None
    import_id = uuid.UUID(str(raw_id))

    job = await jobs_service.enqueue_pending(  # insère sans committer : voir la docstring
        session,
        org_id,
        JOB_TYPE,
        payload={
            "import_id": str(import_id),
            "integration_id": str(integration_id),
            "provider": provider,
            #  Nom de source partagé avec le CSV (décision 2026-09-01) : même fournisseur,
            #  même provenance, un seul rang dans la consolidation.
            "source": provider,
            "site_target_id": str(target_id),
            "locale": locale,
            "fetched_at": fetched_at.isoformat(),
            "domain": chosen_domain,
            #  Absent en découverte, plutôt qu'une liste vide : `jobs.payload` est relu par
            #  `GET /jobs/{id}` et un `"keywords": []` s'y lirait comme « un enrichissement de
            #  rien », ce qui n'est pas ce qui a été demandé. Même règle pour `list_id`.
            **({"keywords": list(wanted)} if wanted else {}),
            **({"list_id": chosen_list} if chosen_list else {}),
            **({"location_id": location_id} if location_id is not None else {}),
            #  L'appartenance à la cible, ce qui rend les mots-clés visibles dans le réservoir.
            "keyword_list_id": str(destination_id),
        },
    )
    #  Posée sur le job encore invisible : l'UNIQUE de `jobs.dedup_key` est ce qui tranche la
    #  course entre deux workers ou deux clics, dans la même transaction que la ligne
    #  `imports` — pas un pré-check, qui laisserait passer les deux gagnants d'une égalité.
    job.dedup_key = dedup_key

    record = Import(
        id=import_id,
        org_id=org_id,
        site_target_id=target_id,
        source=provider,
        kind="api",
        #  Pas de fichier : `file_sha256` n'a pas de sens ici, et `detected_format` porte le
        #  fournisseur. `filename` est l'**étiquette d'affichage** de l'onglet Imports (même
        #  rôle que le nom de fichier d'un import CSV) : en mode liste elle porte le nom de la
        #  liste KWFinder, sans quoi trois imports simultanés sur la même cible seraient trois
        #  lignes indiscernables. `detected_country` dit le pays **retenu**, celui de la cible,
        #  puisque c'est lui qui a servi à écarter le reste.
        filename=(list_name or None) if chosen_list else None,
        detected_format=provider,
        detected_country=target_country if chosen_list else None,
        fetched_at=fetched_at,
        #  La liste de destination — la même d'une exécution à l'autre, retrouvée par son nom
        #  (`_destination_list`). C'est elle qui rattache les mots-clés à la cible, donc la
        #  seule chose qui les rende visibles dans le réservoir.
        keyword_list_id=destination_id,
        job_id=job.id,
        status="queued",
    )
    session.add(record)
    try:
        #  **Le** commit : job et ligne `imports` deviennent visibles ensemble.
        await session.commit()
    except IntegrityError:
        #  `dedup_key` déjà prise : un import de cette période est en vol. No-op, pas d'erreur.
        await session.rollback()
        logger.info(
            "import API déjà en vol pour l'intégration %s / cible %s (période %s)",
            integration_id,
            target_id,
            period,
        )
        return None
    return record, job, True


async def trigger_import(
    session: AsyncSession,
    org_id: str,
    integration_id: uuid.UUID,
    target_id: uuid.UUID,
    *,
    domain: str | None = None,
    keywords: Sequence[str] | None = None,
    list_ids: Sequence[str] | None = None,
) -> list[tuple[Import, Job, bool, str | None, str | None]]:
    """« Importer maintenant » (§1.11). **Un import par (liste, cible)**, ou un seul sans liste.

    Rend, pour chacun, `(import, job, créé, id de liste, nom de liste)`. Un import déjà en vol
    est rendu tel quel avec `créé = False` : la période est `manual` et, avec
    `dedup_scope='active'`, la clé n'est relâchée qu'à la fin du job — donc deux clics ne
    lancent qu'un import, le second retrouve le premier plutôt qu'un 409 à traduire.

    **La dedup ne regarde pas le mode**, sauf la liste, qui entre dans la clé (voir
    `enqueue_api_import`). Demander un enrichissement pendant qu'une découverte tourne sur la
    même cible rend donc la découverte en cours ; demander deux listes en lance bien deux.

    **Les noms de listes sont résolus d'abord, et un fournisseur muet arrête tout.** Le listing
    du compte est gratuit et sert désormais à trois choses : refuser un identifiant que le compte
    ne connaît pas, donner à l'onglet Imports le nom que l'utilisateur reconnaîtra, et **nommer
    la liste de destination** (`_destination_list`). C'est ce troisième usage qui a fait tomber
    la tolérance d'avant : enfiler sans nom fabriquerait une liste « Liste <id> » à côté de celle
    que l'utilisateur reconnaît, et une clé refusée au moment du clic ne produirait que N imports
    voués à échouer un par un. Le refus est immédiat, avec son motif (502).
    """
    #  Les deux lectures **avant** tout le reste : elles valident l'appartenance à
    #  l'organisation (404 sinon), et il ne faut pas les découvrir après avoir enfilé la
    #  première liste. Les instances, elles, sont relues par `_enqueue_or_join` — un
    #  `rollback()` de dédup les expirerait entre deux tours de boucle.
    await get_integration(session, org_id, integration_id)
    await get_target(session, org_id, target_id)
    schedule = await session.scalar(
        select(ImportSchedule).where(
            ImportSchedule.integration_id == integration_id,
            ImportSchedule.site_target_id == target_id,
            ImportSchedule.org_id == org_id,
        )
    )
    #  Capturé **avant** tout enfilage : le `rollback()` du chemin de dedup expire l'instance,
    #  et le ré-essai plus bas doit repartir des mêmes réglages que la première tentative —
    #  sinon le second import interrogerait un autre domaine, ou perdrait le `location_id`.
    schedule_config = dict(schedule.config) if schedule is not None else None
    period = cadence.period_key("manual", datetime.now(UTC))
    wanted_lists = list(list_ids or [])
    names = await _list_names(session, org_id, integration_id, wanted_lists)
    #  `[None]` et non `[]` : sans liste, il y a **un** import à enfiler, celui d'un des deux
    #  modes dormants. La boucle est alors exactement l'ancien corps de cette fonction.
    plan: list[str | None] = list(wanted_lists) or [None]

    launched: list[tuple[Import, Job, bool, str | None, str | None]] = []
    for chosen in plan:
        name = None if chosen is None else names.get(chosen)
        result = await _enqueue_or_join(
            session,
            org_id,
            integration_id,
            target_id,
            period=period,
            config=schedule_config,
            domain=domain,
            keywords=keywords,
            list_id=chosen,
            list_name=name,
        )
        launched.append((*result, chosen, name))
    #  **Le piège des imports multiples.** Chaque tour de boucle peut passer par le chemin de
    #  dédup, qui `rollback()` — et un rollback expire *toutes* les instances de la session, y
    #  compris celles rendues par les tours précédents. Les relire au retour (le routeur lit
    #  `record.id`) déclencherait alors un chargement paresseux impossible en asynchrone : un
    #  `MissingGreenlet` au lieu d'un 202, et seulement quand deux listes sont cochées et qu'une
    #  au moins est déjà en vol. On rafraîchit donc ici, une fois, ce que la boucle a périmé.
    for record, job, *_ in launched:
        if sa_inspect(record).expired:
            await session.refresh(record)
        if sa_inspect(job).expired:
            await session.refresh(job)
    return launched


async def _list_names(
    session: AsyncSession, org_id: str, integration_id: uuid.UUID, list_ids: Sequence[str]
) -> dict[str, str]:
    """`{id: nom}` des listes demandées. Refuse un identifiant inconnu **et un tiers muet**.

    Le nom n'est plus une étiquette : c'est lui qui nomme la liste de destination
    (`_destination_list`), donc l'enfiler sans lui fabriquerait une seconde liste « Liste
    <id> » à côté de celle que l'utilisateur reconnaît. Et un fournisseur qui refuse la clé au
    moment du clic ne peut produire que N imports voués à échouer un par un : mieux vaut un
    refus tout de suite, avec le motif, qu'une grappe d'échecs à lire dans l'onglet Imports.

    Le listing du compte est mémorisé quelques minutes (`_account_lists`) : trois cibles
    lancées à la suite ne coûtent donc qu'un aller-retour, pas trois — le débit du fournisseur
    n'en autorise que trois par courte période, recensement compris.
    """
    if not list_ids:
        return {}
    known = {
        summary.id: summary.name
        for summary in await list_keyword_lists(session, org_id, integration_id)
    }
    unknown = [identifier for identifier in list_ids if identifier not in known]
    if unknown:
        raise AppError(
            404, "Not Found", f"Liste KWFinder inconnue sur ce compte : {', '.join(unknown[:5])}"
        )
    return {identifier: known[identifier] for identifier in list_ids}


async def _enqueue_or_join(
    session: AsyncSession,
    org_id: str,
    integration_id: uuid.UUID,
    target_id: uuid.UUID,
    *,
    period: str,
    config: dict[str, Any] | None,
    domain: str | None,
    keywords: Sequence[str] | None,
    list_id: str | None,
    list_name: str | None,
) -> tuple[Import, Job, bool]:
    """Enfile un import, ou rejoint celui qui est déjà en vol pour la même clé de dédup."""
    integration = await get_integration(session, org_id, integration_id)
    target = await get_target(session, org_id, target_id)
    created = await enqueue_api_import(
        session,
        org_id,
        integration,
        target,
        period=period,
        config=config,
        domain=domain,
        keywords=keywords,
        list_id=list_id,
        list_name=list_name,
    )
    if created is not None:
        return created
    #  Le `rollback()` d'`enqueue_api_import` a expiré les instances : la clé est reconstruite
    #  à partir des identifiants **passés en argument**, jamais relus sur un objet périmé.
    dedup_key = (
        f"{JOB_TYPE}:{integration_id}:{target_id}:{period}"
        if not list_id
        else f"{JOB_TYPE}:{integration_id}:{target_id}:list:{list_id}:{period}"
    )
    running = await _import_in_flight(session, org_id, dedup_key)
    if running is None:
        #  Course rarissime : le job en vol a fini entre l'échec d'insertion et cette
        #  relecture, sa clé est relâchée. Un second essai passe forcément.
        integration = await get_integration(session, org_id, integration_id)
        target = await get_target(session, org_id, target_id)
        retried = await enqueue_api_import(
            session,
            org_id,
            integration,
            target,
            period=period,
            config=config,
            domain=domain,
            keywords=keywords,
            list_id=list_id,
            list_name=list_name,
        )
        if retried is None:
            raise AppError(409, "Conflict", "Un import est déjà en cours pour cette cible")
        return retried
    return running[0], running[1], False


async def _import_in_flight(
    session: AsyncSession, org_id: str, dedup_key: str
) -> tuple[Import, Job] | None:
    job = await session.scalar(select(Job).where(Job.dedup_key == dedup_key, Job.org_id == org_id))
    if job is None:
        return None
    record = await session.scalar(
        select(Import).where(Import.job_id == job.id, Import.org_id == org_id)
    )
    return None if record is None else (record, job)


# --- le planificateur : un sweep du worker, pas un cron -------------------------------------


async def sweep_due_schedules(session: AsyncSession, *, now: datetime | None = None) -> int:
    """Enfile les synchros dues et repointe leur échéance. Rend le nombre d'imports enfilés.

    Un rendez-vous est dû dès que `next_run_at <= now` : le sweep ne consulte aucune horloge
    de cron, donc **un worker arrêté trois semaines ne rate rien** — il retrouve les
    rendez-vous en retard au démarrage et les honore, une fois chacun (`advance` saute les
    périodes manquées au lieu d'en empiler autant d'imports).

    `FOR UPDATE SKIP LOCKED` : deux workers qui balaient en même temps se répartissent les
    rendez-vous au lieu de se disputer les mêmes, exactement comme pour les jobs.
    """
    moment = now or datetime.now(UTC)
    #  Repérage **sans verrou**, puis une transaction courte par rendez-vous : un sweep qui
    #  garderait tous les verrous pendant qu'il enfile bloquerait l'écran Intégrations d'une
    #  organisation pour la lenteur d'une autre.
    candidates = (
        await session.scalars(
            select(ImportSchedule.id)
            .where(
                ImportSchedule.enabled.is_(True),
                ImportSchedule.next_run_at.is_not(None),
                ImportSchedule.next_run_at <= moment,
            )
            .order_by(ImportSchedule.next_run_at)
        )
    ).all()
    await session.rollback()

    enqueued = 0
    for schedule_id in candidates:
        #  **Un rendez-vous fautif n'en emporte aucun autre.** Sans cette garde, la moindre
        #  exception dans `_sweep_one` remontait jusqu'au worker : le planning fautif gardait
        #  son `next_run_at` échu, restait donc premier du tri à la passe suivante, et
        #  affamait indéfiniment tous les rendez-vous de **toutes** les organisations. Un seul
        #  `config` vérolé suffisait à geler le planificateur entier.
        #
        #  La garde est ici et non dans `_sweep_one` pour couvrir aussi ce que le sweep fait
        #  *autour* de l'appel, et parce que c'est l'itération qui est l'unité d'isolement.
        try:
            if await _sweep_one(session, schedule_id, moment):
                enqueued += 1
        except Exception:
            await session.rollback()
            logger.exception(
                "planning %s : rendez-vous ignoré, la passe continue sur les suivants",
                schedule_id,
            )
            #  L'aiguille n'a pas bougé (le rollback l'a emportée). On la pousse séparément,
            #  sinon ce rendez-vous se represente à chaque passe et le log tourne en boucle.
            await _skip_schedule(session, schedule_id, moment)
    return enqueued


async def _skip_schedule(session: AsyncSession, schedule_id: uuid.UUID, moment: datetime) -> None:
    """Avance l'échéance d'un rendez-vous qu'on n'a pas su honorer. **Ne lève jamais.**

    Passer son tour n'est pas l'oublier : l'utilisateur voit toujours son mode de synchro et
    sa prochaine échéance, l'erreur est dans les logs, et le prochain rendez-vous retentera —
    avec le réglage corrigé, si entre-temps il l'a été.
    """
    try:
        record = await session.scalar(
            select(ImportSchedule).where(ImportSchedule.id == schedule_id).with_for_update()
        )
        if record is not None and record.next_run_at is not None:
            record.next_run_at = cadence.advance(record.sync_mode, record.next_run_at, moment)
        await session.commit()
    except Exception:
        await session.rollback()
        logger.exception("planning %s : échéance non repoussée", schedule_id)


async def _sweep_one(session: AsyncSession, schedule_id: uuid.UUID, moment: datetime) -> bool:
    """Un rendez-vous, une transaction. Rend `True` si un import a été enfilé.

    Le rendez-vous est **re-verrouillé et re-jugé** ici : entre le repérage et maintenant, un
    autre worker a pu l'honorer, ou l'utilisateur a pu repasser la cible en manuel.
    """
    record = await session.scalar(
        select(ImportSchedule)
        .where(
            ImportSchedule.id == schedule_id,
            ImportSchedule.enabled.is_(True),
            ImportSchedule.next_run_at.is_not(None),
            ImportSchedule.next_run_at <= moment,
        )
        .with_for_update(skip_locked=True)
        .execution_options(populate_existing=True)
    )
    if record is None:
        await session.rollback()
        return False
    due_at = record.next_run_at or moment
    period = cadence.period_key(record.sync_mode, due_at)
    next_at = cadence.advance(record.sync_mode, due_at, moment)
    integration = await session.get(Integration, record.integration_id)
    target = await session.get(SiteTarget, record.site_target_id)
    if integration is None or target is None:  # pragma: no cover - les FK l'interdisent
        await session.rollback()
        return False
    #  Défense en profondeur : le planning, son intégration et sa cible appartiennent à la
    #  même organisation, ou le rendez-vous est corrompu et on n'y touche pas.
    if integration.org_id != record.org_id or target.org_id != record.org_id:  # pragma: no cover
        logger.error("planning %s : organisations incohérentes, rendez-vous ignoré", record.id)
        await session.rollback()
        return False

    #  L'aiguille avance **dans la même transaction** que l'enfilage (le commit est celui
    #  d'`enqueue_api_import`) : jamais un import enfilé sans que son rendez-vous ait bougé,
    #  jamais un rendez-vous avancé sans import.
    record.next_run_at = next_at
    created = await enqueue_api_import(
        session,
        record.org_id,
        integration,
        target,
        period=period,
        config=dict(record.config),
    )
    if created is not None:
        return True
    #  Dedup : un import de cette période est déjà en vol. Le rollback d'`enqueue_api_import`
    #  a emporté l'avancement de l'aiguille — on le rejoue seul, sans quoi le rendez-vous
    #  resterait dû à chaque passe et le sweep tournerait en rond dessus.
    again = await session.scalar(
        select(ImportSchedule).where(ImportSchedule.id == schedule_id).with_for_update()
    )
    if again is not None:
        again.next_run_at = next_at
    await session.commit()
    return False


registry.register_periodic(registry.PeriodicTask("import_schedules", sweep_due_schedules))