"""Pipeline d'import CSV de bout en bout (spec Imports §4, §11).

Le job est toujours exécuté par le **vrai worker** (`Worker.run_once`), jamais par un appel
direct au handler : ce qui est éprouvé ici, c'est le chemin complet — claim, registre,
handler, `on_fail`, recopie du statut. Un test qui court-circuiterait le worker laisserait
passer exactement le genre de bug que le rapport T0 annonçait (type non enregistré).

Les promesses vérifiées, dans l'ordre du §4 :

- les compteurs du rapport disent la vérité, doublons intra-fichier **attendus** compris ;
- **rejouer le même fichier ne touche pas une seule ligne** — prouvé par `xmin`, pas par
  une simple égalité de valeurs : un `UPDATE` qui réécrit à l'identique se verrait ;
- **le re-import correctif** remplace l'observation de même clé, et elle seule (C1) ;
- une observation plus récente rafraîchit le map, une plus ancienne ne le rétrograde jamais
  (I2), et le map `manual` est hors d'atteinte de l'automatique ;
- l'invariant `keyword_metrics.org_id = keywords.org_id` (dette M3 de T1) ;
- `imports.status` suit le job sur **tous** les chemins, succès comme échec (I3) ;
- **le cloisonnement par pays** (spec « Multi-sources et cloisonnement pays » §4.3) : un
  fichier de onze marchés importé sur `en-US` puis sur `en-GB` donne deux imports disjoints,
  et une ligne écartée ne prend jamais la place de celle qu'on gardait.
"""

import csv
import uuid
from collections.abc import AsyncIterator, Callable, Sequence
from dataclasses import fields
from datetime import UTC, datetime, timedelta
from decimal import Decimal
from pathlib import Path
from typing import Any

import pytest
import pytest_asyncio
from sqlalchemy import delete, func, select, text
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from app.db.engine import get_engine
from app.jobs import service
from app.jobs.models import Job
from app.jobs.worker import Worker
from app.keywords import pipeline
from app.keywords.models import (
    Import,
    Keyword,
    KeywordList,
    KeywordListItem,
    KeywordMetric,
    KeywordPageMap,
)
from tests.conftest import Fixture, MakeTarget

FIXTURES = Path(__file__).parent / "fixtures"
POSITIONS_EN = FIXTURES / "semrush_positions_en.csv"
MAGIC_REAL = FIXTURES / "real" / "semrush_magic_us.csv"
# Export KWFinder réel de onze marchés (§1) : la seule fixture du dépôt qui porte plusieurs
# pays, et dont six n'apparaissent qu'après la 116ᵉ ligne.
MULTI_COUNTRY = FIXTURES / "real" / "kwfinder_multi_country.csv"

FETCHED_AT = datetime(2026, 8, 30, 12, 0, tzinfo=UTC)
LATER = FETCHED_AT + timedelta(days=7)
EARLIER = FETCHED_AT - timedelta(days=7)


def _sha(seed: str) -> str:
    """Un sha256 plausible : le pipeline le refuse s'il n'a pas la bonne forme (§3.12)."""
    return uuid.uuid5(uuid.NAMESPACE_URL, seed).hex * 2


@pytest_asyncio.fixture(autouse=True)
async def _clean_import_jobs(db: AsyncSession) -> AsyncIterator[None]:
    """Le worker prend le plus vieux job `queued` **toutes organisations confondues** : un
    reliquat d'un test précédent serait exécuté à la place de celui du test courant."""
    await _purge(db)
    yield
    await _purge(db)


async def _purge(db: AsyncSession) -> None:
    leftovers = select(Job.id).where(Job.type == pipeline.JOB_TYPE)
    await db.execute(delete(Import).where(Import.job_id.in_(leftovers)))
    await db.execute(delete(Job).where(Job.type == pipeline.JOB_TYPE))
    await db.commit()


Stage = Callable[..., Any]


#  `country` non précisé = déduit de la locale, comme le fait `create_import`. `None`, lui,
#  demande explicitement un payload **sans la clé** : un job enfilé avant le cloisonnement.
_FROM_LOCALE = "…"


@pytest.fixture
def stage(db: AsyncSession, sandbox: Fixture, import_dir: Path) -> Stage:
    """Prépare un import comme le fera T3 : fichier sur disque, liste, job, ligne `imports`."""

    async def _stage(
        fixture: Path | str,
        *,
        source: str = "semrush_positions",
        fetched_at: datetime = FETCHED_AT,
        sha: str | None = None,
        list_name: str | None = "Semrush — 2026-08-30",
        options: dict[str, Any] | None = None,
        target_id: uuid.UUID | None = None,
        locale: str = "fr-FR",
        country: str | None = _FROM_LOCALE,
    ) -> tuple[Job, Import]:
        sha = sha or _sha(str(fixture))
        target_id = target_id or sandbox.target_id
        _place(import_dir / sandbox.org_id / f"{sha}.csv", fixture)

        list_id: uuid.UUID | None = None
        if list_name is not None:
            existing = await db.scalar(
                select(KeywordList).where(
                    KeywordList.site_target_id == target_id, KeywordList.name == list_name
                )
            )
            if existing is None:
                existing = KeywordList(
                    org_id=sandbox.org_id,
                    site_target_id=target_id,
                    name=list_name,
                    source=source,
                )
                db.add(existing)
                await db.commit()
            list_id = existing.id

        import_id = uuid.uuid4()
        payload: dict[str, Any] = {
            "import_id": str(import_id),
            "file_sha256": sha,
            "source": source,
            "site_target_id": str(target_id),
            "locale": locale,
            "fetched_at": fetched_at.isoformat(),
            "keyword_list_id": None if list_id is None else str(list_id),
            "options": options,
        }
        resolved = locale.split("-")[-1] if country == _FROM_LOCALE else country
        if resolved is not None:
            payload["country"] = resolved
        job = await service.enqueue(
            db,
            sandbox.org_id,
            pipeline.JOB_TYPE,
            payload=payload,
            dedup_key=f"{pipeline.JOB_TYPE}:{target_id}:{sha}:{fetched_at.isoformat()}",
        )
        row = Import(
            id=import_id,
            org_id=sandbox.org_id,
            site_target_id=target_id,
            source=source,
            kind="csv",
            filename=str(fixture),
            file_sha256=sha,
            fetched_at=fetched_at,
            keyword_list_id=list_id,
            job_id=job.id,
            status="queued",
        )
        db.add(row)
        await db.commit()
        return job, row

    return _stage


