"""Contrat des adaptateurs de source (spec Imports §5) et coercions canoniques.

Un adaptateur transforme les lignes d'un export en `CanonicalKeywordRow`. Il ne connaît ni
la base, ni le job, ni l'organisation : il ne fait que lire et traduire. Tout le reste — la
locale héritée de la cible, `fetched_at`, les upserts — appartient au pipeline (§4).

**Détection par signature de colonnes, jamais par nom de fichier** (§5). Chaque adaptateur
déclare une `ColumnSignature` **par langue d'export** : Semrush localise ses en-têtes, et
les deux exports Semrush ne se distinguent que par la présence des colonnes URL/Position.
Une signature porte donc aussi les en-têtes **disqualifiants** (`absent`) — c'est ce qui
sépare « Organic Positions » de « Keyword Magic » sans jamais regarder le nom du fichier.

Les lignes inexploitables ne lèvent pas : elles sont **rendues** en `RowError` (§4.2, « on
n'échoue jamais en silence »). Un adaptateur qui lèverait ferait échouer tout l'import pour
une cellule.
"""

import re
import unicodedata
from bisect import bisect_right
from collections.abc import Iterator, Mapping, Sequence
from dataclasses import dataclass, field
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any

# Champs canoniques qu'une signature peut déclarer (data §7). `keyword` est toujours exigé ;
# `url`/`position` n'existent que sur les sources qui observent un classement.
CANONICAL_FIELDS = (
    "keyword",
    "volume",
    "kd",
    "cpc",
    "intent",
    "serp_features",
    "trend",
    "url",
    "position",
)

# Le `trend` canonique est un `SMALLINT[]` (§3.2). Semrush émet douze flottants relatifs
# dans [0, 1] (part du maximum des douze mois) : ils sont stockés en **entiers 0–100**,
# c'est-à-dire ce même ratio en pourcentage — la précision d'une sparkline, sans perte de
# sens. La valeur brute exacte reste dans `raw`, qui n'est jamais réécrit à la baisse.
TREND_SCALE = 100
TREND_LENGTH = 12


@dataclass(frozen=True, slots=True)
class CanonicalKeywordRow:
    """Une ligne d'export traduite dans le modèle canonique (§4.2).

    `line` est le numéro de ligne dans le fichier source : il ne sert qu'au rapport et à
    `errors.csv`. `raw` porte la ligne source complète — position et URL d'origine y sont
    rangées même quand elles alimentent aussi le map (§3.2).
    """

    line: int
    keyword: str
    volume: int | None = None
    kd: Decimal | None = None
    cpc: Decimal | None = None
    intent: str | None = None
    serp_features: list[str] | None = None
    trend: list[int] | None = None
    url: str | None = None
    position: Decimal | None = None
    raw: dict[str, Any] = field(default_factory=dict)


@dataclass(frozen=True, slots=True)
class RowError:
    """Une ligne collectée en erreur : jamais un échec de job (§4.2).

    `line` est **toujours** rendu au numéro de ligne physique du fichier avant d'atterrir dans
    le rapport ou dans `errors.csv` — c'est ce numéro que l'utilisateur ouvre dans son
    tableur. Mais les deux producteurs d'erreurs ne comptent pas dans le même repère :

    - le lecteur CSV (`store_rejects`) rend directement le numéro **physique** ;
    - un adaptateur, lui, ne voit que les lignes que le lecteur a bien voulu lui donner : il
      compte donc en **rang de données**, et son numéro dérive dès qu'une ligne a été écartée
      plus haut dans le fichier.

    `physical` distingue les deux, et `to_physical_lines` (appelé une fois, à la fin du scan,
    quand la liste des lignes écartées est enfin connue) ramène les seconds dans le repère des
    premiers. Sans cela, une ligne déchirée en **milieu** de fichier ne décale pas seulement
    les numéros suivants : elle fabrique une **collision** — la ligne écartée et une ligne
    valide portent le même numéro, et l'utilisateur qui corrige son fichier regarde la
    mauvaise.
    """

    line: int
    reason: str
    physical: bool = False


def to_physical_lines(errors: Sequence[RowError], rejected: Sequence[int]) -> list[RowError]:
    """Ramène les erreurs d'adaptateur dans le repère des numéros de ligne physiques.

    Le rang de données `d` désigne la `d`-ième ligne du fichier après l'en-tête **qui n'a pas
    été écartée** par le lecteur. On cherche donc le `d`-ième entier ≥ 2 absent de `rejected`.
    Le point fixe `p = 1 + d + |{r ∈ rejected : r ≤ p}|` converge en quelques tours (chaque
    itération ne peut qu'ajouter des lignes écartées déjà comptées).
    """
    if not rejected:
        return list(errors)
    ordered = sorted(rejected)
    shifted: list[RowError] = []
    for error in errors:
        if error.physical:
            shifted.append(error)
            continue
        rank = error.line - 1  # `line` = 1 (en-tête) + rang de données
        physical = rank + 1
        while (candidate := rank + 1 + bisect_right(ordered, physical)) != physical:
            physical = candidate
        shifted.append(RowError(physical, error.reason, physical=True))
    return shifted


