"""Le réservoir de mots-clés par cible, et les réglages d'organisation (spec Imports §6).

## Pourquoi cette requête est écrite en SQL et pas en ORM

Elle assemble une fonction table (`keyword_metrics_latest`), deux latéraux d'agrégat et six
pré-filtres corrélés, et **c'est son plan d'exécution qui est le livrable** : le benchmark T2 a
montré qu'une page de 50 lignes filtrée sur une valeur consolidée pouvait coûter 480 000
buffers si le filtre n'était pas préparé. Un `select()` ORM produirait le même SQL en moins
lisible ; ici, la forme du texte est la forme du plan.

Toutes les valeurs passent par des paramètres liés. Les rares fragments assemblés en Python
(mode de vue, sens des comparaisons) sont **des constantes du module**, jamais des chaînes
venues de la requête : le nom de la vue est confronté au registre d'adaptateurs avant d'entrer
dans la requête, et le statut d'attache est validé par un `Literal` Pydantic.

## Filtrer sur une métrique : « au moins une source satisfait » (T2, revu le 2026-09-01)

Le filtre `EXISTS (… keyword_metrics …)` porte sur les observations **brutes**, sur des
colonnes indexables — c'est ce qui évite d'évaluer la consolidation de tous les candidats avant
d'en garder une page (le benchmark T2 : 480 000 buffers).

Il était doublé, jusqu'à la spec 2026-09-01, d'un second test sur la valeur *élue* par la
consolidation, qui faisait foi. Il n'y a plus de valeur élue (§3.3) : le EXISTS **est** le
filtre, et la règle qu'il énonce est « au moins une source satisfait ». Un mot-clé dont une
source annonce 5 000 et l'autre 10 apparaît donc dans `volume_min=1000` **et** dans
`volume_max=100` — c'est voulu, la ligne affiche les deux valeurs côte à côte et l'écran ne
ment pas. C'est la régression du §1 : sous l'ancienne règle, une seule des deux sources
décidait, et le mot-clé disparaissait du filtre qu'une de ses sources satisfaisait.

Conséquence assumée : le EXISTS voit **tout l'historique** d'une source, alors que la ligne
rendue ne porte que la dernière observation de chacune. Un mot-clé peut donc être retenu par
une observation ancienne, absente de ce que l'écran affiche ; le popover d'historique
(`GET /keywords/{id}/metrics`) est là pour la retrouver.
"""

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

from sqlalchemy import delete, func, select, text
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.errors import AppError
from app.core.urls import UrlError, normalize_url
from app.db.cursor import (
    SEARCH_MAX_ROWS,
    decode_cursor,
    decode_offset,
    encode_cursor,
    encode_offset,
)
from app.keywords import resolution, sources
from app.keywords.models import (
    Keyword,
    KeywordList,
    KeywordListItem,
    KeywordMetric,
    KeywordPageMap,
    KeywordTag,
    OrgSettings,
    Tag,
)
from app.keywords.schemas import (
    ATTACH_STATUSES,
    CONSOLIDATED,
    KeywordListOut,
    KeywordRow,
    KeywordSuggestion,
    ListBrief,
    MetricHistory,
    MetricPoint,
    Observation,
    TagBrief,
    TagOut,
)

# Historique rendu par le popover (§1.7). Une source livre au plus une observation par
# `fetched_at` : 200 points couvrent des années d'imports mensuels sur cinq sources, et
# bornent la réponse d'un mot-clé au passé chargé.
HISTORY_LIMIT = 200

# Nombre de suggestions du champ de recherche au typing (§1.7).
SUGGEST_LIMIT = 10

logger = logging.getLogger(__name__)

# Colonnes de la vue consolidée : la fonction SQL rend déjà, pour chaque champ, le **tableau
# jsonb** de ses observations (`{value, source, fetched_at}`), ordonné et prêt à sérialiser
# (spec 2026-09-01 §3.2). Aucune logique de consolidation n'est répliquée ici.
_CONSOLIDATED_COLUMNS = """
           l.volume, l.kd, l.cpc, l.intent, l.serp_features, l.trend,
           l.last_updated"""
# **LATERAL, et surtout pas un simple JOIN.** La fonction s'ouvre sur un `GROUP BY keyword_id`
# de `keyword_metrics` (§3.3) ; écrite en `JOIN … ON l.keyword_id = k.id`, la corrélation est
# une *condition de jointure*, que le planificateur ne peut pas pousser dans l'agrégat : il
# regroupe alors **toute la table**, toutes organisations confondues, pour rendre une page de
# 50 lignes (mesuré : `Seq Scan on keyword_metrics` sur la totalité des observations). En
# LATERAL, `k.id` devient un paramètre et la condition redevient une *restriction* sur la
# colonne de regroupement, que Postgres pousse jusqu'au balayage d'index. Le test de plan de
# `test_keywords_api.py` monte la garde sur exactement ce point.
#
# `LEFT JOIN` et non `JOIN` : un mot-clé attaché à la main mais sans aucune métrique appartient
# au réservoir de la cible (prédicat §6) et doit rester visible, colonnes vides.
# `CAST(… AS text)` et non `:preferred::text` : `text()` refuse de reconnaître un paramètre
# nommé immédiatement suivi de `::`, et le laisserait tel quel dans le SQL envoyé.
_CONSOLIDATED_JOIN = """
    LEFT JOIN LATERAL (
        SELECT c.volume, c.kd, c.cpc, c.intent, c.serp_features, c.trend, c.last_updated
        FROM keyword_metrics_latest(CAST(:preferred AS text)) c
        WHERE c.keyword_id = k.id
        LIMIT 1
    ) l ON true"""