def _remove(path: Path) -> None:
    path.unlink()


def _place(path: Path, fixture: Path | str) -> None:
    """Dépose le fichier là où l'api l'aurait spoolé (§3.12)."""
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(fixture.read_bytes() if isinstance(fixture, Path) else fixture.encode())


async def _run_worker() -> None:
    """Un tour de worker sur sa propre session, comme en production."""
    worker = Worker(async_sessionmaker(get_engine(), expire_on_commit=False))
    assert await worker.run_once(), "aucun job pris : la file était vide"


async def _reload(db: AsyncSession, job: Job, row: Import) -> tuple[Job, Import]:
    #  Les identifiants sont capturés **avant** `expire_all` : les relire après déclencherait
    #  un chargement paresseux impossible en asynchrone.
    job_id, import_id = job.id, row.id
    db.expire_all()
    fresh_job = await db.get(Job, job_id)
    fresh_import = await db.get(Import, import_id)
    assert fresh_job is not None and fresh_import is not None
    return fresh_job, fresh_import


async def _snapshot(db: AsyncSession, org_id: str) -> dict[str, Sequence[Any]]:
    """L'état complet du réservoir de l'org, `xmin` compris.

    `xmin` est le numéro de transaction qui a écrit chaque version de ligne : deux snapshots
    qui le partagent prouvent qu'**aucune ligne n'a été réécrite**, pas seulement que les
    valeurs finales coïncident. C'est la seule façon de démontrer « état strictement
    identique » (§4.4) plutôt qu'« état équivalent ».
    """
    statements = {
        "keywords": "SELECT id, keyword, xmin::text FROM keywords WHERE org_id = :org ORDER BY id",
        "metrics": (
            "SELECT id, keyword_id, source, fetched_at, volume, kd, cpc, intent, trend, raw, "
            "xmin::text FROM keyword_metrics WHERE org_id = :org ORDER BY id"
        ),
        "items": (
            "SELECT list_id, keyword_id, xmin::text FROM keyword_list_items "
            "WHERE org_id = :org ORDER BY list_id, keyword_id"
        ),
        "maps": (
            "SELECT id, keyword_id, url_normalized, source, position, seen_at, page_id, "
            "xmin::text FROM keyword_page_map WHERE org_id = :org ORDER BY id"
        ),
    }
    return {
        name: (await db.execute(text(sql), {"org": org_id})).all()
        for name, sql in statements.items()
    }


# --- compteurs du rapport -----------------------------------------------------------------


async def test_positions_import_reports_exact_counts(
    db: AsyncSession, sandbox: Fixture, stage: Stage
) -> None:
    job, row = await stage(POSITIONS_EN)
    await _run_worker()
    job, row = await _reload(db, job, row)

    assert job.status == "done"
    report = job.report
    assert report is not None
    # 7 lignes de données : 5 exploitables, 2 en erreur (mot-clé vide, URL non http(s)).
    assert report["rows_read"] == 7
    assert report["rows_error"] == 2
    # Insérées = les lignes qui ont **changé** quelque chose. La 4ᵉ ligne répète à
    # l'identique la paire (mot-clé, URL) de la 2ᵉ : elle ne change rien.
    assert report["rows_inserted"] == 4
    assert report["rows_ignored"] == 1
    assert report["keywords_created"] == 3
    assert report["metrics_written"] == 3
    assert report["list_items_added"] == 3
    assert report["maps_written"] == 4
    # Doublon **attendu** : « chaussures rouges » occupe deux URL, plus une ligne répétée —
    # deux métriques ignorées, dont un seul vrai doublon de ligne (§4.5).
    assert report["metrics_deduplicated_in_file"] == 2
    assert report["rows_duplicated_in_file"] == 1
    assert [error["line"] for error in report["errors"]] == [6, 7]
    assert "URL invalide" in report["errors"][1]["reason"]