ParsedRow = CanonicalKeywordRow | RowError


def normalize_header(raw: str) -> str:
    """Forme de comparaison d'un en-tête : sans BOM, sans casse, espaces réduits.

    Les tirets et espaces sont unifiés (« Mot-clé » ≡ « Mot clé » — Semrush a écrit les deux
    selon les versions) et les accents conservés : ils portent du sens et les exports FR sont
    cohérents avec eux-mêmes.
    """
    cleaned = unicodedata.normalize("NFC", raw.replace("", ""))
    return " ".join(cleaned.replace("-", " ").replace("_", " ").split()).casefold()


class AmbiguousHeaderError(ValueError):
    """Deux colonnes du fichier portent le même en-tête une fois normalisé.

    Erreur d'appel, pas une ligne en erreur : elle fait échouer le job (et l'import) plutôt que
    de laisser la dernière colonne écraser silencieusement la première dans l'index. « Mot-clé »
    et « Mot clé » dans le même fichier ne sont pas un cas théorique — un export recollé à la
    main en produit —, et le silence ferait importer la mauvaise colonne sans que rien ne le
    dise. T3 la transformera en 422 à l'aperçu.
    """


def header_index(headers: Sequence[str]) -> dict[str, str]:
    """En-tête normalisé → en-tête réel du fichier. Lève si deux en-têtes se confondent.

    Les doublons **exacts** n'arrivent pas ici : DuckDB les a déjà suffixés (`Keyword_1`), donc
    ils ne se confondent plus après normalisation. Ce qui reste, ce sont les homonymes de
    ponctuation ou de casse, que `normalize_header` rapproche volontairement.
    """
    index: dict[str, str] = {}
    for header in headers:
        key = normalize_header(header)
        if key in index:
            raise AmbiguousHeaderError(
                f"deux colonnes portent le même en-tête une fois normalisé "
                f"({index[key]!r} et {header!r}) : renommez-en une avant l'import"
            )
        index[key] = header
    return index


@dataclass(frozen=True, slots=True)
class ColumnSignature:
    """Les en-têtes d'un export dans **une** langue (§5).

    `columns` associe un champ canonique aux en-têtes acceptés dans cette langue (plusieurs
    orthographes coexistent d'une version de l'export à l'autre) ; `patterns` couvre les
    en-têtes à suffixe variable — `CPC (USD)` devient `CPC (EUR)` selon la devise du compte,
    aucune liste close ne peut les énumérer. `absent` disqualifie la signature : c'est le
    seul mécanisme qui sépare Positions de Magic.
    """

    language: str
    columns: Mapping[str, tuple[str, ...]]
    required: frozenset[str]
    absent: tuple[str, ...] = ()
    patterns: Mapping[str, str] = field(default_factory=dict)

    def match(self, headers: Sequence[str]) -> dict[str, str] | None:
        """Rend le mapping champ canonique → en-tête **réel** du fichier, ou `None`.

        Le mapping rendu porte les en-têtes tels qu'ils apparaissent dans le fichier (casse
        et accents d'origine) : c'est par eux que l'adaptateur lit les cellules.
        """
        index = header_index(headers)
        if any(forbidden in index for forbidden in self.absent):
            return None
        found: dict[str, str] = {}
        for canonical, accepted in self.columns.items():
            for header in accepted:
                if header in index:
                    found[canonical] = index[header]
                    break
        for canonical, pattern in self.patterns.items():
            if canonical in found:
                continue
            compiled = re.compile(pattern)
            for normalized, original in index.items():
                if compiled.fullmatch(normalized):
                    found[canonical] = original
                    break
        if not self.required <= found.keys():
            return None
        return found


@dataclass(frozen=True, slots=True)
class Detection:
    """Une candidature d'adaptateur pour un en-tête donné (étape 2 du wizard, §1.4)."""

    source: str
    language: str
    #  Part des champs déclarés par la signature effectivement présents : départage deux
    #  adaptateurs qui matchent tous les deux (jamais un nom de fichier, §5).
    confidence: float
    columns: dict[str, str]


