"""Schémas Pydantic du module Imports (spec Imports §6). Conventions du module Sites reprises à la lettre : `extra='forbid'` sur toutes les entrées (un champ inattendu est un 422, jamais un silence), `from_attributes` sur les sorties. """ import uuid from datetime import datetime from typing import Annotated, Any, Literal from pydantic import BaseModel, ConfigDict, Field, StringConstraints, model_validator Sha256 = Annotated[str, StringConstraints(pattern=r"^[0-9a-f]{64}$")] ListName = Annotated[str, StringConstraints(min_length=1, max_length=200)] # Aperçu : assez de lignes pour juger un mapping, assez peu pour rester synchrone (§1.4). PREVIEW_ROWS = 20 class UploadOut(BaseModel): """Réponse de `POST /imports/files` — le fichier est sur disque, rien n'est en base.""" sha256: str filename: str size: int # Import existant portant le même sha256 **sur la même cible** (§1.3) : un avertissement # doux au wizard, jamais un refus — le re-import après correction d'adaptateur est légitime. duplicate_of: uuid.UUID | None = None class PreviewIn(BaseModel): model_config = ConfigDict(extra="forbid") sha256: Sha256 # Format imposé par l'utilisateur (override de la détection, étape 2 du wizard). format: str | None = None # Mapping manuel des colonnes, obligatoire pour `generic_csv` (§5). columns: dict[str, str] | None = None # Colonne qui porte le pays, **hors mapping canonique** et pour `generic_csv` seulement # (§4.1, Q4) : les autres adaptateurs connaissent la leur. Absente, le fichier n'a pas de # pays et le recensement rend `[]`. country_column: str | None = None # Facultatif : sans cible, l'api ne connaît pas le site et `target_ids` reste vide partout # — le recensement, lui, est rendu quand même (§4.2). site_target_id: uuid.UUID | None = None class PreviewRow(BaseModel): """Une ligne canonique de l'aperçu — ce que l'adaptateur produirait à l'import.""" line: int keyword: str volume: int | None = None kd: float | None = None cpc: float | None = None intent: str | None = None serp_features: list[str] | None = None trend: list[int] | None = None url: str | None = None position: float | None = None class PreviewWarning(BaseModel): code: Literal[ "ambiguous_format", "ignored_columns", # `country_mismatch` a disparu avec le pays « annoncé par le fichier » : le fichier # n'annonce plus **un** pays, il en porte N, et l'écart ne se dit plus par un # avertissement mais par le recensement lui-même (`PreviewOut.countries`, §4.2). "no_rows", # Proposition automatique de mapping (§5, décision 2026-09-01) : ce qu'on a reconnu, # ce dont on hésite, et le cas où l'on n'a rien pu proposer. "mapping_suggested", "ambiguous_column", "no_mapping_suggested", ] detail: str class DetectionOut(BaseModel): source: str language: str confidence: float class CountryCount(BaseModel): """Un libellé de pays du fichier, son décompte et ce qu'on en sait (§4.2). Une entrée **par libellé brut distinct**, pas par code : c'est ce que l'utilisateur lit dans son fichier, et c'est ce qu'il doit reconnaître à l'écran. Deux libellés peuvent donc rendre le même `code` (« UK » et « United Kingdom ») — le wizard dédoublonne par `code` avant de créer des cibles. `label` n'est **jamais nul**, par contrat avec le miroir Zod (`web/lib/imports.ts`) : pour un libellé non résolu, c'est le libellé brut lui-même. L'écran affiche toujours quelque chose de lisible, y compris pour « New York ». """ # Le libellé tel qu'il est écrit dans le fichier — la seule chose que l'utilisateur # reconnaît, et ce que le rapport d'import lui redira mot pour mot (§4.5). raw: str # ISO 3166-1 alpha-2, ou `None` : ville, libellé inconnu, valeur illisible. Jamais deviné. code: str | None # Nom français du pays, ou `raw` quand il n'est pas résolu. Non nul, voir la docstring. label: str rows: int # Langue à **proposer** pour la cible à créer, ou `None` quand CLDR ne donne rien de sûr # (8 territoires sur 249) : le wizard laisse alors l'utilisateur choisir. suggested_language: str | None # Cibles déjà existantes du site sur ce pays. **Pluriel** : l'unicité porte sur le triplet # (site, pays, langue), un pays peut donc porter `en-GB` et `fr-GB`. Vide quand le preview # n'a pas reçu de `site_target_id` — l'api ne connaît alors aucun site. target_ids: list[uuid.UUID] class PreviewOut(BaseModel): """Détection + aperçu (§6). **Rien n'est persisté** par cet endpoint.""" format: str confidence: float language: str | None # En-têtes réels du fichier, dans leur ordre d'origine. headers: list[str] # Champ canonique → en-tête réel retenu. columns: dict[str, str] # Mapping proposé par heuristique de synonymes quand aucun adaptateur ne reconnaît le # fichier (§5, décision 2026-09-01). Même forme que `ImportCreate.columns` : l'étape 2 du # wizard le pré-remplit et l'utilisateur **confirme** au lieu de câbler. `None` quand aucune # colonne mot-clé n'est trouvable — `generic_csv` refuserait le mapping sans elle. suggested_mapping: dict[str, str] | None = None # Recensement des pays du fichier **entier** (§4.2), trié par nombre de lignes décroissant. # Vide quand la source n'écrit pas son pays (Semrush) : le cloisonnement ne se retourne # jamais contre les sources muettes, leur import reste celui d'avant. countries: list[CountryCount] = [] candidates: list[DetectionOut] sample: list[PreviewRow] warnings: list[PreviewWarning] class ListRef(BaseModel): """Liste de destination : existante (`id`) ou à créer (`name`), jamais les deux (§1.5).""" model_config = ConfigDict(extra="forbid") id: uuid.UUID | None = None name: ListName | None = None @model_validator(mode="after") def _exactly_one(self) -> "ListRef": if (self.id is None) == (self.name is None): raise ValueError("la liste de destination se donne par « id » ou par « name »") return self class ImportCreate(BaseModel): model_config = ConfigDict(extra="forbid") sha256: Sha256 format: str site_target_id: uuid.UUID list: ListRef # Étiquette d'affichage rendue par `POST /imports/files` — jamais un composant de chemin. filename: Annotated[str, StringConstraints(min_length=1, max_length=255)] | None = None # « Date des données » (§1.5) : aucune source n'en porte de fiable, elle est déclarée. fetched_at: datetime columns: dict[str, str] | None = None # Colonne qui porte le pays de chaque ligne, **hors mapping canonique** et pour # `generic_csv` seulement (§4.1, Q4) — le pendant exact de `PreviewIn.country_column`, sans # lequel la colonne désignée à l'aperçu ne suivrait pas jusqu'au job et l'import ne serait # pas cloisonné. Les adaptateurs à signature connaissent la leur et l'ignorent. country_column: str | None = None # Passe outre l'avertissement de doublon sha256 sur cette cible (§1.3). force: bool = False @model_validator(mode="after") def _tz_aware(self) -> "ImportCreate": if self.fetched_at.tzinfo is None: # Sans fuseau, la même « date des données » désignerait deux instants selon le # serveur — et `(keyword_id, source, fetched_at)` cesserait d'identifier une # observation (§3.2). raise ValueError("fetched_at doit porter un fuseau horaire") return self class ImportOut(BaseModel): model_config = ConfigDict(from_attributes=True) # Discriminant du listing fusionné imports + événements de rattachement (§6, §12.2). entry: Literal["import"] = "import" id: uuid.UUID site_target_id: uuid.UUID source: str kind: str filename: str | None file_sha256: str | None detected_format: str | None detected_country: str | None fetched_at: datetime keyword_list_id: uuid.UUID | None job_id: uuid.UUID status: str error: str | None report: dict[str, Any] | None created_at: datetime class ReconcileEventOut(BaseModel): """Un job `map_reconcile` rendu comme événement de l'onglet Imports (§1.2, §12.2). Aucune table d'événements : la trace est le job lui-même. Tant que le module Crawl n'existe pas, aucun job de ce type n'est enfilé — le listing est donc un no-op propre. """ entry: Literal["map_reconcile"] = "map_reconcile" id: uuid.UUID site_target_id: uuid.UUID | None status: str report: dict[str, Any] | None error: str | None created_at: datetime class ImportCreated(BaseModel): import_: Annotated[ImportOut, Field(alias="import")] job_id: uuid.UUID model_config = ConfigDict(populate_by_name=True) class ImportListOut(BaseModel): items: list[ImportOut | ReconcileEventOut] next_cursor: str | None