async def test_import_writes_the_canonical_model(
    db: AsyncSession, sandbox: Fixture, stage: Stage
) -> None:
    await stage(POSITIONS_EN)
    await _run_worker()

    keywords = (
        await db.scalars(select(Keyword.keyword).where(Keyword.org_id == sandbox.org_id))
    ).all()
    # « Chaussures   Rouges » et « chaussures rouges » sont **un seul** mot-clé (§3.1).
    assert sorted(keywords) == ["bottines", "chaussures rouges", "escarpins"]

    metric = await db.scalar(
        select(KeywordMetric)
        .join(Keyword, Keyword.id == KeywordMetric.keyword_id)
        .where(Keyword.keyword == "chaussures rouges", KeywordMetric.org_id == sandbox.org_id)
    )
    assert metric is not None
    assert (metric.volume, metric.kd, metric.cpc) == (1300, Decimal("45"), Decimal("0.55"))
    assert metric.intent == "commercial"
    assert metric.serp_features == ["Featured snippet", "People also ask"]
    assert metric.trend == [50, 100, 75, 25, 10, 60, 60, 80, 90, 100, 30, 20]
    assert metric.source == "semrush_positions"
    assert metric.fetched_at == FETCHED_AT
    # La ligne source complète est conservée : position et URL d'origine y sont (§3.2).
    assert metric.raw["Position"] == "3"

    maps = (
        await db.execute(
            select(KeywordPageMap.url_normalized, KeywordPageMap.position, KeywordPageMap.source)
            .where(KeywordPageMap.org_id == sandbox.org_id)
            .order_by(KeywordPageMap.url_normalized)
        )
    ).all()
    #  Trié en Python : l'ordre de `ORDER BY` dépend de la collation de la base.
    assert sorted((url, source) for url, _, source in maps) == [
        # Tracking retiré, casse du chemin préservée, schéma forcé, slash final conservé.
        ("https://exemple.fr/Chaussures", "import_url"),
        ("https://exemple.fr/bottines", "import_url"),
        ("https://exemple.fr/escarpins/", "import_url"),
        ("https://exemple.fr/rouges", "import_url"),
    ]
    assert dict((url, position) for url, position, _ in maps)[
        "https://exemple.fr/rouges"
    ] == Decimal("12")


async def test_page_id_stays_null_until_the_crawl_module(
    db: AsyncSession, sandbox: Fixture, stage: Stage
) -> None:
    """Amendement I4 : la résolution est encapsulée, pas branchée — no-op propre (§3.9)."""
    await stage(POSITIONS_EN)
    await _run_worker()
    pending = await db.scalar(
        select(func.count())
        .select_from(KeywordPageMap)
        .where(KeywordPageMap.org_id == sandbox.org_id, KeywordPageMap.page_id.is_(None))
    )
    assert pending == 4


# --- idempotence --------------------------------------------------------------------------


async def test_forced_replay_touches_no_row(
    db: AsyncSession, sandbox: Fixture, stage: Stage
) -> None:
    """Le rejeu forcé du même fichier laisse la base **strictement** identique (§4.4)."""
    await stage(POSITIONS_EN)
    await _run_worker()
    before = await _snapshot(db, sandbox.org_id)

    job, row = await stage(POSITIONS_EN, sha=_sha("replay"))
    await _run_worker()
    after = await _snapshot(db, sandbox.org_id)
    job, row = await _reload(db, job, row)

    assert after == before, "une ligne a été réécrite : l'idempotence est rompue"
    report = job.report
    assert report is not None
    assert report["rows_inserted"] == 0
    # Toutes les lignes exploitables sont comptées « ignorées » — elles n'ont rien changé.
    assert report["rows_ignored"] == 5
    assert report["rows_error"] == 2


async def test_corrective_reimport_replaces_the_same_observation(
    db: AsyncSession, sandbox: Fixture, stage: Stage
) -> None:
    """Décision C1 : même `(keyword_id, source, fetched_at)` = **même observation**.

    Le scénario réel est « l'adaptateur lisait mal une colonne, on l'a corrigé et on
    ré-importe » : la valeur parsée change, la clé non. Simulé ici en changeant le fichier —
    ce qui compte est que la clé soit identique. Les observations des **autres** sources ne
    bougent pas : ce n'est jamais une fusion inter-sources.
    """
    await stage(POSITIONS_EN)
    await _run_worker()
    keyword_id = await db.scalar(
        select(Keyword.id).where(
            Keyword.org_id == sandbox.org_id, Keyword.keyword == "chaussures rouges"
        )
    )
    other = KeywordMetric(
        org_id=sandbox.org_id,
        keyword_id=keyword_id,
        source="semrush_magic",
        fetched_at=FETCHED_AT,
        volume=999,
        raw={},
    )
    db.add(other)
    await db.commit()

    corrected = POSITIONS_EN.read_text(encoding="utf-8").replace(",1300,45,", ",2600,50,")
    await stage(corrected, sha=_sha("corrected"))
    await _run_worker()

    metrics = (
        await db.execute(
            select(KeywordMetric.source, KeywordMetric.volume, KeywordMetric.kd)
            .where(KeywordMetric.keyword_id == keyword_id)
            .order_by(KeywordMetric.source)
        )
    ).all()
    assert metrics == [
        ("semrush_magic", 999, None),  # l'autre source est intacte
        ("semrush_positions", 2600, Decimal("50")),  # la même observation, corrigée
    ]


# --- maps : une observation se rafraîchit, une décision jamais ----------------------------


