"""Pipeline d'import CSV — le handler du job `import_csv` (spec Imports §4).

Lit un export, le traduit par un adaptateur (§5), et l'écrit **par lots bornés, une
transaction par lot**. Trois propriétés portent tout le reste :

1. **Idempotence métier** (§4.4) — les upserts portent sur les unicités canoniques. Rejouer
   le même fichier laisse la base strictement identique : chaque `DO UPDATE` est gardé par
   un `IS DISTINCT FROM`, donc une observation inchangée n'écrit rien, ne consomme pas de
   version de ligne, et la ligne est comptée « ignorée ». C'est aussi ce qui rend le
   **re-import correctif** possible (décision C1) : même clé `(keyword_id, source,
   fetched_at)` = **même observation**, remplaçable après correction d'adaptateur — jamais
   une fusion inter-sources, la clé porte la provenance.
2. **Reprise à mi-course sans doublon** (§4.4, §11) — le lot et son curseur sont committés
   **ensemble** : `save_progress` termine la transaction qui porte les écritures du lot. Un
   worker tué entre deux lots perd le lot en vol *et* son curseur du même coup ; la reprise
   repart du dernier lot committé, sans jamais réécrire un lot déjà passé.
3. **Aucune ligne perdue en silence** (§4.2) — une ligne illisible devient une ligne en
   erreur numérotée, jamais une exception. Y compris les lignes que le lecteur CSV lui-même
   écarte : l'export Keyword Magic réel finit par trois lignes de pied de page commerciales
   qui n'ont pas le bon nombre de colonnes.

**Le rapport est exact après une reprise**, et pas seulement l'état de la base. Les
compteurs sont séparés en deux familles, ce qui suffit à le garantir sans rien empiler :

- ceux de **lecture** (`rows_read`, `rows_error`, les doublons intra-fichier, le
  cloisonnement par pays) et la liste des erreurs sont **déterministes** — même fichier, même
  adaptateur, même pays, même résultat — donc recalculés intégralement à chaque run, y
  compris sur les lots sautés ;
- ceux d'**écriture** (`rows_inserted`, `keywords_created`, …) décrivent un travail non
  rejoué : eux seuls voyagent dans `progress` avec le curseur.

**Cloisonnement par pays** (spec « Multi-sources et cloisonnement pays » §4.3) — une ligne
dont le pays n'est pas celui de la cible n'entre pas dans l'import. Le pays de la cible est
figé **à l'enfilage**, dans le payload du job : c'est ce qui garde le déterminisme ci-dessus.
Le lire dans `site_targets` au moment du run ferait qu'un job repris après modification de la
cible filtrerait autrement qu'au premier tour, et le rapport d'un import interrompu cesserait
d'être celui d'un import d'une traite.

**Invariant `keyword_metrics.org_id = keywords.org_id`** (dette M3 de T1, que la fonction
SQL `keyword_metrics_latest` suppose en agrégeant `min(m.org_id)`) : il est structurel ici
et nulle part ailleurs, puisque ce module est le seul à écrire `keyword_metrics`. Les deux
`org_id` viennent de la même source — `job.org_id` — et l'identifiant du mot-clé est relu
**filtré par ce même `org_id`** (`_keyword_ids`), donc un mot-clé d'une autre organisation
ne peut jamais être rattaché. `test_import_pipeline.py` le vérifie par une jointure.

`page_id` reste `NULL` à l'insertion : la résolution est encapsulée dans
`app.keywords.resolution` (amendement I4) et branchée par le module Crawl (T6).
"""

import asyncio
import csv
import logging
import re
import uuid
from collections.abc import Iterable, Iterator, Mapping, Sequence
from dataclasses import asdict, dataclass, field, fields
from datetime import datetime
from pathlib import Path
from typing import Any

from sqlalchemy import or_, select, update
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.config import get_settings
from app.core.urls import UrlError, normalize_url
from app.jobs import registry, service
from app.jobs.models import Job
from app.keywords import countries, resolution, sources
from app.keywords.models import Import, Keyword, KeywordListItem, KeywordMetric, KeywordPageMap
from app.keywords.sources.base import (
    CanonicalKeywordRow,
    RowError,
    header_index,
    normalize_header,
    to_physical_lines,
)
from app.keywords.sources.csvfile import read_header

logger = logging.getLogger(__name__)

JOB_TYPE = "import_csv"
# Lot de lignes (§4.3). Une transaction par lot : assez gros pour amortir les allers-retours,
# assez petit pour qu'un lot perdu à la reprise ne coûte rien et que les verrous ne tiennent
# pas longtemps sur les tables les plus écrites du modèle.
BATCH_SIZE = 5_000
# asyncpg plafonne une instruction à 32 767 paramètres liés : un lot de 5 000 lignes × 11
# colonnes le dépasse. Les VALUES sont donc découpées, sans changer la transaction — le lot
# reste atomique. Marge volontaire sous le plafond.
MAX_BOUND_PARAMS = 30_000
# Plafond de collecte du rapport (§4.5). `errors.csv` en porte davantage, mais pas non plus
# sans borne : au-delà, le fichier a un problème de format, pas de données.
ERROR_REPORT_LIMIT = 1_000
ERRORS_CSV_LIMIT = 50_000

