"""Lectures/écritures du module Imports — toutes filtrées par `org_id` (spec Imports §6).

Conventions du module Sites reprises à la lettre : jamais de `SELECT *` (l'ORM émet la liste
explicite des colonnes mappées), 404 et jamais 403 pour ce qui appartient à une autre
organisation, pagination par curseur UUIDv7, Problem Details pour toute erreur d'entrée.

Deux choses méritent d'être lues avant de modifier ce fichier :

- **Le travail lourd part en thread.** Lire un en-tête CSV, c'est ouvrir DuckDB et valider
  l'encodage du fichier entier (`csvfile.detect_encoding`) : synchrone et long. Dans une
  requête HTTP asynchrone, l'y laisser bloquerait la boucle pour tous les autres appels. Tout
  ce qui touche au disque passe donc par `asyncio.to_thread` (§4.1, même règle que le worker).
- **La création d'un import est atomique.** Liste de destination, job et ligne `imports` sont
  écrits dans **une** transaction et publiés par un seul commit (`jobs.service.enqueue_pending`
  insère sans committer). C'est ce qui interdit l'*import fantôme* : un arrêt entre deux commits
  laisserait sinon un job qui s'exécute et une ligne `imports` qui n'existe pas — réservoir
  peuplé, onglet Imports vide, aucun rapport. La promesse du socle (« visible d'un autre
  process ») tient : c'est le commit qui publie, pas l'insertion.
"""

import asyncio
import contextlib
import logging
import uuid
from collections.abc import Callable, Iterator, Mapping, Sequence
from pathlib import Path
from typing import Any

import duckdb
from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.errors import AppError
from app.db.cursor import decode_cursor, encode_cursor
from app.imports.schemas import (
    PREVIEW_ROWS,
    CountryCount,
    DetectionOut,
    ImportCreate,
    ImportOut,
    ListRef,
    PreviewIn,
    PreviewOut,
    PreviewRow,
    PreviewWarning,
    ReconcileEventOut,
)
from app.jobs import service as jobs_service
from app.jobs.models import Job
from app.keywords import countries, pipeline, sources
from app.keywords.models import Import, KeywordList
from app.keywords.sources.base import (
    AmbiguousHeaderError,
    CanonicalKeywordRow,
    ParsedRow,
    header_index,
    normalize_header,
)
from app.keywords.sources.csvfile import count_by_column, read_header
from app.sites.models import SiteTarget

logger = logging.getLogger(__name__)

# Type du job de rattachement rétroactif (§3.9). Il est **écrit par T6** : d'ici là, aucun job
# de ce type n'existe et la branche « événements » du listing ne rend rien — no-op propre.
RECONCILE_JOB_TYPE = "map_reconcile"

# Lignes lues à la création d'un import pour prouver que l'adaptateur et le mapping tiennent
# debout **avant** le 202 (`_first_rows`). Quelques-unes suffisent : le générateur lève à sa
# première itération, et une ligne en erreur en tête ne doit pas faire refuser le fichier.
_VALIDATION_SCAN_ROWS = 5


# --- cible et locale ---------------------------------------------------------------------


async def get_target(session: AsyncSession, org_id: str, target_id: uuid.UUID) -> SiteTarget:
    target = await session.scalar(
        select(SiteTarget).where(SiteTarget.id == target_id, SiteTarget.org_id == org_id)
    )
    if target is None:
        # 404 et non 403 : une cible d'une autre org n'existe pas (architecture §9).
        raise AppError(404, "Not Found", "Cible inconnue")
    return target


def locale_of(target: SiteTarget) -> str:
    """Le marché complet langue-pays hérité de la cible (décision n°2) : `fr-FR`, `fr-BE`."""
    return f"{target.language.lower()}-{target.country.upper()}"


def _adapter_or_422(name: str) -> Any:
    adapter = sources.get(name)
    if adapter is None:
        raise AppError(
            422,
            "Validation error",
            f"Format inconnu : {name} (connus : {', '.join(sources.names())})",
        )
    return adapter


def _existing_file(org_id: str, sha256: str) -> Path:
    path = pipeline.import_file_path(org_id, sha256)
    if not path.is_file():
        # Le fichier a expiré (rétention 30 jours, décision n°7) ou n'a jamais été téléversé.
        raise AppError(404, "Not Found", "Fichier d'import inconnu ou expiré")
    return path