# Vue « une seule source » (décision n°3.b) : la dernière observation de cette source, lue par
# parcours arrière du btree `uq_keyword_metrics_keyword_source_fetched` (keyword_id, source,
# fetched_at) — aucun index supplémentaire n'est nécessaire.
#
# Elle rend **la même forme que la vue consolidée** : un tableau jsonb de zéro ou une entrée,
# jamais une valeur nue (décision Q13). C'est toute la promesse « une seule branche côté web » —
# sans ça, `_row` reprendrait deux branches et le contrat rendrait deux formes.
#
# `to_char(… AT TIME ZONE 'UTC', …)` et **jamais** le timestamptz nu, exactement comme la
# fonction : `jsonb_build_object` sérialiserait l'instant dans le fuseau de session Postgres
# (mesuré : `+07:00` sous `TimeZone = Asia/Bangkok`), et rien ne fixe `TimeZone` côté api. Le
# `:00` final de l'OF n'est pas un paramètre lié — SQLAlchemy ne reconnaît un `:nom` que
# précédé d'un caractère non-mot, et celui-ci suit le `F` de `OF`.
_UTC_ISO = """to_char(s.fetched_at AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.USOF:00')"""


def _source_observation(column: str) -> str:
    """Le tableau jsonb d'un champ en vue par source : zéro entrée, ou l'unique observation."""
    return f"""
           CASE WHEN s.{column} IS NULL THEN '[]'::jsonb
                ELSE jsonb_build_array(jsonb_build_object(
                         'value', s.{column},
                         'source', s.source,
                         'fetched_at', {_UTC_ISO})) END AS {column}"""


_SOURCE_COLUMNS = (
    ",".join(
        _source_observation(column)
        for column in ("volume", "kd", "cpc", "intent", "serp_features", "trend")
    )
    + """,
           s.fetched_at AS last_updated"""
)
_SOURCE_JOIN = """
    LEFT JOIN LATERAL (
        SELECT m.volume, m.kd, m.cpc, m.intent, m.serp_features, m.trend,
               m.source, m.fetched_at
        FROM keyword_metrics m
        WHERE m.keyword_id = k.id AND m.source = :view
        ORDER BY m.fetched_at DESC
        LIMIT 1
    ) s ON true"""

# Statut d'attache calculé côté SQL (§6). L'agrégat rend toujours une ligne, même sans map :
# `count(*) = 0` donne `orphan`, l'état de premier rang du backlog d'opportunités (§0).
_ATTACH_JOIN = """
    LEFT JOIN LATERAL (
        SELECT CASE
                 WHEN count(*) = 0 THEN 'orphan'
                 WHEN count(DISTINCT m.url_normalized) > 1 THEN 'multi'
                 WHEN count(*) FILTER (WHERE m.page_id IS NOT NULL) > 0 THEN 'attached'
                 ELSE 'pending'
               END AS status
        FROM keyword_page_map m
        WHERE m.keyword_id = k.id AND m.site_target_id = :target
    ) att ON true"""

# Prédicat d'appartenance au réservoir d'une cible (review I6, §6) : membre d'au moins une
# liste de la cible **OU** porteur d'au moins un map sur la cible. Un mot-clé attaché à la main
# puis retiré de toute liste reste donc visible — jamais d'attache invisible.
_BELONGS = """
      AND (EXISTS (SELECT 1
                     FROM keyword_list_items i
                     JOIN keyword_lists kl ON kl.id = i.list_id
                    WHERE i.keyword_id = k.id AND kl.site_target_id = :target)
           OR EXISTS (SELECT 1
                        FROM keyword_page_map m2
                       WHERE m2.keyword_id = k.id AND m2.site_target_id = :target))"""


# --- recherche plein texte (correctif T6) ----------------------------------------------------
#
# ## Ce qui n'allait pas
#
# La recherche s'écrivait `tsv @@ plainto_tsquery(:q) OR keyword % :q`. Le `OR` trigramme est
# une **disjonction**, donc un élargissement : « jewelry manufacturing » remontait tout ce qui
# ressemble d'assez loin à la chaîne entière (seuil 0.3), soit « zbl jewelry » et
# « wire wrapping… » en tête d'une recherche à deux mots. L'ordre restant celui du curseur
# (`id DESC`), rien ne rattrapait le bruit : la pertinence n'était jamais calculée.
#
# ## Ce qui le remplace
#
# 1. **ET entre les mots, préfixe sur le dernier** — `to_tsquery('simple', 'jewelry &
#    manufactur:*')`. L'ET est ce que l'utilisateur veut dire en tapant deux mots ; le préfixe
#    sur le dernier est ce qui rend la recherche utilisable **pendant la frappe**, sans quoi
#    chaque caractère d'un mot inachevé viderait la liste.
# 2. **Le trigramme devient un repli, pas une union** — il ne s'exécute que si la tsquery ne
#    rend rien. La tolérance aux fautes est conservée (« chaussure rouge » → « chaussures
#    rouges ») sans polluer la recherche qui, elle, a trouvé.
# 3. **L'ordre est la pertinence** — égalité exacte, puis préfixe de la chaîne entière, puis
#    `ts_rank`, puis `id` pour la stabilité.
#
# La tsquery est **construite en Python** et passée en paramètre lié. Les lexèmes sont extraits
# par `[^\W_]+` : uniquement des lettres et des chiffres, jamais un caractère que `to_tsquery`
# pourrait lire comme un opérateur (`&`, `|`, `!`, `:`, `(`, `'`). Aucun échappement n'est donc
# nécessaire — il n'y a rien à échapper — et une saisie ne peut pas produire d'erreur de syntaxe
# tsquery, qui serait un 500 sur une frappe. La configuration `simple` ne stemme pas et ne
# retire pas de mots vides : découper sur les non-mots suffit à retrouver ses lexèmes.
_TOKEN_RE = re.compile(r"[^\W_]+")