_ORG_ID = re.compile(r"^[A-Za-z0-9_-]{1,128}$")
_SHA256 = re.compile(r"^[0-9a-f]{64}$")


def import_file_path(org_id: str, sha256: str, *, suffix: str = ".csv") -> Path:
    """`<IMPORT_FILES_DIR>/<org_id>/<sha256>.csv` (§3.12), composants **sanitisés**.

    `org_id` vient du JWT (identifiant opaque Better Auth) et `sha256` est calculé par l'api :
    ni l'un ni l'autre ne devrait contenir de séparateur de chemin. Ils sont vérifiés quand
    même — défense en profondeur exigée par la spec : c'est la seule construction de chemin
    du module, et une traversée y coûterait la lecture de n'importe quel fichier du serveur.
    """
    if not _ORG_ID.fullmatch(org_id):
        raise ValueError(f"org_id inutilisable dans un chemin : {org_id!r}")
    if not _SHA256.fullmatch(sha256):
        raise ValueError(f"sha256 inutilisable dans un chemin : {sha256!r}")
    return Path(get_settings().import_files_dir) / org_id / f"{sha256}{suffix}"


@dataclass(frozen=True, slots=True)
class ImportSpec:
    """Le contrat de `jobs.payload` pour `import_csv` — écrit ici, consommé par T3.

    Réutilisé tel quel par l'import API (T8, `app.integrations.pipeline`) : les écritures
    (`write_batch`) ne dépendent que de la cible, de la locale, de la source et du
    `fetched_at` — le fichier n'intervient qu'à la lecture. `file_sha256` est donc **nul pour
    un import API**, qui n'a pas de fichier ; c'est la seule différence entre les deux specs,
    et l'exiger ferait dupliquer tout le moteur d'écriture pour rien.
    """

    import_id: uuid.UUID
    file_sha256: str | None
    source: str
    site_target_id: uuid.UUID
    locale: str
    fetched_at: datetime
    keyword_list_id: uuid.UUID | None = None
    options: Mapping[str, Any] | None = None
    #  ISO 3166-1 alpha-2 de la cible, figé à l'enfilage (§4.3) : les lignes d'un autre pays
    #  sont écartées. `None` — payload d'avant le cloisonnement dont la `locale` elle-même est
    #  illisible — laisse l'import se comporter comme avant : la cible fait autorité, seule.
    country: str | None = None


_REQUIRED_PAYLOAD = frozenset(
    {"import_id", "file_sha256", "source", "site_target_id", "locale", "fetched_at"}
)
# Forme du pays acceptée, la même que `ck_site_targets_country` : rien d'autre ne peut sortir
# de `site_targets`, et un pays mal formé ne doit jamais faire écarter la moitié d'un fichier.
_COUNTRY = re.compile(r"^[A-Za-z]{2}$")


def _payload_country(payload: Mapping[str, Any], locale: str) -> str | None:
    """Le pays de la cible : la clé `country`, ou le pays de la `locale` à défaut.

    Le repli existe pour les jobs **déjà en file** au moment du déploiement, dont le payload
    n'a pas la clé : `locale_of` (module Imports) écrit toujours `langue-PAYS`, la seconde
    moitié est donc exactement ce que `create_import` aurait mis dans `country`. Un job enfilé
    après le déploiement n'y passe jamais.
    """
    raw = payload.get("country")
    if raw is None:
        _, _, tail = locale.partition("-")
        raw = tail
    text = str(raw).strip()
    return text.upper() if _COUNTRY.fullmatch(text) else None


def parse_payload(payload: Any) -> ImportSpec:
    """Valide le payload. Un payload invalide est une **erreur de programmation** de l'api :
    elle fait échouer le job — et l'import avec lui, par `on_fail` — bruyamment."""
    if not isinstance(payload, Mapping):
        raise ValueError("payload d'import_csv invalide : objet attendu")
    if missing := _REQUIRED_PAYLOAD - payload.keys():
        raise ValueError(f"payload d'import_csv incomplet : {', '.join(sorted(missing))}")
    fetched_at = datetime.fromisoformat(str(payload["fetched_at"]))
    if fetched_at.tzinfo is None:
        # Sans fuseau, la même « date des données » désignerait deux instants selon le
        # serveur — et `(keyword_id, source, fetched_at)` cesserait d'identifier une
        # observation.
        raise ValueError("fetched_at doit porter un fuseau horaire")
    list_id = payload.get("keyword_list_id")
    locale = str(payload["locale"])
    return ImportSpec(
        import_id=uuid.UUID(str(payload["import_id"])),
        file_sha256=str(payload["file_sha256"]),
        source=str(payload["source"]),
        site_target_id=uuid.UUID(str(payload["site_target_id"])),
        locale=locale,
        fetched_at=fetched_at,
        keyword_list_id=None if list_id is None else uuid.UUID(str(list_id)),
        options=payload.get("options"),
        country=_payload_country(payload, locale),
    )