# --- aperçu (POST /imports/preview) -------------------------------------------------------


def _consume(rows: Iterator[ParsedRow], limit: int) -> list[ParsedRow]:
    """Les `limit` premières lignes, puis **fermeture du générateur**.

    L'adaptateur tient une connexion DuckDB et, le cas échéant, un fichier transcrit en
    temporaire : abandonner le générateur les laisserait ouverts jusqu'au ramasse-miettes.
    `contextlib.closing` lui envoie un `GeneratorExit`, ce qui déroule son `with CsvScan(...)`.
    """
    collected: list[ParsedRow] = []
    with contextlib.closing(rows):
        for row in rows:
            collected.append(row)
            if len(collected) >= limit:
                break
    return collected


async def _read_or_422(work: Callable[..., Any], *args: Any) -> Any:
    """Exécute une lecture de fichier dans un thread, et traduit ses refus en 422.

    Trois familles d'échec, toutes imputables au **fichier ou au mapping fournis**, jamais à un
    bug : en-têtes qui se confondent une fois normalisés (`AmbiguousHeaderError` — importer
    l'une pour l'autre en silence serait pire qu'un refus, cf. `sources/base.py`), mapping
    manuel inutilisable (`MappingError`), et fichier que le lecteur CSV n'ouvre pas du tout
    (`duckdb.Error`). Un 500 sur le CSV mal formé d'un utilisateur serait un aveu d'impuissance
    déguisé en panne ; le reste des exceptions remonte intact.
    """
    try:
        return await asyncio.to_thread(work, *args)
    except AmbiguousHeaderError as exc:
        raise AppError(422, "Validation error", str(exc)) from exc
    except sources.MappingError as exc:
        raise AppError(422, "Validation error", str(exc)) from exc
    except duckdb.Error as exc:
        logger.warning("fichier d'import illisible : %s", exc)
        raise AppError(422, "Validation error", f"Fichier CSV illisible : {exc}") from exc


def _scan(
    path: Path,
    requested: str | None,
    columns: Mapping[str, str] | None,
    country_column: str | None = None,
) -> dict[str, Any]:
    """Partie **synchrone** de l'aperçu : en-tête, détection, N lignes canoniques, pays.

    Exécutée dans un thread par `preview`. Ne touche ni à la base ni au réseau.

    Un effet de bord de la proposition automatique mérite d'être connu : un fichier à en-têtes
    homonymes (« Mot-clé » **et** « Mot clé ») rendait auparavant l'aperçu des en-têtes seuls,
    puisque `generic_csv` sans mapping ne lisait aucune ligne ; il fait maintenant lever
    `AmbiguousHeaderError` dès le premier aperçu, donc un 422. C'est **assumé** : le message
    est actionnable (« renommez-en une avant l'import »), et l'import aurait de toute façon
    échoué au même endroit — le dire tout de suite vaut mieux que le dire après le mapping.
    """
    headers = read_header(path)
    candidates = sources.detect(headers)
    best = sources.best(headers)
    ambiguous = best is None

    if requested is not None:
        chosen = requested
        detection = next((c for c in candidates if c.source == requested), None)
    elif best is not None:
        chosen = best.source
        detection = best
    else:
        # Ambiguïté ou aucune correspondance : `generic_csv` et son mapping manuel (§5).
        chosen = sources.GENERIC.name
        detection = None

    adapter = sources.get(chosen)
    if adapter is None:
        raise AppError(
            422,
            "Validation error",
            f"Format inconnu : {chosen} (connus : {', '.join(sources.names())})",
        )

    generic = adapter.name == sources.GENERIC.name
    #  « Le mapping, nous on le fait » (décision 2026-09-01) : sur le chemin `generic_csv`,
    #  l'heuristique de synonymes propose un mapping **avant** que l'utilisateur ne touche à
    #  quoi que ce soit. Elle n'est calculée que là — un adaptateur reconnu a déjà ses colonnes.
    suggestion = sources.suggest_columns(headers) if generic else None
    #  Le mapping explicite de l'utilisateur fait toujours autorité : la proposition ne sert
    #  qu'à peupler l'aperçu du **premier** passage, quand il n'a encore rien choisi. Le test
    #  est `is not None` et non la simple vérité du dict : un `columns: {}` explicite est un
    #  mapping vide **choisi**, pas une absence de choix. Sans cette distinction, il appliquait
    #  la proposition tout en taisant l'avertissement qui l'annonce (les avertissements, eux,
    #  se taisent sur `payload.columns is not None`) — un mapping appliqué sans être dit.
    applied = (
        columns
        if columns is not None
        else (suggestion.columns if suggestion is not None and suggestion.usable else None)
    )

    options: dict[str, Any] = {}
    if applied:
        options["columns"] = dict(applied)
    if country_column:
        options["country_column"] = country_column

    rows: list[ParsedRow] = []
    #  `generic_csv` sans mapping **ni** proposition n'a rien à parser : l'aperçu rend alors les
    #  en-têtes seuls, ce dont l'étape 2 du wizard a besoin pour construire le mapping à la main.
    if not generic or applied:
        rows = _consume(adapter.parse_csv(path, options=options or None), PREVIEW_ROWS)

    return {
        "headers": headers,
        "format": chosen,
        "language": detection.language if detection else None,
        "confidence": detection.confidence if detection else 0.0,
        "columns": dict(detection.columns) if detection else dict(applied or {}),
        "suggestion": suggestion,
        "candidates": candidates,
        "rows": rows,
        "ambiguous": ambiguous,
        "countries": _census(path, headers, adapter, options, explicit=country_column),
    }