async def test_map_is_refreshed_by_a_newer_observation_only(
    db: AsyncSession, sandbox: Fixture, stage: Stage
) -> None:
    """Review I2 : `EXCLUDED.seen_at > seen_at` — jamais de rétrogradation par le passé."""
    await stage(POSITIONS_EN)
    await _run_worker()

    async def position_of(url: str) -> Decimal | None:
        return await db.scalar(
            select(KeywordPageMap.position).where(
                KeywordPageMap.org_id == sandbox.org_id, KeywordPageMap.url_normalized == url
            )
        )

    assert await position_of("https://exemple.fr/bottines") == Decimal("7")

    fresher = POSITIONS_EN.read_text(encoding="utf-8").replace(
        "bottines,Informational,7,7,", "bottines,Informational,5,7,"
    )
    await stage(fresher, sha=_sha("fresher"), fetched_at=LATER)
    await _run_worker()
    assert await position_of("https://exemple.fr/bottines") == Decimal("5")

    staler = POSITIONS_EN.read_text(encoding="utf-8").replace(
        "bottines,Informational,7,7,", "bottines,Informational,30,7,"
    )
    await stage(staler, sha=_sha("staler"), fetched_at=EARLIER)
    await _run_worker()
    assert await position_of("https://exemple.fr/bottines") == Decimal("5"), (
        "une observation plus ancienne a rétrogradé le map"
    )


async def test_manual_map_is_never_touched_by_an_import(
    db: AsyncSession, sandbox: Fixture, stage: Stage
) -> None:
    """Décision n°6 : la planification de l'utilisateur est hors d'atteinte de l'automatique."""
    await stage(POSITIONS_EN)
    await _run_worker()
    keyword_id = await db.scalar(
        select(Keyword.id).where(Keyword.org_id == sandbox.org_id, Keyword.keyword == "bottines")
    )
    manual = KeywordPageMap(
        org_id=sandbox.org_id,
        site_target_id=sandbox.target_id,
        keyword_id=keyword_id,
        url_normalized="https://exemple.fr/bottines",
        source="manual",
        position=None,
        seen_at=EARLIER,
    )
    db.add(manual)
    await db.commit()

    fresher = POSITIONS_EN.read_text(encoding="utf-8").replace(
        "bottines,Informational,7,7,", "bottines,Informational,2,7,"
    )
    await stage(fresher, sha=_sha("over-manual"), fetched_at=LATER)
    await _run_worker()

    manual_id = manual.id  # capturé avant l'expiration (chargement paresseux impossible)
    db.expire_all()
    reloaded = await db.get(KeywordPageMap, manual_id)
    assert reloaded is not None
    assert (reloaded.position, reloaded.seen_at, reloaded.source) == (None, EARLIER, "manual")


# --- invariant d'organisation (dette M3 de T1) ---------------------------------------------


async def test_metrics_always_share_the_org_of_their_keyword(
    db: AsyncSession, sandbox: Fixture, stage: Stage
) -> None:
    """`keyword_metrics_latest` agrège `min(m.org_id)` : l'invariant doit tenir en amont.

    Le pipeline est le seul écrivain de `keyword_metrics`. Il relit l'identifiant du mot-clé
    **filtré par `org_id`**, donc il ne peut pas rattacher une observation au mot-clé d'une
    autre organisation — même texte, même locale, deux lignes distinctes.
    """
    await stage(POSITIONS_EN)
    await _run_worker()

    breaches = await db.scalar(
        text(
            "SELECT count(*) FROM keyword_metrics m JOIN keywords k ON k.id = m.keyword_id "
            "WHERE k.org_id <> m.org_id"
        )
    )
    assert breaches == 0

    # Le même texte importé par une autre organisation crée son propre mot-clé.
    twins = await db.scalar(
        select(func.count())
        .select_from(Keyword)
        .where(Keyword.keyword == "chaussures rouges", Keyword.locale == "fr-FR")
    )
    assert twins is not None and twins >= 1
    mine = await db.scalar(
        select(func.count())
        .select_from(Keyword)
        .where(
            Keyword.org_id == sandbox.org_id,
            Keyword.keyword == "chaussures rouges",
        )
    )
    assert mine == 1


# --- statut de l'import : le worker reste neutre (I3) --------------------------------------


async def test_import_status_and_report_follow_the_job(
    db: AsyncSession, sandbox: Fixture, stage: Stage
) -> None:
    job, row = await stage(POSITIONS_EN)
    assert row.status == "queued"
    await _run_worker()
    job, row = await _reload(db, job, row)
    assert (job.status, row.status) == ("done", "done")
    assert row.report is not None and row.report["rows_read"] == 7
    assert row.error is None


async def test_failed_job_marks_the_import_failed(
    db: AsyncSession, sandbox: Fixture, stage: Stage, import_dir: Path
) -> None:
    """Callback `on_fail` du registre : plus aucun import `queued` à perpétuité (§3.11)."""
    job, row = await stage(POSITIONS_EN)
    # Le fichier disparaît entre l'enfilage et l'exécution : le handler lève, le worker met
    # le job en `failed`, et c'est le callback du module qui recopie l'échec.
    _remove(import_dir / sandbox.org_id / f"{row.file_sha256}.csv")
    await _run_worker()
    job, row = await _reload(db, job, row)
    assert job.status == "failed"
    assert (row.status, row.error) == ("failed", job.error)
    assert row.error is not None