# Rang de pertinence, avant `ts_rank` : la chaîne tapée **est** le mot-clé (2), le mot-clé
# **commence par** la chaîne tapée (1), ou ni l'un ni l'autre (0). `ts_rank` seul ne sait pas
# faire cette différence — pour lui « jewelry manufacturing » et « jewelry manufacturing tools
# wholesale » portent les mêmes deux lexèmes —, et c'est pourtant la première attente de qui
# tape une requête exacte. `starts_with` plutôt que `LIKE` : pas de motif à échapper.
_TIER = """CASE WHEN lower(k.keyword) = :q_exact THEN 2
                     WHEN starts_with(lower(k.keyword), :q_exact) THEN 1
                     ELSE 0 END"""
_RANK = "ts_rank(k.keyword_tsv, to_tsquery('simple', :tsq))"
_TS_MATCH = "\n      AND k.keyword_tsv @@ to_tsquery('simple', :tsq)"
_TRGM_MATCH = "\n      AND k.keyword % :q"


def build_tsquery(q: str) -> str | None:
    """« jewelry manufactur » → `jewelry & manufactur:*`. `None` si la saisie n'a aucun mot.

    Le dernier mot est un préfixe parce qu'il est, pendant la frappe, celui qu'on n'a pas fini
    de taper. Une saisie qui ne rend aucun lexème (« ??? », des espaces) n'a pas de tsquery :
    l'appelant bascule alors directement sur le repli trigramme.
    """
    tokens = _TOKEN_RE.findall(q.lower())
    if not tokens:
        return None
    *head, last = tokens
    return " & ".join([*head, f"{last}:*"])


def search_params(q: str) -> dict[str, Any]:
    """Les paramètres liés que la recherche dérive de `q` — jamais du SQL assemblé."""
    return {"q": q, "tsq": build_tsquery(q), "q_exact": " ".join(q.lower().split())}


class Filters:
    """Les filtres du réservoir, déjà validés (§6). Objet plat : il n'y a rien à encapsuler."""

    __slots__ = (
        "attach_status",
        "intent",
        "list_id",
        "q",
        "source",
        "tag_id",
        "updated_since",
        "volume_max",
        "volume_min",
    )

    def __init__(
        self,
        *,
        q: str | None = None,
        list_id: uuid.UUID | None = None,
        tag_id: uuid.UUID | None = None,
        source: str | None = None,
        intent: str | None = None,
        attach_status: str | None = None,
        volume_min: int | None = None,
        volume_max: int | None = None,
        updated_since: datetime | None = None,
    ) -> None:
        self.q = q
        self.list_id = list_id
        self.tag_id = tag_id
        self.source = source
        self.intent = intent
        self.attach_status = attach_status
        self.volume_min = volume_min
        self.volume_max = volume_max
        self.updated_since = updated_since


def _metric_exists(condition: str, *, view: str) -> str:
    """Filtre « au moins une observation satisfait » (voir l'en-tête du module).

    En vue par source, la condition est en plus restreinte à cette source.

    Il n'y a **pas** de second test sur la valeur affichée : plus aucune valeur n'est élue,
    la ligne porte toutes ses sources côte à côte. Le prix, assumé : ce EXISTS balaie tout
    l'historique de la source, alors que la ligne rendue ne montre que la **dernière**
    observation de chacune — un mot-clé peut donc être retenu par une observation ancienne,
    absente de ce que l'écran affiche.
    """
    scope = "" if view == CONSOLIDATED else " AND m.source = :view"
    return f"""
      AND EXISTS (SELECT 1 FROM keyword_metrics m
                   WHERE m.keyword_id = k.id{scope} AND {condition})"""


def build_query(
    view: str, filters: Filters, *, cursor: uuid.UUID | None, fuzzy: bool = False
) -> tuple[str, list[str]]:
    """Le SQL du réservoir et la liste des paramètres qu'il attend (hors `org`/`target`/`limit`).

    `fuzzy` demande le **repli trigramme** : il ne s'emploie que sur une recherche dont la
    tsquery n'a rien rendu (voir `list_keywords`), jamais en union avec elle.
    """
    consolidated = view == CONSOLIDATED
    columns = _CONSOLIDATED_COLUMNS if consolidated else _SOURCE_COLUMNS
    join = _CONSOLIDATED_JOIN if consolidated else _SOURCE_JOIN
    where = [_BELONGS]
    params: list[str] = []
    #  Deux régimes de pagination, parce qu'il y a deux ordres. Hors recherche, l'ordre est
    #  celui de la clé et le keyset s'applique. En recherche, l'ordre est la pertinence : la
    #  reprise se fait par rang (`OFFSET`), borné à `SEARCH_MAX_ROWS`.
    searching = filters.q is not None

    if cursor is not None and not searching:
        where.append("\n      AND k.id < :cursor")
        params.append("cursor")
    if searching:
        if fuzzy:
            where.append(_TRGM_MATCH)
            order = "similarity(k.keyword, :q) DESC, k.id DESC"
            params.append("q")
        else:
            where.append(_TS_MATCH)
            order = f"{_TIER} DESC, {_RANK} DESC, k.id DESC"
            params.extend(("tsq", "q_exact"))
    else:
        order = "k.id DESC"
    if filters.list_id is not None:
        where.append(
            "\n      AND EXISTS (SELECT 1 FROM keyword_list_items i"
            "\n                   WHERE i.keyword_id = k.id AND i.list_id = :list_id)"
        )
        params.append("list_id")
    if filters.tag_id is not None:
        where.append(
            "\n      AND EXISTS (SELECT 1 FROM keyword_tags kt"
            "\n                   WHERE kt.keyword_id = k.id AND kt.tag_id = :tag_id)"
        )
        params.append("tag_id")
    if filters.source is not None:
        where.append(
            "\n      AND EXISTS (SELECT 1 FROM keyword_metrics m"
            "\n                   WHERE m.keyword_id = k.id AND m.source = :source)"
        )
        params.append("source")
    if filters.intent is not None:
        where.append(_metric_exists("m.intent = :intent", view=view))
        params.append("intent")
    if filters.volume_min is not None:
        where.append(_metric_exists("m.volume >= :volume_min", view=view))
        params.append("volume_min")
    if filters.volume_max is not None:
        where.append(_metric_exists("m.volume <= :volume_max", view=view))
        params.append("volume_max")
    if filters.updated_since is not None:
        where.append(_metric_exists("m.fetched_at >= :updated_since", view=view))
        params.append("updated_since")
    if filters.attach_status is not None:
        where.append("\n      AND att.status = :attach_status")
        params.append("attach_status")

    sql = f"""
    SELECT k.id, k.keyword, k.locale,{columns},
           att.status AS attach_status
    FROM keywords k{join}{_ATTACH_JOIN}
    WHERE k.org_id = :org{"".join(where)}
    ORDER BY {order}
    LIMIT :limit{" OFFSET :offset" if searching else ""}
    """
    return sql, params