def _census(
    path: Path,
    headers: Sequence[str],
    adapter: Any,
    options: Mapping[str, Any],
    *,
    explicit: str | None,
) -> list[tuple[str, int]]:
    """Les libellés de pays du fichier **entier** et leur décompte, ou `[]` (§4.2).

    Court-circuité — aucune ouverture de fichier — quand l'adaptateur ne déclare pas de
    colonne pays : c'est le cas de tout Semrush et de `generic_csv` sans `country_column`, et
    ils sont la majorité des aperçus. Le `GROUP BY` ne se paie que par les fichiers qui ont
    quelque chose à recenser.

    L'en-tête déclaré est confronté à ceux du fichier avec la même tolérance que le mapping
    (`normalize_header` : casse, tirets, espaces). Deux issues distinctes quand il n'y est
    pas : un **choix** de l'utilisateur (`explicit`) est un 422 immédiat, comme une colonne de
    mapping absente ; une colonne **déclarée par l'adaptateur** ne fait que priver l'aperçu de
    son recensement — un export KWFinder dont l'utilisateur a masqué `Location` reste un
    export KWFinder parfaitement importable.

    Les valeurs vides sont écartées : une cellule vide n'est pas un libellé, il n'y a rien à
    nommer à l'écran ni à proposer comme cible. Les lignes correspondantes seront comptées par
    le rapport d'import, qui compte des lignes et non des libellés (§4.5).
    """
    declared = adapter.country_header(options)
    if declared is None:
        return []
    real = header_index(headers).get(normalize_header(declared))
    if real is None:
        if explicit is not None:
            raise AppError(422, "Validation error", f"Colonne absente du fichier : {explicit!r}")
        return []
    counted = count_by_column(path, real)
    return [(raw.strip(), rows) for raw, rows in counted if raw is not None and raw.strip()]


# Libellés des champs canoniques dans les avertissements — les mêmes mots que les `Label` de
# l'étape 2 du wizard (`web/lib/keywords.ts`), pour que l'utilisateur relie l'un à l'autre.
_FIELD_LABELS = {
    "keyword": "mot-clé",
    "volume": "volume",
    "kd": "difficulté",
    "cpc": "CPC",
    "intent": "intention",
    "serp_features": "fonctionnalités SERP",
    "trend": "tendance",
    "url": "URL",
    "position": "position",
}
# Un export réel porte des centaines de colonnes (147 chez KWFinder) : les énumérer toutes
# ferait un pavé illisible là où l'utilisateur attend une information.
_LISTING_MAX = 8


def _listing(items: Sequence[str]) -> str:
    """Une énumération lisible, tronquée à `_LISTING_MAX` et comptée au-delà."""
    if len(items) <= _LISTING_MAX:
        return ", ".join(items)
    return f"{', '.join(items[:_LISTING_MAX])} … et {len(items) - _LISTING_MAX} autres"