async def test_unknown_adapter_fails_loudly(
    db: AsyncSession, sandbox: Fixture, stage: Stage
) -> None:
    job, row = await stage(POSITIONS_EN, source="generic_csv", options=None)
    await _run_worker()
    job, row = await _reload(db, job, row)
    # `generic_csv` sans mapping : erreur d'appel, pas 7 lignes en erreur silencieuses.
    assert job.status == "failed"
    assert row.status == "failed"
    assert "keyword" in (job.error or "")


# --- lignes en erreur ----------------------------------------------------------------------


async def test_errors_csv_is_written_next_to_the_source_file(
    db: AsyncSession, sandbox: Fixture, stage: Stage, import_dir: Path
) -> None:
    job, row = await stage(POSITIONS_EN)
    sha = row.file_sha256
    await _run_worker()
    job, row = await _reload(db, job, row)
    path = import_dir / sandbox.org_id / f"{sha}.errors.csv"
    assert path.exists()
    lines = path.read_text(encoding="utf-8").splitlines()
    assert lines[0] == "line,reason"
    assert len(lines) == 3
    assert job.report is not None and job.report["errors_csv"] == path.name
    assert job.report["errors_complete"] is True


async def test_error_lines_are_the_physical_lines_of_the_file(
    db: AsyncSession, sandbox: Fixture, stage: Stage, import_dir: Path
) -> None:
    """Le câblage complet de `_numbered` : recalage adaptateur → numéros physiques (§ `RowError`).

    Les deux producteurs d'erreurs ne comptent pas dans le même repère — le lecteur CSV rend un
    numéro **physique**, l'adaptateur un **rang de données**, aveugle aux lignes que le lecteur
    a écartées plus haut. Le fichier ci-dessous déchire une ligne **au milieu** : sans recalage,
    les deux mots-clés vides seraient annoncés lignes 3 et 5 — soit une **collision** avec la
    ligne déchirée (3) et un décalage sur l'autre. L'utilisateur qui ouvre son tableur pour
    corriger regarderait deux fois la mauvaise ligne. Les unités sont couvertes ailleurs ; ce
    qui est vérifié ici est qu'elles sont **branchées**, de bout en bout, jusqu'à `errors.csv`.
    """
    csv_text = (
        "Keyword,Position,Search Volume,URL\n"  # 1 — en-tête
        "alpha,1,100,https://exemple.fr/a\n"  # 2 — valide
        "déchirée,2\n"  # 3 — écartée par le lecteur (colonnes manquantes)
        ",3,300,https://exemple.fr/b\n"  # 4 — mot-clé vide (rang de données 3)
        "beta,4,400,https://exemple.fr/c\n"  # 5 — valide
        ",5,500,https://exemple.fr/d\n"  # 6 — mot-clé vide (rang de données 5)
    )
    job, row = await stage(csv_text, sha=_sha("numbered"))
    await _run_worker()
    job, row = await _reload(db, job, row)

    assert job.report is not None
    assert job.report["rows_error"] == 3
    #  3 = la ligne déchirée (repère physique du lecteur), 4 et 6 = les mots-clés vides,
    #  ramenés du repère de l'adaptateur (3 et 5) au repère du fichier.
    assert [error["line"] for error in job.report["errors"]] == [3, 4, 6]
    assert "mot-clé vide" in job.report["errors"][1]["reason"]
    #  Le fichier téléchargeable porte les mêmes numéros : c'est lui que l'utilisateur ouvre.
    errors_csv = import_dir / sandbox.org_id / f"{row.file_sha256}.errors.csv"
    assert [line.split(",", 1)[0] for line in errors_csv.read_text("utf-8").splitlines()[1:]] == [
        "3",
        "4",
        "6",
    ]


# --- export réel de bout en bout -----------------------------------------------------------


async def test_real_magic_export_imports_end_to_end(
    db: AsyncSession, sandbox: Fixture, stage: Stage
) -> None:
    """L'export Keyword Magic réel, pied de page compris (§5, règle de la fixture réelle)."""
    job, row = await stage(MAGIC_REAL, source="semrush_magic")
    await _run_worker()
    job, row = await _reload(db, job, row)

    assert job.status == "done"
    report = job.report
    assert report is not None
    assert report["rows_read"] == 503  # 500 lignes de données + 3 lignes de pied de page
    assert report["rows_error"] == 3
    assert [error["line"] for error in report["errors"]] == [502, 503, 504]
    assert report["keywords_created"] == report["metrics_written"]
    # Aucune URL dans cet export : que des orphelins, aucun map (§0).
    assert report["maps_written"] == 0
    #  Keyword Magic n'écrit pas son pays : le cloisonnement (§4.3) est un **no-op strict**
    #  sur cette source, et un import Semrush est celui d'avant, à la ligne près.
    assert report["rows_off_target"] == 0
    assert report["rows_country_unresolved"] == 0
    assert report["ignored_by_country"] == {}
    assert report["unresolved_by_country"] == {}

    orphans = await db.scalar(
        select(func.count())
        .select_from(KeywordListItem)
        .where(KeywordListItem.org_id == sandbox.org_id)
    )
    assert orphans == report["list_items_added"] > 400


# --- cloisonnement par pays (§4.3, §4.5) ----------------------------------------------------