async def preferred_source(session: AsyncSession, org_id: str) -> str | None:
    return await session.scalar(
        select(OrgSettings.preferred_source).where(OrgSettings.org_id == org_id)
    )


def resolve_view(view: str | None) -> str:
    """`consolidated` (défaut) ou le nom d'une source du registre — 422 sinon."""
    if view is None or view == CONSOLIDATED:
        return CONSOLIDATED
    if sources.get(view) is None:
        raise AppError(
            422,
            "Validation error",
            f"Vue inconnue : {view} (attendu « {CONSOLIDATED} » ou "
            f"l'une de : {', '.join(sources.names())})",
        )
    return view


async def list_keywords(
    session: AsyncSession,
    org_id: str,
    *,
    site_target_id: uuid.UUID,
    view: str | None,
    filters: Filters,
    cursor: str | None,
    limit: int,
) -> tuple[list[KeywordRow], str | None, str, str | None, bool]:
    """Une page du réservoir. Rend `(lignes, next_cursor, vue, source préférée, repli flou)`.

    En **mode recherche** (`filters.q`), l'ordre n'est plus celui de la clé mais la pertinence,
    et la pagination bascule donc sur le rang (`OFFSET`, borné à `SEARCH_MAX_ROWS`) : voir la
    note des deux dialectes de curseur dans `app/db/cursor.py`.

    Le **repli trigramme** est une seconde requête, tentée seulement quand la première page
    d'une recherche ne rend rien : la tolérance aux fautes ne doit jamais élargir une recherche
    qui a trouvé. Le drapeau rendu dit à l'écran qu'il montre des à-peu-près.
    """
    if filters.attach_status is not None and filters.attach_status not in ATTACH_STATUSES:
        raise AppError(
            422,
            "Validation error",
            f"Statut d'attache inconnu (attendu : {', '.join(ATTACH_STATUSES)})",
        )
    resolved = resolve_view(view)
    searching = filters.q is not None
    derived = search_params(filters.q) if filters.q is not None else {}
    after: uuid.UUID | None = None
    offset = 0
    if cursor is not None:
        if searching:
            offset = decode_offset(cursor)
        else:
            after = decode_cursor(cursor)
    preferred = await preferred_source(session, org_id)

    async def page(*, fuzzy: bool) -> Sequence[Mapping[str, Any]]:
        sql, names = build_query(resolved, filters, cursor=after, fuzzy=fuzzy)
        params: dict[str, Any] = {
            "org": org_id,
            "target": site_target_id,
            "limit": limit + 1,
            "offset": offset,
            "preferred": preferred,
            "view": resolved,
            "cursor": after,
            **{name: getattr(filters, name, None) for name in names if name != "cursor"},
            **derived,
        }
        started = time.perf_counter()
        found = (await session.execute(text(sql), params)).mappings().all()
        elapsed = (time.perf_counter() - started) * 1000
        #  Mesure simple exigée par le brief T3 : c'est cette requête que le benchmark T2 a
        #  désignée comme le point chaud du module (pré-filtre EXISTS, voir l'en-tête).
        logger.info(
            "réservoir : vue=%s filtres=%d flou=%s lignes=%d en %.1f ms",
            resolved,
            len(names),
            fuzzy,
            len(found),
            elapsed,
        )
        return found

    #  Une saisie sans le moindre lexème (« ??? ») n'a pas de tsquery : elle part directement au
    #  repli, qui rendra le plus souvent zéro — et c'est la réponse juste.
    fuzzy = searching and derived.get("tsq") is None
    rows = await page(fuzzy=fuzzy)
    if searching and not rows and not fuzzy and offset == 0:
        fuzzy = True
        rows = await page(fuzzy=True)

    next_cursor = None
    if len(rows) > limit:
        rows = rows[:limit]
        if fuzzy:
            #  **Le repli ne pagine pas** (review F1). Il n'est atteint que sur la *première*
            #  page — c'est la condition qui le déclenche —, donc un `next_cursor` émis ici
            #  mènerait à une page 2 qui rejouerait la recherche exacte, celle qui n'a rien
            #  rendu : une page morte. Et le corriger dans l'autre sens (propager le mode flou
            #  dans le curseur) reviendrait à faire du filet à fautes de frappe un mode de
            #  navigation, ce qu'il n'est pas : dix à-peu-près suffisent à dire « reformulez ».
            #  `fuzzy: true` dans la réponse dit au client qu'il n'y a pas de suite à demander.
            next_cursor = None
        elif searching:
            #  Au-delà de la borne, plus de page suivante : la recherche s'affine, elle ne se
            #  feuillette pas jusqu'au bout du réservoir.
            #
            #  Dérive assumée de l'`OFFSET` : entre deux pages, un import qui insère ou retire
            #  des lignes décale la fenêtre — une ligne peut être vue deux fois ou sautée. Le
            #  keyset n'a pas ce défaut, mais il exige que l'ordre soit celui de la clé, ce
            #  que la pertinence n'est pas. Sur une recherche — lue une fois, sur quelques
            #  pages, pendant que le réservoir ne bouge presque jamais en séance — le
            #  compromis est le bon ; il ne le serait pas sur le listing complet, qui garde
            #  bien son keyset.
            following = offset + limit
            next_cursor = encode_offset(following) if following < SEARCH_MAX_ROWS else None
        else:
            next_cursor = encode_cursor(rows[-1]["id"])
    keyword_ids = [row["id"] for row in rows]
    lists = await _lists_by_keyword(session, site_target_id, keyword_ids)
    tags = await _tags_by_keyword(session, org_id, keyword_ids)
    items = [_row(row, lists, tags) for row in rows]
    return items, next_cursor, resolved, preferred, fuzzy