def _mapping_warnings(
    suggestion: sources.Suggestion | None, *, explicit: bool
) -> list[PreviewWarning]:
    """Ce que la proposition automatique a trouvé, et ce dont elle n'est pas sûre (§5).

    `explicit` : l'utilisateur a fourni son propre mapping — il a donc déjà tranché, et la
    proposition n'est plus appliquée. Continuer à commenter ce qu'*on* aurait choisi
    contredirait l'écran qu'il a sous les yeux : on se tait.
    """
    if suggestion is None or explicit:
        return []
    warnings: list[PreviewWarning] = []
    if not suggestion.usable:
        return [
            PreviewWarning(
                code="no_mapping_suggested",
                detail=(
                    "Aucune colonne de ce fichier ne ressemble à un mot-clé : le mapping est à "
                    "faire à la main, en commençant par le champ « mot-clé »."
                ),
            )
        ]
    recognised = _listing([_FIELD_LABELS.get(name, name) for name in suggestion.columns])
    warnings.append(
        PreviewWarning(
            code="mapping_suggested",
            detail=(
                f"Colonnes reconnues automatiquement ({recognised}) : l'aperçu ci-dessous les "
                "applique déjà — vérifiez, et ajustez si besoin."
            ),
        )
    )
    for name, others in suggestion.alternatives.items():
        label = _FIELD_LABELS.get(name, name)
        warnings.append(
            PreviewWarning(
                code="ambiguous_column",
                detail=(
                    f"Plusieurs colonnes pouvaient servir de {label} : "
                    f"« {suggestion.columns[name]} » a été retenue, "
                    f"{_listing(others)} ne l'{'ont' if len(others) > 1 else 'a'} pas été."
                ),
            )
        )
    return warnings


async def preview(session: AsyncSession, org_id: str, payload: PreviewIn) -> PreviewOut:
    """Détection + aperçu borné. **Rien n'est persisté** (§6)."""
    path = _existing_file(org_id, payload.sha256)
    if payload.format is not None:
        _adapter_or_422(payload.format)
    target = (
        None
        if payload.site_target_id is None
        else await get_target(session, org_id, payload.site_target_id)
    )
    scan = await _read_or_422(_scan, path, payload.format, payload.columns, payload.country_column)

    suggestion: sources.Suggestion | None = scan["suggestion"]
    warnings: list[PreviewWarning] = []
    if scan["ambiguous"] and payload.format is None:
        warnings.append(
            PreviewWarning(
                code="ambiguous_format",
                detail=(
                    "Aucune signature de colonnes ne reconnaît ce fichier sans ambiguïté : "
                    "« CSV générique » est proposé."
                ),
            )
        )
    warnings.extend(_mapping_warnings(suggestion, explicit=payload.columns is not None))
    used = set(scan["columns"].values())
    if ignored := [header for header in scan["headers"] if header not in used]:
        warnings.append(
            PreviewWarning(
                code="ignored_columns",
                detail=(
                    "Colonnes non utilisées par ce format (conservées dans les données brutes) : "
                    + _listing(ignored)
                ),
            )
        )
    if not any(isinstance(row, CanonicalKeywordRow) for row in scan["rows"]):
        warnings.append(
            PreviewWarning(
                code="no_rows",
                detail="Aucune ligne exploitable dans l'aperçu : vérifiez le format ou le mapping.",
            )
        )

    return PreviewOut(
        format=scan["format"],
        confidence=scan["confidence"],
        language=scan["language"],
        headers=scan["headers"],
        columns=scan["columns"],
        suggested_mapping=(
            dict(suggestion.columns) if suggestion is not None and suggestion.usable else None
        ),
        countries=await _country_counts(session, org_id, target, scan["countries"]),
        candidates=[
            DetectionOut(source=c.source, language=c.language, confidence=c.confidence)
            for c in scan["candidates"]
        ],
        sample=[_preview_row(row) for row in scan["rows"] if isinstance(row, CanonicalKeywordRow)],
        warnings=warnings,
    )


async def _country_counts(
    session: AsyncSession,
    org_id: str,
    target: SiteTarget | None,
    counted: Sequence[tuple[str, int]],
) -> list[CountryCount]:
    """Le recensement brut, résolu en pays et rapproché des cibles déjà créées (§4.2).

    Une **seule** requête pour tous les pays du fichier plutôt qu'une par libellé : les cibles
    d'un site sont plafonnées à vingt, les charger en bloc coûte moins qu'onze allers-retours.
    Filtrée par `org_id` comme toute lecture DATA, et par `site_id` : ce sont les cibles du
    site de l'import, pas celles de toute l'organisation.

    Sans `site_target_id` au preview, l'api ne connaît aucun site : `target_ids` reste vide et
    le wizard proposera de créer chaque cible. Dégradé, jamais faux.
    """
    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 counted:
        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