def _fixture_rows() -> list[dict[str, str]]:
    """La fixture multi-pays relue par la bibliothèque standard, jamais par l'adaptateur.

    Les décomptes attendus plus bas ne sont pas des constantes recopiées de la spec (qui
    parlait d'un autre fichier) : ils sont **dérivés du fichier lui-même**. Une fixture
    régénérée fait donc bouger les attentes avec elle, au lieu de rendre les tests faux.
    """
    with MULTI_COUNTRY.open(newline="", encoding="utf-8-sig") as handle:
        return list(csv.DictReader(handle))


def _rows_per_market() -> dict[str, int]:
    counted: dict[str, int] = {}
    for row in _fixture_rows():
        counted[row["Location"]] = counted.get(row["Location"], 0) + 1
    return counted


def _market_keywords(market: str) -> set[str]:
    """Les mots-clés que le fichier écrit sous un libellé de marché, forme canonique comprise."""
    return {
        " ".join(row["Keyword"].split()).lower()
        for row in _fixture_rows()
        if row["Location"] == market
    }


def _stolen_carriers(market: str) -> set[str]:
    """Les mots-clés de `market` qu'une ligne d'un **autre** marché précède dans le fichier.

    Ce sont exactement ceux qu'un filtrage mal placé — après `seen` — amputerait de leur
    métrique : la ligne étrangère prendrait le rôle de porteuse, et la bonne ligne, arrivée
    plus tard, serait comptée « doublon intra-fichier ». Sans eux, le test du piège n'aurait
    rien à prouver ; il exige donc qu'il y en ait.
    """
    first_seen: dict[str, str] = {}
    stolen: set[str] = set()
    for row in _fixture_rows():
        keyword = " ".join(row["Keyword"].split()).lower()
        if row["Location"] == market and first_seen.get(keyword, market) != market:
            stolen.add(keyword)
        first_seen.setdefault(keyword, row["Location"])
    return stolen


async def _import_market(
    db: AsyncSession, stage: Stage, target_id: uuid.UUID, locale: str, **overrides: Any
) -> dict[str, Any]:
    """Importe la fixture multi-pays sur une cible, par le vrai worker, et rend son rapport."""
    job, row = await stage(
        MULTI_COUNTRY,
        source="mangools",
        target_id=target_id,
        locale=locale,
        list_name=f"KWFinder — {locale}",
        sha=_sha(f"multi-country-{locale}"),
        **overrides,
    )
    await _run_worker()
    job, _ = await _reload(db, job, row)
    assert job.status == "done", job.error
    assert job.report is not None
    return dict(job.report)


async def _keywords_of(db: AsyncSession, org_id: str, locale: str) -> set[str]:
    return set(
        (
            await db.scalars(
                select(Keyword.keyword).where(Keyword.org_id == org_id, Keyword.locale == locale)
            )
        ).all()
    )


async def test_the_multi_country_export_keeps_only_the_market_of_the_target(
    db: AsyncSession, sandbox: Fixture, stage: Stage, make_target: MakeTarget
) -> None:
    """Un export KWFinder de onze marchés importé sur `en-US` : 74 lignes, et 74 seulement (§4.3).

    Le reste n'est **pas** en erreur — ni `rows_error`, ni `errors.csv` : ces lignes existent,
    elles sont lisibles, elles n'appartiennent simplement pas à ce marché. Elles sont ventilées
    en deux motifs distincts, « hors cible » et « pays non reconnu », par libellé brut.
    """
    market = await make_target("US", "en")
    per_market = _rows_per_market()
    report = await _import_market(db, stage, market, "en-US")

    assert report["rows_read"] == sum(per_market.values()) == 229
    assert report["rows_error"] == 0
    assert report["errors"] == [] and report["errors_csv"] is None
    assert report["rows_inserted"] == per_market["United States"] == 74
    assert report["keywords_created"] == report["metrics_written"] == 74

    #  « New York » est une localité : `countries.resolve` refuse de la deviner, la ligne est
    #  ignorée et son libellé remonte tel quel — c'est ce qui rend visible un trou de table.
    assert report["rows_country_unresolved"] == per_market["New York"] == 3
    assert report["unresolved_by_country"] == {"New York": 3}
    off_target = {
        raw: count for raw, count in per_market.items() if raw not in ("United States", "New York")
    }
    assert report["rows_off_target"] == sum(off_target.values()) == 152
    assert report["ignored_by_country"] == off_target
    #  `rows_ignored` reste le TOTAL : les deux motifs en sont des sous-ensembles (§4.5).
    assert report["rows_ignored"] >= report["rows_off_target"] + report["rows_country_unresolved"]

    kept = await _keywords_of(db, sandbox.org_id, "en-US")
    assert len(kept) == 74
    assert kept == _market_keywords("United States")