@dataclass(slots=True)
class ScanCounters:
    """Ce que la lecture du fichier établit — **déterministe**, jamais porté par `progress`.

    `metrics_deduplicated_in_file` isole le doublon **attendu** : Semrush Positions émet une
    ligne par (mot-clé, URL), donc un mot-clé qui classe sur trois pages produit deux
    métriques ignorées — c'est normal, la cannibalisation vit dans les maps (§4.5). À ne pas
    confondre avec `rows_duplicated_in_file`, les vraies lignes en double.

    Les quatre champs du cloisonnement (§4.3) sont ici et **jamais** dans `WriteCounters` :
    ils sont recalculés intégralement à chaque run, lots sautés compris, donc le rapport d'un
    import repris est celui d'un import d'une traite. Les deux motifs restent séparés — « pays
    d'un autre marché » et « pays non reconnu » — parce que le second est le seul signal qu'on
    ait d'un trou dans la table d'alias (`app.keywords.countries`) : le confondre avec le
    premier le rendrait invisible.
    """

    rows_read: int = 0
    rows_error: int = 0
    metrics_deduplicated_in_file: int = 0
    rows_duplicated_in_file: int = 0
    #  Lignes écartées parce que leur pays est **résolu** et n'est pas celui de la cible.
    rows_off_target: int = 0
    #  Lignes écartées parce que leur pays n'est **pas résolu** : localité, libellé inconnu,
    #  cellule vide. Une cellule vide compte ici comme ligne mais n'entre dans aucun
    #  dictionnaire — il n'y a pas de libellé à nommer à l'écran (même règle qu'au recensement
    #  de l'aperçu). D'où `sum(unresolved_by_country.values()) <= rows_country_unresolved`.
    rows_country_unresolved: int = 0
    #  Libellé **brut** du fichier → nombre de lignes, pour le texte du §4.5 (« hors cible :
    #  United Kingdom 55, Australia 32… »). Brut et non traduit : c'est ce que l'utilisateur
    #  lit dans son propre fichier.
    ignored_by_country: dict[str, int] = field(default_factory=dict)
    unresolved_by_country: dict[str, int] = field(default_factory=dict)


@dataclass(slots=True)
class WriteCounters:
    """Ce que les écritures ont changé — restauré depuis `progress` à la reprise."""

    rows_inserted: int = 0
    keywords_created: int = 0
    metrics_written: int = 0
    list_items_added: int = 0
    maps_written: int = 0

    @classmethod
    def from_dict(cls, raw: Any) -> "WriteCounters":
        if not isinstance(raw, Mapping):
            return cls()
        known = {f.name: raw.get(f.name) for f in fields(cls)}
        return cls(**{name: value for name, value in known.items() if isinstance(value, int)})

    def as_dict(self) -> dict[str, int]:
        return asdict(self)


@dataclass(slots=True)
class _Batch:
    """Un lot prêt à écrire : lignes canoniques, URL normalisées, dédup intra-fichier faite."""

    rows: list[CanonicalKeywordRow] = field(default_factory=list)
    #  Rang de la ligne dans le lot → URL normalisée (absent si la ligne n'en porte pas, ou
    #  si la paire (mot-clé, URL) est déjà revendiquée par une ligne antérieure du lot).
    urls: dict[int, str] = field(default_factory=dict)
    #  Rangs des lignes qui **portent** la métrique de leur mot-clé (première occurrence).
    carriers: set[int] = field(default_factory=set)