def _preview_row(row: CanonicalKeywordRow) -> PreviewRow:
    return PreviewRow(
        line=row.line,
        keyword=row.keyword,
        volume=row.volume,
        kd=None if row.kd is None else float(row.kd),
        cpc=None if row.cpc is None else float(row.cpc),
        intent=row.intent,
        serp_features=row.serp_features,
        trend=row.trend,
        url=row.url,
        position=None if row.position is None else float(row.position),
    )


# --- création (POST /imports) -------------------------------------------------------------


async def duplicate_import_id(
    session: AsyncSession, org_id: str, target_id: uuid.UUID, sha256: str
) -> uuid.UUID | None:
    """Le dernier import de **même sha sur la même cible** (§1.3), s'il existe.

    Avertissement doux, jamais un refus : le re-import après correction d'adaptateur est
    légitime, et l'idempotence à la ligne (§4.4) rend le rejeu inoffensif.
    """
    return await session.scalar(
        select(Import.id)
        .where(
            Import.org_id == org_id,
            Import.site_target_id == target_id,
            Import.file_sha256 == sha256,
        )
        .order_by(Import.id.desc())
        .limit(1)
    )


def _first_rows(path: Path, adapter: Any, columns: Mapping[str, str] | None) -> list[ParsedRow]:
    """Les premières lignes du fichier, telles que l'adaptateur les rendra à l'import.

    **Valider, et rien d'autre** : le générateur lève `MappingError` ou `AmbiguousHeaderError`
    à sa première itération — jamais à sa création, d'où la consommation. Un mapping faux doit
    être un 422 tout de suite, pas un job entier de lignes en erreur découvert à la fin.

    Ce parcours lisait aussi le « pays annoncé » du fichier, du temps où `detected_country`
    voulait dire cela. Il ne le veut plus : depuis le cloisonnement (§4.5), la colonne porte le
    pays **de la cible**, celui qui a été gardé — un fichier peut en annoncer onze.
    """
    options = {"columns": dict(columns)} if columns else None
    return _consume(adapter.parse_csv(path, options=options), _VALIDATION_SCAN_ROWS)


async def _resolve_list(
    session: AsyncSession, org_id: str, target: SiteTarget, ref: ListRef, source: str
) -> tuple[uuid.UUID, KeywordList | None]:
    """La liste de destination : existante (validée) ou nouvelle (créée, non encore committée)."""
    if ref.id is not None:
        found = await session.scalar(
            select(KeywordList.id).where(
                KeywordList.id == ref.id,
                KeywordList.site_target_id == target.id,
                KeywordList.org_id == org_id,
            )
        )
        if found is None:
            raise AppError(404, "Not Found", "Liste de mots-clés inconnue")
        return found, None
    assert ref.name is not None  # garanti par `ListRef._exactly_one`
    taken = await session.scalar(
        select(KeywordList.id).where(
            KeywordList.site_target_id == target.id, KeywordList.name == ref.name
        )
    )
    if taken is not None:
        raise AppError(409, "Conflict", f"Une liste « {ref.name} » existe déjà pour cette cible")
    created = KeywordList(org_id=org_id, site_target_id=target.id, name=ref.name, source=source)
    session.add(created)
    try:
        await session.flush()
    except IntegrityError as exc:
        # Filet anti-course : l'UNIQUE (site_target_id, name) est l'arbitre, pas le pré-check.
        await session.rollback()
        raise AppError(
            409, "Conflict", f"Une liste « {ref.name} » existe déjà pour cette cible"
        ) from exc
    return created.id, created