def _row(
    row: Mapping[str, Any],
    lists: Mapping[uuid.UUID, list[ListBrief]],
    tags: Mapping[uuid.UUID, list[TagBrief]],
) -> KeywordRow:
    #  `or []` sur **chaque** champ, et pas seulement le `coalesce` de la fonction SQL : celui-ci
    #  ne couvre que le champ sans observation d'un mot-clé qui en a d'autres. Un mot-clé
    #  attaché à la main n'a aucune observation du tout — il n'existe donc pas dans
    #  `keyword_metrics_latest` (`GROUP BY keyword_id`) et le `LEFT JOIN LATERAL` rend NULL sur
    #  les six colonnes. Sans ce garde-fou, Pydantic lèverait sur `list[Observation[T]]` : un
    #  500 sur une ligne parfaitement normale du réservoir.
    keyword_id = row["id"]
    return KeywordRow(
        id=keyword_id,
        keyword=row["keyword"],
        locale=row["locale"],
        volume=[Observation[int](**entry) for entry in (row["volume"] or [])],
        kd=[Observation[float](**entry) for entry in (row["kd"] or [])],
        cpc=[Observation[float](**entry) for entry in (row["cpc"] or [])],
        intent=[Observation[str](**entry) for entry in (row["intent"] or [])],
        serp_features=[Observation[Any](**entry) for entry in (row["serp_features"] or [])],
        trend=[Observation[list[int]](**entry) for entry in (row["trend"] or [])],
        last_updated=row["last_updated"],
        attach_status=row["attach_status"],
        lists=lists.get(keyword_id, []),
        tags=tags.get(keyword_id, []),
    )


async def _lists_by_keyword(
    session: AsyncSession, site_target_id: uuid.UUID, keyword_ids: Sequence[uuid.UUID]
) -> dict[uuid.UUID, list[ListBrief]]:
    """Les listes **de la cible** auxquelles appartient chaque mot-clé de la page.

    Une seule requête pour toute la page, jamais une par ligne : le lot est borné par `limit`.
    """
    if not keyword_ids:
        return {}
    rows = await session.execute(
        select(KeywordListItem.keyword_id, KeywordList.id, KeywordList.name)
        .join(KeywordList, KeywordList.id == KeywordListItem.list_id)
        .where(
            KeywordListItem.keyword_id.in_(keyword_ids),
            KeywordList.site_target_id == site_target_id,
        )
        .order_by(KeywordList.name)
    )
    grouped: dict[uuid.UUID, list[ListBrief]] = {}
    for keyword_id, list_id, name in rows:
        grouped.setdefault(keyword_id, []).append(ListBrief(id=list_id, name=name))
    return grouped


async def _tags_by_keyword(
    session: AsyncSession, org_id: str, keyword_ids: Sequence[uuid.UUID]
) -> dict[uuid.UUID, list[TagBrief]]:
    if not keyword_ids:
        return {}
    rows = await session.execute(
        select(KeywordTag.keyword_id, Tag.id, Tag.name)
        .join(Tag, Tag.id == KeywordTag.tag_id)
        .where(KeywordTag.keyword_id.in_(keyword_ids), Tag.org_id == org_id)
        .order_by(Tag.name)
    )
    grouped: dict[uuid.UUID, list[TagBrief]] = {}
    for keyword_id, tag_id, name in rows:
        grouped.setdefault(keyword_id, []).append(TagBrief(id=tag_id, name=name))
    return grouped


# --- suggestions au typing (§1.7) -----------------------------------------------------------
#
# La forme du SQL **est** le budget de latence. Le classement et la coupe à dix se font dans une
# sous-requête qui ne touche que `keywords` (index GIN sur `keyword_tsv`, prédicat
# d'appartenance sur des index) ; le volume n'est cherché **qu'après**, en LATERAL sur les dix
# survivants. L'écrire d'un seul tenant ferait consolider tous les candidats — le travail que
# le benchmark T2 a mesuré à 480 000 buffers — pour n'en afficher que dix.
#
# `max(m.volume)` et non la fonction de consolidation : une suggestion n'a la place que d'un
# chiffre (§3.5), et il n'y a plus de valeur élue à lui donner. Le maximum toutes sources est
# le repère de taille attendu, il ne dépend d'aucun réglage — `preferred_source` n'a donc plus
# rien à faire ici. L'agrégat sans `GROUP BY` rend toujours une ligne (`NULL` sur ensemble
# vide) : un mot-clé sans aucune observation garde bien son `volume: null`.
_SUGGEST_SQL = f"""
SELECT s.id, s.keyword, l.volume
FROM (
    SELECT k.id, k.keyword
    FROM keywords k
    WHERE k.org_id = :org{{match}}{_BELONGS}
    ORDER BY {{order}}
    LIMIT :limit
) s
LEFT JOIN LATERAL (
    SELECT max(m.volume) AS volume
    FROM keyword_metrics m
    WHERE m.keyword_id = s.id
) l ON true
"""