async def test_the_same_file_on_a_second_target_keeps_its_own_market(
    db: AsyncSession, sandbox: Fixture, stage: Stage, make_target: MakeTarget
) -> None:
    """Le geste du §4.4 : N imports du même fichier, un par cible, chacun cloisonné chez lui.

    Le second import ne touche pas au premier — deux marchés font deux jeux de mots-clés
    (`UNIQUE (org_id, locale, keyword)`), et aucune ligne d'un pays n'atterrit dans la locale
    d'un autre. C'est la promesse entière du cloisonnement, vérifiée dans les deux sens.
    """
    per_market = _rows_per_market()
    us = await make_target("US", "en")
    gb = await make_target("GB", "en")

    first = await _import_market(db, stage, us, "en-US")
    second = await _import_market(db, stage, gb, "en-GB")

    assert first["rows_inserted"] == 74
    assert second["rows_inserted"] == per_market["United Kingdom"] == 55
    assert second["rows_country_unresolved"] == 3
    off_target = {
        raw: count for raw, count in per_market.items() if raw not in ("United Kingdom", "New York")
    }
    assert second["rows_off_target"] == sum(off_target.values()) == 171
    assert second["ignored_by_country"] == off_target

    american = await _keywords_of(db, sandbox.org_id, "en-US")
    british = await _keywords_of(db, sandbox.org_id, "en-GB")
    #  Ce que chaque locale contient est exactement ce que le fichier écrit sous ce libellé :
    #  aucune ligne d'un marché n'a filé dans la locale d'un autre, dans un sens ni dans l'autre.
    assert american == _market_keywords("United States")
    assert british == _market_keywords("United Kingdom")
    #  Les deux témoins qui portent la démonstration : « jewellery casting » est écrit par cinq
    #  marchés du fichier (GB, AU, SG, HK, NZ) et n'entre que dans la locale britannique ;
    #  « manufacturer jewelry » en compte sept, dont la localité « New York », et n'entre que
    #  dans l'américaine. Ni l'un ni l'autre ne traverse.
    assert "jewellery casting" in british and "jewellery casting" not in american
    assert "manufacturer jewelry" in american and "manufacturer jewelry" not in british
    total = await db.scalar(
        select(func.count()).select_from(Keyword).where(Keyword.org_id == sandbox.org_id)
    )
    assert total == 74 + 55


async def test_a_filtered_row_never_steals_the_metric_carrier_role(
    db: AsyncSession, sandbox: Fixture, stage: Stage, make_target: MakeTarget
) -> None:
    """Le piège du cloisonnement : filtrer trop tard ampute des mots-clés, en silence.

    `seen` désigne la première ligne d'un mot-clé comme **porteuse de sa métrique**. Si une
    ligne écartée y entrait quand même, la ligne du bon marché rencontrée plus loin serait
    comptée « doublon intra-fichier » et sa métrique ne serait jamais écrite : le mot-clé
    existerait, sans volume ni difficulté, et le rapport ne dirait rien.

    La fixture porte le cas naturellement : sept mots-clés britanniques sont précédés d'une
    ligne australienne. `costume jewellery manufacturer` les résume — AU 22 de difficulté à la
    ligne 198, GB 24 à la 209.
    """
    stolen = _stolen_carriers("United Kingdom")
    assert len(stolen) == 7, "la fixture ne porte plus le piège : le test ne prouve plus rien"
    gb = await make_target("GB", "en")
    report = await _import_market(db, stage, gb, "en-GB")

    #  Une métrique par mot-clé gardé, et **aucun** doublon intra-fichier : les 55 lignes
    #  britanniques portent 55 mots-clés distincts. Avec le piège, il en manquerait sept.
    assert report["metrics_written"] == 55
    assert report["metrics_deduplicated_in_file"] == 0
    assert report["rows_duplicated_in_file"] == 0

    rows = (
        await db.execute(
            select(Keyword.keyword, KeywordMetric.kd)
            .join(KeywordMetric, KeywordMetric.keyword_id == Keyword.id)
            .where(Keyword.org_id == sandbox.org_id, Keyword.keyword.in_(sorted(stolen)))
        )
    ).all()
    assert {keyword for keyword, _ in rows} == stolen
    #  Et c'est bien la ligne britannique qui a écrit, pas l'australienne qui la précède.
    assert dict(rows)["costume jewellery manufacturer"] == Decimal("24")


async def test_a_job_enqueued_before_the_partitioning_falls_back_on_the_locale(
    db: AsyncSession, sandbox: Fixture, stage: Stage, make_target: MakeTarget
) -> None:
    """Un payload **sans** clé `country` : le pays est lu dans la locale, et rien ne change.

    C'est le seul état transitoire du déploiement — les jobs déjà en file quand l'api part.
    `locale_of` écrit toujours `langue-PAYS`, la seconde moitié est donc exactement ce que
    `create_import` aurait mis dans `country`.
    """
    us = await make_target("US", "en")
    report = await _import_market(db, stage, us, "en-US", country=None)

    job = await db.scalar(select(Job).where(Job.type == pipeline.JOB_TYPE))
    assert job is not None and "country" not in job.payload
    assert pipeline.parse_payload(job.payload).country == "US"
    assert report["rows_inserted"] == 74
    assert report["rows_off_target"] == 152
    assert report["rows_country_unresolved"] == 3