async def create_import(
    session: AsyncSession, org_id: str, payload: ImportCreate
) -> tuple[Import, uuid.UUID]:
    """Valide, crée la ligne `imports` et enfile le job `import_csv` (§6). Rend `(import, job_id)`.

    **Un seul commit publie tout** : la liste de destination, le job et la ligne `imports`
    entrent dans la même transaction (`jobs_service.enqueue_pending` insère sans committer).
    C'est ce qui interdit l'*import fantôme* — un arrêt du process entre deux commits laisserait
    sinon un job qui s'exécute, un réservoir qui se peuple, et **rien** dans l'onglet Imports :
    aucun rapport, aucune trace, rien à quoi l'utilisateur puisse se raccrocher. La promesse du
    socle (« le job doit être visible d'un autre process ») reste tenue : c'est le commit qui
    rend visible, pas l'insertion.
    """
    target = await get_target(session, org_id, payload.site_target_id)
    adapter = _adapter_or_422(payload.format)
    path = _existing_file(org_id, payload.sha256)

    if adapter.name != sources.GENERIC.name and payload.columns is not None:
        raise AppError(
            422,
            "Validation error",
            "Le mapping manuel de colonnes n'a de sens que pour le format « generic_csv »",
        )
    #  Validation du mapping avant le 202 : un mapping `generic_csv` faux doit être un 422
    #  maintenant, pas un job entier de lignes en erreur découvert à la fin (`sources/generic`).
    await _read_or_422(_first_rows, path, adapter, payload.columns)

    if not payload.force:
        duplicate = await duplicate_import_id(session, org_id, target.id, payload.sha256)
        if duplicate is not None:
            raise AppError(
                409,
                "Conflict",
                f"Ce fichier a déjà été importé sur cette cible (import {duplicate}) — "
                "relancer avec « force » pour le rejouer",
            )

    list_id, created_list = await _resolve_list(session, org_id, target, payload.list, adapter.name)

    #  L'identifiant est demandé à Postgres plutôt que tiré en Python : `uuidv7()` est
    #  temporellement croissant, ce dont dépend le curseur du listing (§6). Un `uuid4` y
    #  ferait des trous d'ordre invisibles jusqu'à la première page manquante.
    import_id = await session.scalar(select(func.uuidv7()))
    assert import_id is not None
    import_id = uuid.UUID(str(import_id))

    #  Les `options` du job, reprises telles quelles par l'adaptateur au run — le mapping manuel
    #  et, **hors de lui**, la colonne pays choisie par l'utilisateur (Q4 : un canal parallèle,
    #  jamais un champ canonique — le pays est une propriété de la ligne, pas une métrique du
    #  mot-clé). L'un et l'autre facultatifs : sans eux, pas de clé `options`, comme avant.
    options: dict[str, Any] = {}
    if payload.columns:
        options["columns"] = dict(payload.columns)
    if payload.country_column:
        options["country_column"] = payload.country_column

    job = await jobs_service.enqueue_pending(  # insère sans committer : voir la docstring
        session,
        org_id,
        pipeline.JOB_TYPE,
        payload={
            "import_id": str(import_id),
            "file_sha256": payload.sha256,
            "source": adapter.name,
            "site_target_id": str(target.id),
            "locale": locale_of(target),
            #  Le pays de la cible, **figé ici** : le pipeline cloisonne le fichier dessus
            #  (§4.3). Le lire au run ferait dépendre le filtrage de l'état de la cible à cet
            #  instant, et le rapport d'un import repris cesserait d'être celui d'un import
            #  d'une traite.
            "country": target.country.upper(),
            "fetched_at": payload.fetched_at.isoformat(),
            "keyword_list_id": str(list_id),
            **({"options": options} if options else {}),
        },
    )

    record = Import(
        id=import_id,
        org_id=org_id,
        site_target_id=target.id,
        source=adapter.name,
        kind="csv",
        filename=payload.filename,
        file_sha256=payload.sha256,
        detected_format=adapter.name,
        #  Le pays **retenu** (§4.5) : celui de la cible, donc celui des lignes que le pipeline
        #  gardera. Ce n'est plus « ce que le fichier annonçait » — un export multi-marchés en
        #  annonce onze, et c'est le recensement de l'aperçu qui les dit. **Un seul écrivain
        #  pour cette colonne**, ici : le worker n'y touche pas.
        detected_country=target.country.upper(),
        fetched_at=payload.fetched_at,
        keyword_list_id=list_id,
        job_id=job.id,
        status="queued",
    )
    session.add(record)
    #  Flush explicite : la ligne `imports` doit exister avant que `keyword_lists` ne la
    #  référence, et l'unité de travail n'ordonne pas insertions et mises à jour entre tables.
    await session.flush()
    if created_list is not None:
        created_list.created_from_import_id = import_id
    #  **Le** commit : liste, job et ligne `imports` deviennent visibles ensemble.
    await session.commit()
    return record, job.id