async def suggest_keywords(
    session: AsyncSession, org_id: str, *, site_target_id: uuid.UUID, q: str
) -> list[KeywordSuggestion]:
    """Les dix mots-clés du réservoir de la cible qui complètent le mieux `q`.

    Même prédicat d'appartenance et même classement que le réservoir : ce que la liste propose
    est exactement ce que la recherche rendra. Un repli trigramme est tenté si la tsquery ne
    rend rien, pour la même raison qu'ailleurs — sans lui, une faute de frappe viderait la
    liste alors que la recherche, elle, trouverait.

    Le volume rendu est le **maximum toutes sources** (§3.5) : la source préférée de
    l'organisation n'entre pas dans ce calcul, une suggestion n'affiche qu'un chiffre et il ne
    doit pas dépendre d'un réglage.
    """
    derived = search_params(q)
    params: dict[str, Any] = {
        "org": org_id,
        "target": site_target_id,
        "limit": SUGGEST_LIMIT,
        **derived,
    }

    async def run(*, fuzzy: bool) -> Sequence[Mapping[str, Any]]:
        sql = _SUGGEST_SQL.format(
            match=_TRGM_MATCH if fuzzy else _TS_MATCH,
            order=(
                "similarity(k.keyword, :q) DESC, k.id DESC"
                if fuzzy
                else f"{_TIER} DESC, {_RANK} DESC, k.id DESC"
            ),
        )
        return (await session.execute(text(sql), params)).mappings().all()

    rows = await run(fuzzy=derived["tsq"] is None)
    if not rows and derived["tsq"] is not None:
        rows = await run(fuzzy=True)
    return [
        KeywordSuggestion(keyword_id=row["id"], keyword=row["keyword"], volume=row["volume"])
        for row in rows
    ]


# --- historique d'un mot-clé (popover, §1.7) -----------------------------------------------


async def get_keyword(session: AsyncSession, org_id: str, keyword_id: uuid.UUID) -> Keyword:
    """Le mot-clé de l'organisation, ou 404 — jamais 403 (architecture §9)."""
    keyword = await session.scalar(
        select(Keyword).where(Keyword.id == keyword_id, Keyword.org_id == org_id)
    )
    if keyword is None:
        raise AppError(404, "Not Found", "Mot-clé inconnu")
    return keyword


async def metric_history(
    session: AsyncSession, org_id: str, keyword_id: uuid.UUID
) -> MetricHistory:
    """Toutes les observations connues du mot-clé, la plus récente d'abord.

    Le popover affiche la **provenance de chaque valeur** et la sparkline du volume : il lui
    faut les lignes brutes, pas la consolidation — celle-ci est déjà dans la table, et la
    question posée ici est justement « d'où vient ce chiffre, et qu'en disaient les autres ? ».

    Aucun filtre par cible : les observations sont mutualisées à l'organisation (décision n°2),
    une même ligne sert tous les sites du même marché. La cible de l'appel n'est validée que
    pour refuser un appel hors organisation.
    """
    keyword = await get_keyword(session, org_id, keyword_id)
    rows = await session.execute(
        select(
            KeywordMetric.source,
            KeywordMetric.fetched_at,
            KeywordMetric.volume,
            KeywordMetric.kd,
            KeywordMetric.cpc,
            KeywordMetric.intent,
            KeywordMetric.serp_features,
            KeywordMetric.trend,
        )
        .where(KeywordMetric.keyword_id == keyword_id, KeywordMetric.org_id == org_id)
        .order_by(KeywordMetric.fetched_at.desc(), KeywordMetric.source)
        .limit(HISTORY_LIMIT)
    )
    return MetricHistory(
        keyword_id=keyword.id,
        keyword=keyword.keyword,
        locale=keyword.locale,
        items=[MetricPoint.model_validate(row) for row in rows.mappings()],
    )


# --- listes de destination (§3.4) ------------------------------------------------------------


async def get_list(session: AsyncSession, org_id: str, list_id: uuid.UUID) -> KeywordList:
    found = await session.scalar(
        select(KeywordList).where(KeywordList.id == list_id, KeywordList.org_id == org_id)
    )
    if found is None:
        raise AppError(404, "Not Found", "Liste de mots-clés inconnue")
    return found


def _item_count() -> Any:
    """Décompte corrélé des membres d'une liste — servi par la PK `(list_id, keyword_id)`."""
    return (
        select(func.count())
        .select_from(KeywordListItem)
        .where(KeywordListItem.list_id == KeywordList.id)
        .scalar_subquery()
        .label("item_count")
    )


async def list_keyword_lists(
    session: AsyncSession,
    org_id: str,
    *,
    site_target_id: uuid.UUID,
    cursor: str | None,
    limit: int,
) -> tuple[list[KeywordListOut], str | None]:
    after = None if cursor is None else decode_cursor(cursor)
    query = (
        select(
            KeywordList.id,
            KeywordList.site_target_id,
            KeywordList.name,
            KeywordList.source,
            KeywordList.created_from_import_id,
            KeywordList.created_at,
            _item_count(),
        )
        .where(KeywordList.org_id == org_id, KeywordList.site_target_id == site_target_id)
        .order_by(KeywordList.id.desc())
        .limit(limit + 1)
    )
    if after is not None:
        query = query.where(KeywordList.id < after)
    rows = (await session.execute(query)).mappings().all()
    next_cursor = None
    if len(rows) > limit:
        rows = rows[:limit]
        next_cursor = encode_cursor(rows[-1]["id"])
    return [KeywordListOut.model_validate(row) for row in rows], next_cursor