async def test_a_generic_csv_is_partitioned_by_the_column_the_user_designates(
    db: AsyncSession, sandbox: Fixture, stage: Stage, make_target: MakeTarget
) -> None:
    """`generic_csv` n'a pas de colonne pays : l'utilisateur la désigne (Q4), hors mapping.

    Et la casse de l'en-tête n'a pas à correspondre : `country_header` rend l'en-tête
    **déclaré**, que le pipeline confronte aux en-têtes réels avec la même tolérance que le
    mapping (`normalize_header`). Sans cette résolution, `raw.get("marché")` rendrait `None`
    sur chaque ligne — c'est tout le fichier qui partirait en « pays non reconnu ».
    """
    gb = await make_target("GB", "en")
    raw = (
        "Requete,Vol,Marché\n"
        "shoes,100,United Kingdom\n"
        "boots,200,uk\n"
        "sneakers,300,United States\n"
        "sandals,400,New York\n"
        "clogs,500,\n"
    )
    job, row = await stage(
        raw,
        source="generic_csv",
        target_id=gb,
        locale="en-GB",
        list_name="Générique",
        #  En-tête réel : « Marché ». Déclaré ici en minuscules, comme le mapping l'autorise.
        options={"columns": {"keyword": "Requete", "volume": "Vol"}, "country_column": "marché"},
    )
    await _run_worker()
    job, _ = await _reload(db, job, row)
    assert job.status == "done", job.error
    report = job.report
    assert report is not None

    #  « uk » passe par la table d'alias de `app.keywords.countries` : deux libellés, un pays.
    assert report["rows_read"] == 5
    assert report["rows_inserted"] == 2
    assert report["rows_off_target"] == 1
    assert report["ignored_by_country"] == {"United States": 1}
    #  Deux lignes « non reconnues » : une localité, et une cellule vide. La cellule vide compte
    #  comme ligne mais n'entre dans aucun libellé — il n'y a rien à nommer à l'écran, et le
    #  recensement de l'aperçu l'écarte déjà pour la même raison.
    assert report["rows_country_unresolved"] == 2
    assert report["unresolved_by_country"] == {"New York": 1}
    assert await _keywords_of(db, sandbox.org_id, "en-GB") == {"shoes", "boots"}


async def test_a_declared_country_column_absent_from_the_file_filters_nothing(
    db: AsyncSession, sandbox: Fixture, stage: Stage, make_target: MakeTarget
) -> None:
    """Colonne pays déclarée mais absente du fichier : l'import passe entier, comme avant.

    Même arbitrage qu'au recensement de l'aperçu, qui rend `[]` plutôt que de refuser le
    fichier. Écarter toutes les lignes serait pire que ne pas cloisonner : l'utilisateur
    n'aurait ni ses données, ni d'explication.
    """
    gb = await make_target("GB", "en")
    job, row = await stage(
        "Requete,Vol\nshoes,100\nboots,200\n",
        source="generic_csv",
        target_id=gb,
        locale="en-GB",
        list_name="Sans pays",
        options={"columns": {"keyword": "Requete", "volume": "Vol"}, "country_column": "Marché"},
    )
    await _run_worker()
    job, _ = await _reload(db, job, row)
    assert job.status == "done", job.error
    assert job.report is not None
    assert job.report["rows_inserted"] == 2
    assert job.report["rows_off_target"] == 0
    assert job.report["rows_country_unresolved"] == 0


async def test_the_country_breakdown_is_recomputed_on_a_resumed_run(
    db: AsyncSession, sandbox: Fixture, stage: Stage, make_target: MakeTarget
) -> None:
    """La ventilation vit dans `ScanCounters`, donc **jamais** dans `progress` (Q6).

    Le job est mis dans l'état d'une reprise : son curseur dit que le premier lot est passé.
    Le fichier compte moins que `BATCH_SIZE`, donc **aucune écriture** n'a lieu — et pourtant
    le lot est relu, ligne à ligne, et la ventilation sort complète. La mettre dans
    `WriteCounters`, restauré depuis `progress`, la ferait repartir de zéro : le rapport d'un
    import repris ne serait plus celui d'un import d'une traite.
    """
    us = await make_target("US", "en")
    job, row = await stage(
        MULTI_COUNTRY,
        source="mangools",
        target_id=us,
        locale="en-US",
        list_name="KWFinder — en-US",
        sha=_sha("multi-country-resume"),
    )
    job.progress = {"batch": 1, "rows_read": 229, "counters": {}}
    await db.commit()
    await _run_worker()
    job, _ = await _reload(db, job, row)
    assert job.status == "done", job.error
    assert job.report is not None

    #  Le lot a bien été sauté : rien n'a été écrit.
    assert job.report["rows_inserted"] == 0
    assert await _keywords_of(db, sandbox.org_id, "en-US") == set()
    #  Et pourtant tout a été relu et reventilé.
    assert job.report["rows_read"] == 229
    assert job.report["rows_off_target"] == 152
    assert job.report["rows_country_unresolved"] == 3
    assert sum(job.report["ignored_by_country"].values()) == 152
    assert job.report["rows_ignored"] >= 152 + 3


def test_the_country_breakdown_never_travels_in_progress() -> None:
    """La décision Q6, figée par les dataclasses elles-mêmes.

    `WriteCounters` est le seul des deux à voyager dans `progress` et à être restauré à la
    reprise (`from_dict`). Un compteur de cloisonnement qui y atterrirait serait restauré au
    lieu d'être recalculé, et le test ci-dessus ne le verrait pas : il ne teste qu'un fichier
    à un seul lot.
    """
    partitioning = {
        "rows_off_target",
        "rows_country_unresolved",
        "ignored_by_country",
        "unresolved_by_country",
    }
    scan = {f.name for f in fields(pipeline.ScanCounters)}
    written = {f.name for f in fields(pipeline.WriteCounters)}
    assert partitioning <= scan
    assert partitioning & written == set()
    assert set(pipeline.WriteCounters().as_dict()) == written