def _chunks(rows: Sequence[Any], columns: int) -> Iterator[Sequence[Any]]:
    """Découpe des VALUES pour rester sous le plafond de paramètres liés d'asyncpg."""
    size = max(1, MAX_BOUND_PARAMS // max(columns, 1))
    for start in range(0, len(rows), size):
        yield rows[start : start + size]


# --- écriture d'un lot ------------------------------------------------------------------


async def _keyword_ids(
    session: AsyncSession, org_id: str, locale: str, keywords: Sequence[str]
) -> dict[str, uuid.UUID]:
    """Relit les identifiants des mots-clés du lot, **filtrés par organisation**.

    C'est ici que se joue l'invariant M3 : un mot-clé d'une autre organisation ne peut pas
    entrer dans le mapping, donc aucune métrique ne peut lui être rattachée.
    """
    found: dict[str, uuid.UUID] = {}
    for chunk in _chunks(keywords, 1):
        rows = await session.execute(
            select(Keyword.keyword, Keyword.id).where(
                Keyword.org_id == org_id,
                Keyword.locale == locale,
                Keyword.keyword.in_(chunk),
            )
        )
        found.update({keyword: keyword_id for keyword, keyword_id in rows})
    return found


async def _insert_keywords(
    session: AsyncSession, org_id: str, locale: str, keywords: Sequence[str]
) -> set[str]:
    created: set[str] = set()
    for chunk in _chunks(keywords, 3):
        statement = (
            pg_insert(Keyword)
            .values([{"org_id": org_id, "locale": locale, "keyword": k} for k in chunk])
            .on_conflict_do_nothing(index_elements=["org_id", "locale", "keyword"])
            .returning(Keyword.keyword)
        )
        created.update((await session.scalars(statement)).all())
    return created


# Colonnes qu'un re-import correctif remplace (§4.3). `org_id`, `keyword_id`, `source` et
# `fetched_at` n'y sont pas : ce sont l'identité de l'observation, jamais sa valeur.
_METRIC_VALUES = ("volume", "kd", "cpc", "intent", "serp_features", "trend", "raw")


async def _upsert_metrics(
    session: AsyncSession, values: Sequence[dict[str, Any]]
) -> set[uuid.UUID]:
    """Upsert des observations. Le `DO UPDATE` est **gardé** : rien n'est réécrit à l'identique.

    Sans cette garde, rejouer un import de 500 000 lignes réécrirait 500 000 lignes pour rien
    — bloat, WAL, autovacuum — et surtout le rapport compterait « insérées » des lignes qui
    n'ont rien changé, en contradiction directe avec §4.4. `RETURNING` ne rend donc que les
    observations réellement écrites : c'est la mesure du changement, pas de l'effort.
    """
    touched: set[uuid.UUID] = set()
    for chunk in _chunks(values, 11):
        insert = pg_insert(KeywordMetric).values(list(chunk))
        excluded = insert.excluded
        statement = insert.on_conflict_do_update(
            index_elements=["keyword_id", "source", "fetched_at"],
            set_={column: excluded[column] for column in _METRIC_VALUES},
            where=or_(
                *(
                    getattr(KeywordMetric, column).is_distinct_from(excluded[column])
                    for column in _METRIC_VALUES
                )
            ),
        ).returning(KeywordMetric.keyword_id)
        touched.update((await session.scalars(statement)).all())
    return touched


async def _insert_list_items(
    session: AsyncSession, values: Sequence[dict[str, Any]]
) -> set[uuid.UUID]:
    added: set[uuid.UUID] = set()
    for chunk in _chunks(values, 3):
        statement = (
            pg_insert(KeywordListItem)
            .values(list(chunk))
            .on_conflict_do_nothing(index_elements=["list_id", "keyword_id"])
            .returning(KeywordListItem.keyword_id)
        )
        added.update((await session.scalars(statement)).all())
    return added


async def _upsert_maps(
    session: AsyncSession, values: Sequence[dict[str, Any]]
) -> set[tuple[uuid.UUID, str]]:
    """Upsert des attaches observées (`source='import_url'`, §4.3).

    « Une observation se rafraîchit, une décision jamais » : le `DO UPDATE` est gardé par
    `EXCLUDED.seen_at > seen_at`, donc une observation **plus ancienne** ne rétrograde jamais
    une position déjà vue (review I2), et rejouer le même fichier est un no-op strict. Le map
    `manual` n'est jamais touché : il porte une autre `source`, donc une autre ligne — la
    décision de planification de l'utilisateur est hors d'atteinte de l'automatique.
    """
    touched: set[tuple[uuid.UUID, str]] = set()
    for chunk in _chunks(values, 8):
        insert = pg_insert(KeywordPageMap).values(list(chunk))
        excluded = insert.excluded
        statement = insert.on_conflict_do_update(
            index_elements=["site_target_id", "keyword_id", "url_normalized", "source"],
            set_={"position": excluded.position, "seen_at": excluded.seen_at},
            where=excluded.seen_at > KeywordPageMap.seen_at,
        ).returning(KeywordPageMap.keyword_id, KeywordPageMap.url_normalized)
        touched.update((keyword_id, url) for keyword_id, url in await session.execute(statement))
    return touched


async def write_batch(
    session: AsyncSession, job: Job, spec: ImportSpec, batch: _Batch, counters: WriteCounters
) -> None:
    """Écrit un lot et met les compteurs d'écriture à jour.

    **Ne commit pas** : c'est `save_progress` qui termine la transaction, pour que le lot et
    son curseur soient publiés d'un seul geste (propriété 2 de l'en-tête du module).
    """
    org_id = job.org_id
    keywords = sorted({row.keyword for row in batch.rows})
    created = await _insert_keywords(session, org_id, spec.locale, keywords)
    ids = await _keyword_ids(session, org_id, spec.locale, keywords)

    touched_metrics = await _upsert_metrics(
        session,
        [
            {
                "org_id": org_id,
                "keyword_id": ids[batch.rows[index].keyword],
                "source": spec.source,
                "fetched_at": spec.fetched_at,
                "volume": batch.rows[index].volume,
                "kd": batch.rows[index].kd,
                "cpc": batch.rows[index].cpc,
                "intent": batch.rows[index].intent,
                "serp_features": batch.rows[index].serp_features,
                "trend": batch.rows[index].trend,
                "raw": batch.rows[index].raw,
            }
            for index in sorted(batch.carriers)
        ],
    )

    added_items: set[uuid.UUID] = set()
    if spec.keyword_list_id is not None:
        added_items = await _insert_list_items(
            session,
            [
                {"list_id": spec.keyword_list_id, "keyword_id": ids[keyword], "org_id": org_id}
                for keyword in keywords
            ],
        )

    #  Point d'extension du module Crawl (§3.9) : no-op propre tant que `pages` n'existe pas.
    pages = await resolution.resolve_page_ids(
        session, spec.site_target_id, set(batch.urls.values())
    )
    touched_maps = await _upsert_maps(
        session,
        [
            {
                "org_id": org_id,
                "site_target_id": spec.site_target_id,
                "keyword_id": ids[batch.rows[index].keyword],
                "url_normalized": url,
                "page_id": pages.get(url),
                "source": "import_url",
                "position": batch.rows[index].position,
                "seen_at": spec.fetched_at,
            }
            for index, url in sorted(batch.urls.items())
        ],
    )

    counters.keywords_created += len(created)
    counters.metrics_written += len(touched_metrics)
    counters.list_items_added += len(added_items)
    counters.maps_written += len(touched_maps)
    #  Une ligne est « insérée » si **elle** a changé quelque chose. La création du mot-clé,
    #  sa métrique et son appartenance de liste sont portées par la ligne qui les a produites
    #  (la première du mot-clé) : sans cela, un mot-clé sur trois URL gonflerait le compte de
    #  trois. Chaque ligne garde en propre son attache.
    for index, row in enumerate(batch.rows):
        keyword_id = ids[row.keyword]
        url = batch.urls.get(index)
        changed = (url is not None and (keyword_id, url) in touched_maps) or (
            index in batch.carriers
            and (
                row.keyword in created or keyword_id in touched_metrics or keyword_id in added_items
            )
        )
        if changed:
            counters.rows_inserted += 1


# --- cloisonnement par pays ---------------------------------------------------------------


@dataclass(frozen=True, slots=True)
class _CountryFilter:
    """De quoi trancher une ligne : l'en-tête **réel** qui porte son pays, et le pays gardé."""

    header: str
    country: str


def _country_filter(adapter: Any, spec: ImportSpec, path: Path) -> _CountryFilter | None:
    """Le cloisonnement de cet import, ou `None` quand il n'y a rien à cloisonner (§4.3).

    Trois façons de ne rien filtrer, et c'est bien le comportement d'avant le cloisonnement
    qu'elles rendent — la cible fait autorité, seule, et toutes les lignes entrent :

    1. l'adaptateur ne déclare pas de colonne pays (tout Semrush, `generic_csv` sans
       `country_column`) — le fichier n'écrit pas son pays, il n'y a rien à lire ;
    2. le payload ne donne pas de pays exploitable (voir `_payload_country`) ;
    3. la colonne déclarée est **absente du fichier**. Un export KWFinder dont l'utilisateur a
       masqué `Location` à l'export reste un export KWFinder parfaitement importable : c'est
       la même issue qu'au recensement de l'aperçu, qui rend alors `[]` plutôt que de refuser.
       Le cas est journalisé — ne rien filtrer en silence quand on avait promis de le faire
       serait la seule façon de perdre des lignes sans rien en dire.

    L'en-tête **déclaré** par l'adaptateur n'est pas forcément l'en-tête **réel** : il est
    résolu ici, une fois, avec la même tolérance que le mapping (`normalize_header` : casse,
    tirets, espaces). `CanonicalKeywordRow.raw` étant indexé par l'en-tête réel du fichier,
    chercher le déclaré rendrait `None` sur tout export dont la casse diffère — c'est-à-dire
    filtrerait tout le fichier.
    """
    declared = adapter.country_header(spec.options)
    if declared is None or spec.country is None:
        return None
    #  Le fichier plutôt que la première ligne : `raw` est un `zip(headers, values,
    #  strict=False)`, donc une ligne tronquée n'a pas toutes les clés et déduire l'en-tête
    #  d'elle laisserait le reste du fichier sans pays. Une ouverture de plus par import qui a
    #  quelque chose à cloisonner, jamais pour les autres.
    real = header_index(read_header(path)).get(normalize_header(declared))
    if real is None:
        logger.warning(
            "import %s : colonne pays %r absente du fichier, aucun cloisonnement",
            spec.import_id,
            declared,
        )
        return None
    return _CountryFilter(header=real, country=spec.country)


# Le « pays de la ligne » d'une source API : `CanonicalKeywordRow.raw` porte un objet sous cette
# clé, pas une cellule de texte, et c'est `LABEL_KEY` qui donne le libellé à résoudre. Nommés
# ici plutôt qu'importés d'un adaptateur : ce module ne connaît aucun fournisseur, et il n'a pas
# à en connaître un pour lire deux clés.
API_COUNTRY_FIELD = "location"
API_COUNTRY_LABEL_KEY = "label"


def api_country_filter(country: str | None) -> _CountryFilter | None:
    """Le cloisonnement d'un import de **liste KWFinder**, ou `None` sans pays de cible.

    Le pendant de `_country_filter` pour le chemin API, et il n'a pas de fichier à ouvrir : le
    « pays de la ligne » n'est pas une cellule mais l'objet `location` que le fournisseur
    attache à chaque mot-clé (`app.keywords.sources.mangools_api`). Une liste enregistrée peut
    mélanger plusieurs marchés — le compte relevé en a des « USA » et des « France » —, donc
    ce filtre est exactement aussi nécessaire ici que sur un CSV KWFinder à onze pays.

    Les deux autres modes API (domaine, mots-clés nommés) n'en ont **pas** besoin et ne
    doivent pas l'obtenir : ils passent un `location_id` en entrée, leur réponse ne contient
    déjà que le marché demandé.
    """
    code = (country or "").strip().upper()
    if not code:
        return None
    return _CountryFilter(header=API_COUNTRY_FIELD, country=code)


def _keeps(row: CanonicalKeywordRow, wanted: _CountryFilter, counters: ScanCounters) -> bool:
    """La ligne appartient-elle au marché de la cible ? Sinon, elle est **ventilée** puis jetée.

    Jamais une `RowError` : une ligne d'un autre pays n'est pas une ligne fautive, elle
    n'appartient simplement pas à cet import — elle ne doit ni gonfler `rows_error`, ni
    apparaître dans `errors.csv`, ni faire croire à l'utilisateur que son fichier est cassé.

    Deux formes de « pays de la ligne », une seule résolution : une **cellule** de CSV, et
    l'**objet** `location` d'une réponse d'API. `str()` sur le second rendrait
    `"{'_id': 2840, …}"`, que `countries.resolve` refuserait — tout un import de liste
    partirait alors en « pays non reconnu », sans erreur ni message.
    """
    cell = row.raw.get(wanted.header)
    if isinstance(cell, Mapping):
        label = cell.get(API_COUNTRY_LABEL_KEY)
        raw = label.strip() if isinstance(label, str) else ""
    else:
        raw = str(cell or "").strip()
    code = countries.resolve(raw)
    if code == wanted.country:
        return True
    if code is None:
        counters.rows_country_unresolved += 1
        if raw:
            counters.unresolved_by_country[raw] = counters.unresolved_by_country.get(raw, 0) + 1
    else:
        counters.rows_off_target += 1
        counters.ignored_by_country[raw] = counters.ignored_by_country.get(raw, 0) + 1
    return False


# --- lecture et découpage en lots --------------------------------------------------------


def batches(
    parsed: Iterable[CanonicalKeywordRow | RowError],
    counters: ScanCounters,
    errors: list[RowError],
    wanted: _CountryFilter | None = None,
) -> Iterator[_Batch]:
    """Découpe le flux de l'adaptateur en lots, dédup intra-fichier comprise.

    La dédup n'est pas une optimisation : `ON CONFLICT DO UPDATE` **refuse** d'affecter deux
    fois la même ligne dans une seule instruction (« cannot affect row a second time »). Un
    export Positions, qui répète le mot-clé à chaque URL, ferait échouer l'insertion des
    métriques sans elle — un import entier perdu sur un doublon parfaitement normal.

    `seen` **et** `pairs` sont globaux au fichier, pas au lot. Pour `seen`, parce que les
    occurrences d'un même mot-clé ne sont pas forcément adjacentes et que sa métrique n'est
    écrite qu'une fois par import. Pour `pairs`, parce qu'un `pairs` remis à zéro à chaque lot
    ferait dépendre le **rapport** du découpage : une ligne strictement répétée dont la
    première occurrence est dans un lot antérieur serait comptée « doublon attendu » (le
    mot-clé multi-URL de Semrush Positions) au lieu de « vrai doublon », et le chiffre
    changerait avec `BATCH_SIZE`. Rien ne se perd à les garder : dans un même fichier, la paire
    répétée porte le même `fetched_at`, donc son upsert n'avait de toute façon rien à
    rafraîchir — et la protection « cannot affect row a second time » reste assurée, un
    surensemble de la dédup par lot.

    **Le cloisonnement par pays se fait ici**, et nulle part ailleurs : c'est la seule boucle
    qui voit chaque ligne, et l'adaptateur, lui, ne connaît ni la cible ni l'organisation
    (§5). Il tombe **après** `rows_read` — sinon le rapport mentirait sur « lignes lues » — et
    **avant tout le reste**, `seen` et `pairs` compris. Cet ordre n'est pas un détail : une
    ligne d'un autre pays qui entrerait dans `seen` volerait le rôle de **porteur de métrique**
    à la ligne du bon pays rencontrée plus loin (les exports groupent par marché, l'ordre joue
    donc contre nous), et le mot-clé serait importé sans son volume, sans un mot dans le
    rapport. Il tombe aussi avant la normalisation d'URL : une URL invalide sur une ligne qui
    n'appartient pas à cet import n'est pas une erreur de cet import.
    """
    seen: set[str] = set()
    pairs: set[tuple[str, str]] = set()
    batch = _Batch()
    for item in parsed:
        counters.rows_read += 1
        if isinstance(item, RowError):
            counters.rows_error += 1
            if len(errors) < ERRORS_CSV_LIMIT:
                errors.append(item)
            continue
        if wanted is not None and not _keeps(item, wanted, counters):
            continue
        url: str | None = None
        if item.url:
            try:
                url = normalize_url(item.url)
            except UrlError as exc:
                #  Contrat §3.8 : une URL invalide fait une ligne en erreur à l'import.
                counters.rows_error += 1
                if len(errors) < ERRORS_CSV_LIMIT:
                    errors.append(RowError(item.line, f"URL invalide : {exc}"))
                continue
        index = len(batch.rows)
        batch.rows.append(item)
        carrier = item.keyword not in seen
        if carrier:
            seen.add(item.keyword)
            batch.carriers.add(index)
        else:
            counters.metrics_deduplicated_in_file += 1
        if url is not None and (item.keyword, url) not in pairs:
            pairs.add((item.keyword, url))
            batch.urls[index] = url
        elif not carrier:
            #  Ni métrique ni attache nouvelles : la ligne est un vrai doublon, pas la
            #  seconde URL d'un mot-clé.
            counters.rows_duplicated_in_file += 1
        if len(batch.rows) >= BATCH_SIZE:
            yield batch
            batch = _Batch()
    if batch.rows:
        yield batch


# --- handler -----------------------------------------------------------------------------


async def set_import_status(
    session: AsyncSession, import_id: uuid.UUID, org_id: str, status: str, **values: Any
) -> None:
    """Recopie du statut du job dans `imports` (§3.11) — le worker, lui, reste neutre."""
    await session.execute(
        update(Import)
        .where(Import.id == import_id, Import.org_id == org_id)
        .values(status=status, **values)
        .execution_options(synchronize_session=False)
    )


def _write_errors_csv(spec: ImportSpec, org_id: str, errors: Sequence[RowError]) -> str | None:
    """`<sha256>.errors.csv` à côté du fichier source (§3.12) : même rétention, même purge."""
    if not errors:
        return None
    path = import_file_path(org_id, spec.file_sha256, suffix=".errors.csv")
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.writer(handle)
        writer.writerow(("line", "reason"))
        writer.writerows((error.line, error.reason) for error in errors)
    return path.name


async def run_import_csv(session: AsyncSession, job: Job) -> Mapping[str, Any]:
    """Handler du job `import_csv` (§4). Rend le rapport final, recopié dans `imports`."""
    spec = parse_payload(job.payload)
    adapter = sources.get(spec.source)
    if adapter is None:
        raise ValueError(f"adaptateur inconnu : {spec.source}")
    path = import_file_path(job.org_id, spec.file_sha256)

    progress = job.progress if isinstance(job.progress, Mapping) else {}
    cursor = progress.get("batch")
    start_batch = cursor if isinstance(cursor, int) and cursor > 0 else 0
    scan = ScanCounters()
    written = WriteCounters.from_dict(progress.get("counters"))
    errors: list[RowError] = []

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

    #  La lecture DuckDB et le parse sont **synchrones et lourds** : les exécuter dans la
    #  boucle la bloquerait pour toute la durée d'un lot (§4.1, « tout dans le worker, en
    #  thread »). Un aller-retour de thread par lot — cent pour 500 000 lignes — ne coûte
    #  rien à côté du travail qu'il déporte. `_country_filter` ouvre le fichier lui aussi
    #  (en-têtes seuls) : même règle, et seulement quand il y a un pays à lire.
    wanted = await asyncio.to_thread(_country_filter, adapter, spec, path)
    stream = batches(adapter.parse_csv(path, options=spec.options), scan, errors, wanted)
    index = -1
    while (batch := await asyncio.to_thread(next, stream, None)) is not None:
        index += 1
        if index < start_batch:
            #  Lot déjà committé par un run précédent : relu pour avancer dans le fichier et
            #  pour recompter ce que la lecture établit, jamais réécrit.
            continue
        await write_batch(session, job, 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:
            #  Dépossédé par le sweep : `save_progress` a déjà annulé le lot en vol.
            logger.warning("import %s : job repris par un autre worker, arrêt", spec.import_id)
            return {"dispossessed": True, **build_report(scan, written, _numbered(errors), None)}

    #  Les numéros de ligne ne sont ramenés au repère physique qu'ici : la liste des lignes
    #  écartées par le lecteur n'est connue qu'une fois le fichier parcouru (§ `RowError`).
    numbered = _numbered(errors)
    report = build_report(scan, written, numbered, _write_errors_csv(spec, job.org_id, numbered))
    await set_import_status(session, spec.import_id, job.org_id, "done", report=dict(report))
    await session.commit()
    return report


def _numbered(errors: Sequence[RowError]) -> list[RowError]:
    """Toutes les erreurs dans le repère des lignes physiques, triées par numéro.

    Les lignes écartées par le lecteur CSV sont celles qui portent déjà un numéro physique :
    ce sont elles qui décalent les autres.
    """
    rejected = [error.line for error in errors if error.physical]
    return sorted(to_physical_lines(errors, rejected), key=lambda error: error.line)


def _by_country(counted: Mapping[str, int]) -> dict[str, int]:
    """Le décompte par libellé, du plus fréquent au moins fréquent puis par ordre alphabétique."""
    return dict(sorted(counted.items(), key=lambda item: (-item[1], item[0])))


def build_report(
    scan: ScanCounters,
    written: WriteCounters,
    errors: Sequence[RowError],
    errors_csv: str | None,
) -> dict[str, Any]:
    return {
        "rows_read": scan.rows_read,
        "rows_inserted": written.rows_inserted,
        #  Ce qui n'a rien changé : le rejeu d'un fichier déjà importé les met toutes ici.
        #  **Reste le TOTAL** après le cloisonnement (§4.5) : les deux motifs ci-dessous en
        #  sont des sous-ensembles, jamais des colonnes qui s'ajouteraient à lui. D'où
        #  l'invariant `rows_ignored >= rows_off_target + rows_country_unresolved`, vrai par
        #  construction — une ligne écartée par le pays est lue, n'est pas une erreur et n'est
        #  jamais insérée — et vrai aussi après une reprise, où `rows_inserted` seul est
        #  restauré tandis que la lecture est refaite en entier.
        "rows_ignored": scan.rows_read - scan.rows_error - written.rows_inserted,
        "rows_error": scan.rows_error,
        #  Cloisonnement par pays (§4.3, §4.5). Toujours présents, à zéro pour les sources qui
        #  n'écrivent pas leur pays : un rapport qui tait ses compteurs se lit « inconnu », un
        #  rapport qui les met à zéro se lit « rien à signaler ».
        "rows_off_target": scan.rows_off_target,
        "rows_country_unresolved": scan.rows_country_unresolved,
        #  Libellés bruts, du plus fréquent au moins fréquent. L'ordre est une courtoisie, pas
        #  un contrat : ces rapports finissent en `jsonb`, qui range ses clés à sa façon.
        "ignored_by_country": _by_country(scan.ignored_by_country),
        "unresolved_by_country": _by_country(scan.unresolved_by_country),
        "keywords_created": written.keywords_created,
        "metrics_written": written.metrics_written,
        "list_items_added": written.list_items_added,
        "maps_written": written.maps_written,
        "metrics_deduplicated_in_file": scan.metrics_deduplicated_in_file,
        "rows_duplicated_in_file": scan.rows_duplicated_in_file,
        #  `physical` est un détail de calcul du numéro : le rapport n'expose que le
        #  résultat, dans le repère du fichier de l'utilisateur.
        "errors": [
            {"line": error.line, "reason": error.reason} for error in errors[:ERROR_REPORT_LIMIT]
        ],
        "errors_total": len(errors),
        "errors_csv": errors_csv,
        #  Faux seulement si le fichier a dépassé le plafond de collecte : `errors.csv` est
        #  alors tronqué, et le rapport le dit plutôt que de laisser croire à l'exhaustivité.
        "errors_complete": len(errors) < ERRORS_CSV_LIMIT,
    }


async def mark_import_failed(session: AsyncSession, job: Job) -> None:
    """Callback `on_fail` du registre (§3.11) : recopie l'échec dans la ligne `imports`.

    Invoqué par le worker après **tout** passage à `failed` — exception du handler comme
    `failed` définitif posé par le sweep des zombies au plafond de tentatives. C'est ce qui
    garantit qu'aucun import ne reste `queued`/`running` à perpétuité, sans que le socle ait
    à connaître la table `imports`.
    """
    payload = job.payload if isinstance(job.payload, Mapping) else {}
    raw_id = payload.get("import_id")
    if raw_id is None:
        #  Payload illisible : il n'y a pas d'import à mettre en cohérence. Le job, lui, est
        #  déjà `failed` avec son erreur — rien n'est perdu.
        return
    await set_import_status(session, uuid.UUID(str(raw_id)), job.org_id, "failed", error=job.error)


registry.register(
    registry.JobType(
        JOB_TYPE,
        run_import_csv,
        #  `'active'` (socle §1) : un import est un travail **long et exclusif**. La clé n'est
        #  relâchée qu'à `done`/`failed`, donc jamais deux instances du même import en vol —
        #  deux runs concurrents sur les mêmes unicités se bloqueraient l'un l'autre pour
        #  aboutir au même état. Rien à voir avec `map_reconcile`, bon marché et `'queue'`.
        dedup_scope="active",
        on_fail=mark_import_failed,
    )
)