async def create_keyword_list(
    session: AsyncSession, org_id: str, *, site_target_id: uuid.UUID, name: str
) -> KeywordListOut:
    """Crée une liste vide sur la cible. L'UNIQUE (cible, nom) est l'arbitre du 409."""
    created = KeywordList(org_id=org_id, site_target_id=site_target_id, name=name)
    session.add(created)
    try:
        await session.commit()
    except IntegrityError as exc:
        await session.rollback()
        raise AppError(
            409, "Conflict", f"Une liste « {name} » existe déjà pour cette cible"
        ) from exc
    return KeywordListOut(
        id=created.id,
        site_target_id=created.site_target_id,
        name=created.name,
        source=created.source,
        created_from_import_id=created.created_from_import_id,
        created_at=created.created_at,
        item_count=0,
    )


async def _keywords_of_org(
    session: AsyncSession, org_id: str, keyword_ids: Sequence[uuid.UUID]
) -> list[uuid.UUID]:
    """Les identifiants qui désignent vraiment un mot-clé de l'organisation.

    Un identifiant étranger est **ignoré**, pas refusé : un geste de masse porte sur une
    sélection qu'un autre onglet a pu vider entre-temps, et faire échouer les 49 autres
    lignes pour une disparue serait le mauvais arbitrage. Le décompte rendu dit la vérité.
    """
    rows = await session.execute(
        select(Keyword.id).where(Keyword.id.in_(keyword_ids), Keyword.org_id == org_id)
    )
    return list(rows.scalars())


async def add_list_items(
    session: AsyncSession, org_id: str, list_id: uuid.UUID, keyword_ids: Sequence[uuid.UUID]
) -> int:
    target_list = await get_list(session, org_id, list_id)
    valid = await _keywords_of_org(session, org_id, keyword_ids)
    if not valid:
        return 0
    result = await session.execute(
        pg_insert(KeywordListItem)
        .values(
            [
                {"list_id": target_list.id, "keyword_id": keyword_id, "org_id": org_id}
                for keyword_id in valid
            ]
        )
        .on_conflict_do_nothing()
    )
    await session.commit()
    return result.rowcount


async def remove_list_items(
    session: AsyncSession, org_id: str, list_id: uuid.UUID, keyword_ids: Sequence[uuid.UUID]
) -> int:
    """Retire des mots-clés d'une liste — jamais du réservoir de l'organisation.

    Un mot-clé retiré de toutes les listes d'une cible reste visible tant qu'il porte une
    attache sur cette cible (prédicat d'appartenance §6) : c'est voulu, une attache invisible
    serait pire qu'une ligne en trop.
    """
    target_list = await get_list(session, org_id, list_id)
    result = await session.execute(
        delete(KeywordListItem).where(
            KeywordListItem.list_id == target_list.id,
            KeywordListItem.keyword_id.in_(keyword_ids),
            KeywordListItem.org_id == org_id,
        )
    )
    await session.commit()
    return result.rowcount


# --- tags (§3.6) ------------------------------------------------------------------------------


async def list_tags(
    session: AsyncSession, org_id: str, *, cursor: str | None, limit: int
) -> tuple[list[TagOut], str | None]:
    """Les tags de l'organisation — ils ne sont pas scopés à une cible (§3.6)."""
    after = None if cursor is None else decode_cursor(cursor)
    query = (
        select(Tag.id, Tag.name, Tag.created_at)
        .where(Tag.org_id == org_id)
        .order_by(Tag.id.desc())
        .limit(limit + 1)
    )
    if after is not None:
        query = query.where(Tag.id < after)
    rows = (await session.execute(query)).mappings().all()
    next_cursor = None
    if len(rows) > limit:
        rows = rows[:limit]
        next_cursor = encode_cursor(rows[-1]["id"])
    return [TagOut.model_validate(row) for row in rows], next_cursor


async def create_tag(session: AsyncSession, org_id: str, name: str) -> Tag:
    created = Tag(org_id=org_id, name=name)
    session.add(created)
    try:
        await session.commit()
    except IntegrityError as exc:
        await session.rollback()
        raise AppError(409, "Conflict", f"Le tag « {name} » existe déjà") from exc
    return created


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


async def add_keyword_tags(
    session: AsyncSession, org_id: str, tag_id: uuid.UUID, keyword_ids: Sequence[uuid.UUID]
) -> int:
    tag = await get_tag(session, org_id, tag_id)
    valid = await _keywords_of_org(session, org_id, keyword_ids)
    if not valid:
        return 0
    result = await session.execute(
        pg_insert(KeywordTag)
        .values(
            [{"keyword_id": keyword_id, "tag_id": tag.id, "org_id": org_id} for keyword_id in valid]
        )
        .on_conflict_do_nothing()
    )
    await session.commit()
    return result.rowcount


async def remove_keyword_tags(
    session: AsyncSession, org_id: str, tag_id: uuid.UUID, keyword_ids: Sequence[uuid.UUID]
) -> int:
    tag = await get_tag(session, org_id, tag_id)
    result = await session.execute(
        delete(KeywordTag).where(
            KeywordTag.tag_id == tag.id,
            KeywordTag.keyword_id.in_(keyword_ids),
            KeywordTag.org_id == org_id,
        )
    )
    await session.commit()
    return result.rowcount


# --- attache manuelle (§3.7, §3.9) ------------------------------------------------------------