# --- listings et détail --------------------------------------------------------------------


async def list_entries(
    session: AsyncSession,
    org_id: str,
    *,
    site_target_id: uuid.UUID | None,
    cursor: str | None,
    limit: int,
) -> tuple[list[ImportOut | ReconcileEventOut], str | None]:
    """Imports **et** événements `map_reconcile`, entrelacés par `id` décroissant (§6, §12.2).

    L'UNION est faite en Python plutôt qu'en SQL : les deux branches portent des colonnes
    différentes, la page est bornée à 100 lignes, et chaque branche applique déjà le même
    keyset (`id < cursor`, `ORDER BY id DESC`, `LIMIT limit + 1`). Fusionner `limit + 1` lignes
    de chacune suffit donc à produire une page exacte — aucune branche ne peut être affamée par
    l'autre — pour le prix de deux requêtes indexées au lieu d'un UNION à colonnes fictives.
    """
    after = None if cursor is None else decode_cursor(cursor)

    imports_query = select(Import).where(Import.org_id == org_id)
    if site_target_id is not None:
        imports_query = imports_query.where(Import.site_target_id == site_target_id)
    if after is not None:
        imports_query = imports_query.where(Import.id < after)
    records = list(
        (await session.scalars(imports_query.order_by(Import.id.desc()).limit(limit + 1))).all()
    )

    events_query = select(Job).where(Job.org_id == org_id, Job.type == RECONCILE_JOB_TYPE)
    if site_target_id is not None:
        events_query = events_query.where(
            Job.payload["site_target_id"].astext == str(site_target_id)
        )
    if after is not None:
        events_query = events_query.where(Job.id < after)
    events = list(
        (await session.scalars(events_query.order_by(Job.id.desc()).limit(limit + 1))).all()
    )

    merged: list[ImportOut | ReconcileEventOut] = [
        *(ImportOut.model_validate(record) for record in records),
        *(_event(job) for job in events),
    ]
    merged.sort(key=lambda entry: entry.id, reverse=True)
    next_cursor = encode_cursor(merged[limit - 1].id) if len(merged) > limit else None
    return merged[:limit], next_cursor


def _event(job: Job) -> ReconcileEventOut:
    payload = job.payload if isinstance(job.payload, Mapping) else {}
    raw_target = payload.get("site_target_id")
    return ReconcileEventOut(
        id=job.id,
        site_target_id=None if raw_target is None else uuid.UUID(str(raw_target)),
        status=job.status,
        report=job.report,
        error=job.error,
        created_at=job.created_at,
    )


async def get_import(session: AsyncSession, org_id: str, import_id: uuid.UUID) -> Import:
    record = await session.scalar(
        select(Import).where(Import.id == import_id, Import.org_id == org_id)
    )
    if record is None:
        raise AppError(404, "Not Found", "Import inconnu")
    return record


async def errors_csv(session: AsyncSession, org_id: str, import_id: uuid.UUID) -> Path:
    """Le `<sha256>.errors.csv` d'un import (§4.5) — **404 ou 410**, la spec distingue les deux.

    Le rapport tranche : `report["errors_csv"]` porte le nom du fichier que le job a écrit.
    - absent → l'import n'a jamais produit de lignes en erreur : la ressource n'a **jamais**
      existé, c'est un 404 ;
    - présent mais fichier introuvable sur le disque → il a été **purgé** (rétention 30 jours,
      décision n°7) : la ressource a existé et ne reviendra pas, c'est un 410. L'écran d'import
      peut alors dire « rapport expiré » plutôt que « aucune erreur », ce qui n'est pas la même
      information pour l'utilisateur.
    """
    record = await get_import(session, org_id, import_id)
    report = record.report if isinstance(record.report, Mapping) else {}
    if record.file_sha256 is None or not report.get("errors_csv"):
        raise AppError(404, "Not Found", "Aucun fichier d'erreurs pour cet import")
    path = pipeline.import_file_path(org_id, record.file_sha256, suffix=".errors.csv")
    if not path.is_file():
        raise AppError(
            410, "Gone", "Fichier d'erreurs expiré : les fichiers d'import sont purgés à 30 jours"
        )
    return path