class Adapter:
    """Interface canonique (§5). `parse_csv` **rend** les erreurs, ne les lève pas."""

    name: str
    signatures: tuple[ColumnSignature, ...]
    #  En-tête **réel** qui porte le pays de la ligne (« Location » chez KWFinder), ou `None`
    #  quand la source n'écrit pas son pays — c'est alors la cible qui fait autorité, comme
    #  avant le cloisonnement. Le défaut est obligatoire : sans lui, tout adaptateur qui ne le
    #  déclare pas casserait en `AttributeError` au premier aperçu.
    #
    #  **Délibérément hors des `signatures`** : la confiance de détection est le ratio
    #  `len(columns) / len(signature.columns)`, et ajouter une colonne au dénominateur
    #  déplacerait la confiance de *tous* les fichiers, y compris ceux qui ne la portent pas.
    #  `sources.best()` rendant `None` dès que les deux meilleures candidatures sont ex æquo,
    #  des fichiers aujourd'hui reconnus basculeraient en silence vers `generic_csv`.
    country_column: str | None = None

    def country_header(self, options: Mapping[str, Any] | None = None) -> str | None:
        """L'en-tête qui porte le pays pour **cet** appel, `options` comprises.

        Seul `generic_csv` a besoin d'`options` (l'utilisateur désigne la colonne lui-même) ;
        les adaptateurs à signature connaissent la leur, et ignorent ce qu'on leur passe.
        """
        return self.country_column

    def detect(self, headers: Sequence[str]) -> Detection | None:
        best: Detection | None = None
        for signature in self.signatures:
            columns = signature.match(headers)
            if columns is None:
                continue
            confidence = len(columns) / max(len(signature.columns), 1)
            if best is None or confidence > best.confidence:
                best = Detection(self.name, signature.language, confidence, columns)
        return best

    def parse_csv(
        self, path: Path, *, options: Mapping[str, Any] | None = None
    ) -> Iterator[ParsedRow]:
        raise NotImplementedError


# --- coercions ------------------------------------------------------------------------
#
# Les exports sont du texte : tout est lu en VARCHAR et converti ici, jamais par DuckDB. Une
# cellule illisible ne doit pas faire tomber la ligne entière quand elle n'est pas le
# mot-clé — un volume vide (fréquent dans l'export réel) rend simplement `None`.

# Espaces de séparation des milliers, insécables compris (Excel, exports FR).
_SPACES = str.maketrans({" ": "", " ": "", " ": "", " ": "", "'": ""})


def to_decimal(raw: str | None) -> Decimal | None:
    """Nombre d'une cellule d'export, ou `None` si la cellule est vide ou illisible.

    Gère les deux conventions décimales : `1234.5` (export « CSV ») et `1234,5` (export
    « CSV Semicolon » d'un compte européen). La virgule n'est lue comme séparateur décimal
    que s'il n'y a pas de point — sans quoi `1,234.5` deviendrait faux.
    """
    if raw is None:
        return None
    text = raw.strip().translate(_SPACES)
    if not text or text in {"-", "—", "n/a", "N/A"}:
        return None
    if "," in text and "." not in text:
        text = text.replace(",", ".")
    else:
        text = text.replace(",", "")
    text = text.removesuffix("%")
    try:
        return Decimal(text)
    except InvalidOperation:
        return None


def to_int(raw: str | None) -> int | None:
    value = to_decimal(raw)
    return None if value is None else int(value)


def to_trend(raw: str | None) -> list[int] | None:
    """Les douze valeurs de tendance en entiers 0–100 (voir `TREND_SCALE`).

    Semrush écrit `"0.17,0.33,…"` — champ **quoté**, donc un export à virgule reste lisible.
    Une tendance incomplète ou illisible rend `None` : c'est un ornement d'affichage, jamais
    une raison de rejeter la ligne.
    """
    if raw is None or not raw.strip():
        return None
    values: list[int] = []
    for part in raw.split(","):
        value = to_decimal(part)
        if value is None:
            return None
        values.append(int(round(value * TREND_SCALE)))
    return values if len(values) == TREND_LENGTH else None


def to_serp_features(raw: str | None) -> list[str] | None:
    if raw is None or not raw.strip():
        return None
    features = [part.strip() for part in raw.split(",") if part.strip()]
    return features or None


def to_intent(raw: str | None) -> str | None:
    """Intention(s) de recherche, en minuscules et séparées par des virgules.

    Une ligne peut en porter plusieurs (`informational, commercial`) : la colonne canonique
    est du texte (§3.2), la forme est donc figée ici pour rester filtrable.
    """
    if raw is None or not raw.strip():
        return None
    parts = [part.strip().casefold() for part in raw.split(",") if part.strip()]
    return ",".join(parts) or None