async def upsert_manual_map(
    session: AsyncSession,
    org_id: str,
    *,
    keyword_id: uuid.UUID,
    site_target_id: uuid.UUID,
    url: str,
) -> KeywordPageMap:
    """Déclare **la** page cible d'un mot-clé — le geste central du module (§1.8).

    Sémantique d'`upsert`, et c'est un choix de fond : l'attache manuelle est *unique* par
    (cible, mot-clé) (décision n°6), et l'utilisateur qui change d'avis sur la page à écrire
    doit pouvoir le dire d'un geste. Un 409 « détachez d'abord » l'obligerait à un aller-retour
    pour exprimer la même décision — d'où `PUT`, verbe idempotent : ce corps décrit l'état
    voulu de l'attache, pas un événement à ajouter.

    Ce qui n'est **jamais** touché ici : les maps `import_url` et `gsc` de ce mot-clé. Ce sont
    des observations (« telle source a vu ce mot-clé sur telle URL »), elles coexistent avec la
    décision de planification et se rafraîchissent toutes seules. Réciproquement, l'automatique
    n'écrase jamais ce `manual`.

    `page_id` est posé si l'URL correspond à une page déjà connue de la cible ; sinon il reste
    `NULL` — « en attente de crawl », l'état normal d'une page qui n'existe pas encore.
    """
    try:
        url_normalized = normalize_url(url)
    except UrlError as exc:
        raise AppError(422, "Validation error", f"URL invalide : {exc}") from exc
    resolved = await resolution.resolve_page_ids(session, site_target_id, [url_normalized])
    page_id = resolved.get(url_normalized)
    now = datetime.now(UTC)
    result = await session.execute(
        pg_insert(KeywordPageMap)
        .values(
            org_id=org_id,
            site_target_id=site_target_id,
            keyword_id=keyword_id,
            url_normalized=url_normalized,
            page_id=page_id,
            source="manual",
            # Une décision n'a pas de position observée : `position` reste NULL, contrairement
            # aux maps `import_url` qui portent celle de leur export.
            seen_at=now,
        )
        # L'arbitre est l'index partiel `uq_keyword_page_map_manual` — le seul qui dise
        # « une attache de planification par (cible, mot-clé) ». Redéclarer la même URL
        # retombe sur la même ligne : `seen_at` est rafraîchi, rien n'est dupliqué.
        .on_conflict_do_update(
            index_elements=[KeywordPageMap.site_target_id, KeywordPageMap.keyword_id],
            index_where=text("source = 'manual'"),
            set_={"url_normalized": url_normalized, "page_id": page_id, "seen_at": now},
        )
        .returning(KeywordPageMap)
    )
    await session.commit()
    return result.scalar_one()


async def delete_manual_map(
    session: AsyncSession, org_id: str, *, keyword_id: uuid.UUID, site_target_id: uuid.UUID
) -> None:
    """Détache — **le map `manual` uniquement** (§6).

    Les observations `import_url`/`gsc` ne se suppriment pas à la main : elles disent ce
    qu'une source a vu, et l'utilisateur ne réécrit pas le passé. Rien à supprimer = 404,
    pas un 204 complaisant : l'écran doit pouvoir dire que l'attache n'existait plus.
    """
    result = await session.execute(
        delete(KeywordPageMap).where(
            KeywordPageMap.keyword_id == keyword_id,
            KeywordPageMap.site_target_id == site_target_id,
            KeywordPageMap.org_id == org_id,
            KeywordPageMap.source == "manual",
        )
    )
    if result.rowcount == 0:
        await session.rollback()
        raise AppError(404, "Not Found", "Aucune attache manuelle sur cette cible")
    await session.commit()


# --- réglages d'organisation (§3.10) -------------------------------------------------------


async def get_org_settings(session: AsyncSession, org_id: str) -> OrgSettings | None:
    """La ligne de réglages, ou `None` : une lecture ne crée jamais de ligne."""
    return await session.scalar(select(OrgSettings).where(OrgSettings.org_id == org_id))


async def update_org_settings(
    session: AsyncSession, org_id: str, updates: Mapping[str, Any]
) -> OrgSettings | None:
    """Applique un PATCH partiel. `updates` ne porte que les champs **explicitement fournis**.

    Sémantique PATCH tenue à la lettre : un champ absent du corps est **inchangé**, un champ à
    `null` est effacé. Confondre les deux est anodin avec un seul réglage — `{}` effacerait la
    source préférée — et devient une perte de données silencieuse dès qu'un second réglage
    arrivera dans cette table. D'où `model_dump(exclude_unset=True)` côté route.

    `preferred_source` est validée contre le **registre d'adaptateurs** (§3.10 : 422), jamais
    contre une contrainte en base : la liste des sources est du code et bouge à chaque tranche.

    L'écriture passe par l'ORM et non par un `update()` Core : `updated_at` est un `onupdate`
    **côté SQLAlchemy** — Postgres n'a pas d'ON UPDATE, un UPDATE en SQL brut laisserait la
    colonne figée (commentaire du modèle).
    """
    if "preferred_source" in updates:
        preferred = updates["preferred_source"]
        if preferred is not None and sources.get(preferred) is None:
            raise AppError(
                422,
                "Validation error",
                f"Source inconnue : {preferred} (connues : {', '.join(sources.names())})",
            )
    settings = await get_org_settings(session, org_id)
    if not updates:
        #  Rien à écrire : un PATCH vide ne crée pas de ligne et n'en touche aucune.
        return settings
    if settings is None:
        settings = OrgSettings(org_id=org_id, preferred_source=updates.get("preferred_source"))
        session.add(settings)
    else:
        for field, value in updates.items():
            setattr(settings, field, value)
    await session.commit()
    return settings