"""Tables DATA du réservoir de mots-clés et des imports (spec Imports §3). Pur DDL : aucune logique de pipeline, aucun endpoint (T2/T3). Les tables sont regroupées ici plutôt que réparties entre `app/keywords/` et `app/imports/` parce que `keyword_lists` et `imports` se référencent mutuellement (§3.4 / §3.11) — une seule unité de définition rend la dépendance circulaire lisible et évite un import croisé entre deux modules. Invariants transverses (architecture §8, data §7) : - toute table porte `org_id TEXT NOT NULL`, filtré dans chaque requête ; - **jamais d'index dédié quand un composite existant est préfixé par la colonne** — la règle vaut pour `org_id` comme pour `site_target_id` et toute autre colonne de filtre (cf. commit « index org_id redondant retiré ») ; chaque colonne concernée porte le commentaire disant quel btree la sert déjà ; - PK `UUID` avec `server_default=uuidv7()` (Postgres 18) : ordre temporel, curseur sur une seule colonne, id opaque ; - horodatages en `timestamptz` ; - **rétention org à la suppression** (décision n°4) : ce qui est rattaché à une cible part en CASCADE (`keyword_lists` + items, `keyword_page_map`, `imports`), le réservoir de l'org survit (`keywords`, `keyword_metrics`, `tags`). """ import uuid from datetime import datetime from decimal import Decimal from typing import Any from pgvector.sqlalchemy import HALFVEC from sqlalchemy import ( CheckConstraint, Computed, DateTime, ForeignKey, Index, Integer, Numeric, SmallInteger, Text, UniqueConstraint, func, text, ) from sqlalchemy.dialects.postgresql import ARRAY, JSONB, TSVECTOR, UUID from sqlalchemy.orm import Mapped, mapped_column from app.db.base import Base # Recopié du statut du job porteur (§3.11) — même vocabulaire que `jobs.status`. IMPORT_STATUSES = ("queued", "running", "done", "failed") IMPORT_KINDS = ("csv", "api") # `gsc` est **réservée au module 3** : aucune hydratation GSC ici (§3.7). MAP_SOURCES = ("import_url", "manual", "gsc") # **Aucune liste de sources ici** (spec 2026-09-01 §3.1, décision A1). `CONSOLIDATION_SOURCES`, # `FALLBACK_SOURCES` et `KNOWN_SOURCES` n'existaient que pour verrouiller le nommage des # adaptateurs contre les `CASE` de priorité de `keyword_metrics_latest` ; ces `CASE` ont # disparu avec la hiérarchie, et une source inconnue du SQL n'est plus traitée autrement # qu'une source connue — le verrou n'avait plus d'objet. # Conséquence assumée : `keyword_metrics.source` n'a ni CHECK ni FK, donc une faute de frappe # dans un futur adaptateur ne fera plus échouer aucun test — elle créera une source fantôme, # visible comme filtre du réservoir et comme `?view=`. Le remplacement honnête, le jour # où un adaptateur s'ajoute (T7/T8 DataForSEO), est un test qui fige `sources.names()`. class Keyword(Base): """Le mot-clé canonique de l'organisation (canon data §7). Mutualisé entre les sites d'une même org sur le même marché : un fetch payé une fois sert tous les sites (décision n°2). Aucune colonne de score, aucune métrique ici — les valeurs vivent dans `keyword_metrics`, sourcées et horodatées. """ __tablename__ = "keywords" __mapper_args__ = {"eager_defaults": True} id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, server_default=text("uuidv7()") ) # Pas d'index dédié : le btree de UNIQUE (org_id, locale, keyword) est préfixé par org_id. org_id: Mapped[str] = mapped_column(Text, nullable=False) # Marché complet langue-pays (décision n°2) : `fr-FR` ≠ `fr-BE`. Hérité de la cible # (`site_targets.language` + `country` recombinés), jamais déduit du fichier importé. locale: Mapped[str] = mapped_column(Text, nullable=False) # Normalisé par `app.keywords.normalize.normalize_keyword` (NFC + trim + minuscules + # espaces réduits) : c'est ce qui rend l'unicité ci-dessous fiable entre sources. keyword: Mapped[str] = mapped_column(Text, nullable=False) # Colonne générée : la recherche plein texte du réservoir ne recalcule jamais le vecteur, # et aucun chemin d'écriture ne peut le laisser désynchronisé. `'simple'` (et non une # config par langue) : le réservoir est multilingue et une config est exigée immuable. keyword_tsv: Mapped[str] = mapped_column( TSVECTOR, Computed("to_tsvector('simple', keyword)", persisted=True), nullable=False ) # Embeddings v2.x (data §3) : colonne prête, **sans index** — l'index HNSW est hors # périmètre et coûterait à l'écriture pour une colonne encore vide. embedding: Mapped[Any | None] = mapped_column(HALFVEC(1024), nullable=True) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) __table_args__ = ( CheckConstraint("locale ~ '^[a-z]{2}-[A-Z]{2}$'", name="ck_keywords_locale"), UniqueConstraint("org_id", "locale", "keyword", name="uq_keywords_org_locale_keyword"), # Recherche plein texte du réservoir (§6 : jamais de LIKE '%…%' non indexé). Index("ix_keywords_tsv", "keyword_tsv", postgresql_using="gin"), # Repli flou de la même recherche (fautes de frappe, sous-chaînes). Index( "ix_keywords_keyword_trgm", "keyword", postgresql_using="gin", postgresql_ops={"keyword": "gin_trgm_ops"}, ), ) class KeywordMetric(Base): """Une observation d'une source à une date — jamais écrasée par une autre source. L'historique et le trend sortent gratuitement de l'empilement ; la consolidation est une lecture (fonction `keyword_metrics_latest`, §3.3), jamais une écriture. """ __tablename__ = "keyword_metrics" __mapper_args__ = {"eager_defaults": True} id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, server_default=text("uuidv7()") ) # Index dédié : aucune contrainte de cette table n'est préfixée par org_id. org_id: Mapped[str] = mapped_column(Text, nullable=False, index=True) keyword_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("keywords.id", ondelete="CASCADE", name="fk_keyword_metrics_keyword_id"), nullable=False, ) # Nom d'adaptateur (`semrush_positions`, `generic_csv`, …) : la provenance de la valeur. source: Mapped[str] = mapped_column(Text, nullable=False) # « Date des données » déclarée au wizard (CSV) ou horodatage du fetch (API). fetched_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) # Valeurs telles que fournies par la source : nullables, jamais complétées ni corrigées. volume: Mapped[int | None] = mapped_column(Integer, nullable=True) kd: Mapped[Decimal | None] = mapped_column(Numeric, nullable=True) cpc: Mapped[Decimal | None] = mapped_column(Numeric, nullable=True) intent: Mapped[str | None] = mapped_column(Text, nullable=True) serp_features: Mapped[Any | None] = mapped_column(JSONB, nullable=True) trend: Mapped[list[int] | None] = mapped_column(ARRAY(SmallInteger), nullable=True) # La ligne source complète : position et URL d'origine de l'export y sont rangées # (l'URL vit aussi dans `keyword_page_map`, §3.7). raw: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False) __table_args__ = ( # « Même clé = même observation » : le re-import après correction d'adaptateur # remplace le parse erroné (ON CONFLICT DO UPDATE, §4.3) — ce n'est jamais une # fusion inter-sources, la clé porte la provenance. UniqueConstraint( "keyword_id", "source", "fetched_at", name="uq_keyword_metrics_keyword_source_fetched" ), # Pas d'index (keyword_id, source, fetched_at DESC) : le btree de l'unicité ci-dessus # porte exactement ces colonnes et sert les deux accès réels — l'upsert par clé # complète, et « dernière valeur d'une source » (égalités sur keyword_id et source, # puis parcours arrière sur fetched_at). Un second index serait un doublon payé à # chaque écriture sur la plus grosse table du modèle. ) class KeywordList(Base): """Liste de destination d'un import — le lien cible ↔ réservoir (flow §5).""" __tablename__ = "keyword_lists" __mapper_args__ = {"eager_defaults": True} id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, server_default=text("uuidv7()") ) org_id: Mapped[str] = mapped_column(Text, nullable=False, index=True) # Pas d'index dédié : le btree de UNIQUE (site_target_id, name) est préfixé par # site_target_id — il sert le listing par cible comme la CASCADE de suppression. site_target_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("site_targets.id", ondelete="CASCADE", name="fk_keyword_lists_site_target_id"), nullable=False, ) name: Mapped[str] = mapped_column(Text, nullable=False) # Adaptateur d'origine, NULL pour une liste créée à la main. source: Mapped[str | None] = mapped_column(Text, nullable=True) # `use_alter` : `imports` référence `keyword_lists` et réciproquement — la contrainte est # posée par un ALTER TABLE après création des deux tables, jamais dans le CREATE TABLE. created_from_import_id: Mapped[uuid.UUID | None] = mapped_column( UUID(as_uuid=True), ForeignKey( "imports.id", ondelete="SET NULL", name="fk_keyword_lists_created_from_import_id", use_alter=True, ), nullable=True, ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) __table_args__ = ( UniqueConstraint("site_target_id", "name", name="uq_keyword_lists_target_name"), ) class KeywordListItem(Base): """Appartenance d'un mot-clé à une liste. **Aucune URL ici** : elle vit dans le map.""" __tablename__ = "keyword_list_items" __mapper_args__ = {"eager_defaults": True} list_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("keyword_lists.id", ondelete="CASCADE", name="fk_keyword_list_items_list_id"), primary_key=True, ) keyword_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("keywords.id", ondelete="CASCADE", name="fk_keyword_list_items_keyword_id"), primary_key=True, ) org_id: Mapped[str] = mapped_column(Text, nullable=False, index=True) added_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) __table_args__ = ( # « Quels mots-clés dans cette liste » passe par la PK ; « dans quelles listes ce # mot-clé » (prédicat d'appartenance au réservoir, §6) a besoin du sens inverse. Index("ix_keyword_list_items_keyword_id", "keyword_id"), ) class Tag(Base): __tablename__ = "tags" __mapper_args__ = {"eager_defaults": True} id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, server_default=text("uuidv7()") ) # Pas d'index dédié : le btree de UNIQUE (org_id, name) est préfixé par org_id. org_id: Mapped[str] = mapped_column(Text, nullable=False) name: Mapped[str] = mapped_column(Text, nullable=False) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) __table_args__ = (UniqueConstraint("org_id", "name", name="uq_tags_org_name"),) class KeywordTag(Base): __tablename__ = "keyword_tags" __mapper_args__ = {"eager_defaults": True} keyword_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("keywords.id", ondelete="CASCADE", name="fk_keyword_tags_keyword_id"), primary_key=True, ) tag_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("tags.id", ondelete="CASCADE", name="fk_keyword_tags_tag_id"), primary_key=True, ) org_id: Mapped[str] = mapped_column(Text, nullable=False, index=True) __table_args__ = ( # Filtre « mots-clés portant ce tag » : la PK ne le sert pas (tag_id en seconde # position). Index("ix_keyword_tags_tag_id", "tag_id"), ) class KeywordPageMap(Base): """L'attache keyword → page : LA sortie du module, consommée par la Matrice (§3.7). `page_id` est **nullable et sans FK** : la table `pages` appartient au module Crawl et n'existe pas encore. Sa migration posera la FK (`ON DELETE SET NULL`). D'ici là, une URL déclarée à la main vit très bien sans page — c'est la planification d'une page pas encore publiée, le geste central du module. « Une observation se rafraîchit, une décision jamais » : les maps `import_url`/`gsc` mettent à jour `position`/`seen_at` quand une observation plus récente arrive ; le map `manual` n'est jamais écrasé par l'automatique. """ __tablename__ = "keyword_page_map" __mapper_args__ = {"eager_defaults": True} id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, server_default=text("uuidv7()") ) org_id: Mapped[str] = mapped_column(Text, nullable=False, index=True) # Pas d'index dédié : site_target_id est la colonne de tête de l'UNIQUE à 4 colonnes # ci-dessous. C'est la table la plus écrite du modèle — un index de plus se paie à # chaque ligne d'import. site_target_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey( "site_targets.id", ondelete="CASCADE", name="fk_keyword_page_map_site_target_id" ), nullable=False, ) keyword_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("keywords.id", ondelete="CASCADE", name="fk_keyword_page_map_keyword_id"), nullable=False, ) # Produit par `app.core.urls.normalize_url` — la clé de résolution, jamais NULL : une # attache sans URL serait une page fantôme. url_normalized: Mapped[str] = mapped_column(Text, nullable=False) # Sans FK tant que `pages` n'existe pas (module Crawl). page_id: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True), nullable=True) source: Mapped[str] = mapped_column(Text, nullable=False) position: Mapped[Decimal | None] = mapped_column(Numeric, nullable=True) seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) __table_args__ = ( CheckConstraint( "source IN ('import_url', 'manual', 'gsc')", name="ck_keyword_page_map_source" ), # Observations multiples représentables nativement (cannibalisation GSC = zéro # migration au module 3). UniqueConstraint( "site_target_id", "keyword_id", "url_normalized", "source", name="uq_keyword_page_map_target_keyword_url_source", ), # **Une** attache de planification par (cible, keyword) — décision n°6. Index( "uq_keyword_page_map_manual", "site_target_id", "keyword_id", unique=True, postgresql_where=text("source = 'manual'"), ), # Résolution ensembliste de fin de crawl (§3.9) : ne référence que les maps encore # en attente, la table entière n'a pas à être parcourue. Index( "ix_keyword_page_map_pending", "site_target_id", "url_normalized", postgresql_where=text("page_id IS NULL"), ), # Prédicat d'appartenance au réservoir et statut d'attache par mot-clé (§6). Index("ix_keyword_page_map_keyword_id", "keyword_id"), ) class OrgSettings(Base): """Réglages d'organisation (§3.10). En v1, seule la source préférée existe. `preferred_source` **ordonne** les entrées d'une liste — la préférée en tête — et n'en masque aucune (spec 2026-09-01 §3.4). Depuis la suppression de la hiérarchie de sources (décision A1), plus rien n'élit une valeur : le mot-clé porte les observations de toutes ses sources côte à côte, cette colonne dit seulement laquelle se lit en premier. Validée au PATCH contre le registre d'adaptateurs (422 si inconnue) — pas de contrainte en base, le registre est du code et la liste des adaptateurs bouge à chaque tranche. """ __tablename__ = "org_settings" __mapper_args__ = {"eager_defaults": True} # Id Better Auth opaque : la PK est l'org elle-même, une ligne par org. org_id: Mapped[str] = mapped_column(Text, primary_key=True) preferred_source: Mapped[str | None] = mapped_column(Text, nullable=True) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) # `onupdate` est **ORM-only** : Postgres n'a pas d'ON UPDATE, et un UPDATE écrit en SQL # brut (ou un `update()` Core) ne rafraîchira pas cette colonne. Le PATCH du T3 doit # passer par l'ORM — sinon il faudra un trigger, pas un `server_onupdate` (qui n'est # qu'une indication pour SQLAlchemy et ne crée rien en base). updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now() ) class Import(Base): """Persistance de domaine d'un import — **jamais dans `jobs.payload`** (§3.11). `status` recopie celui du job porteur sur tous les chemins (handler au démarrage et à la fin, callback `on_fail` du socle pour les échecs, y compris le `failed` définitif du sweep des zombies) : plus aucun import `queued` à perpétuité, et le listing reste sans join sur la file. """ __tablename__ = "imports" __mapper_args__ = {"eager_defaults": True} id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, server_default=text("uuidv7()") ) org_id: Mapped[str] = mapped_column(Text, nullable=False, index=True) # Pas d'index dédié : le btree de `ix_imports_target_sha256 (site_target_id, # file_sha256)` est préfixé par site_target_id et sert le listing de l'onglet Imports. site_target_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("site_targets.id", ondelete="CASCADE", name="fk_imports_site_target_id"), nullable=False, ) source: Mapped[str] = mapped_column(Text, nullable=False) kind: Mapped[str] = mapped_column(Text, nullable=False) filename: Mapped[str | None] = mapped_column(Text, nullable=True) # **Pas d'UNIQUE dur** : le re-import légitime après correction d'adaptateur reste # possible ; le doublon n'est qu'un warning soft au wizard (§1.3). file_sha256: Mapped[str | None] = mapped_column(Text, nullable=True) detected_format: Mapped[str | None] = mapped_column(Text, nullable=True) # Le pays de la cible, c'est-à-dire celui qui a été RETENU — pas celui que le fichier # annonçait, puisqu'un même fichier peut en annoncer onze (§4.5). Champ d'audit : il dit # ce qui a été gardé, le rapport dit ce qui a été écarté et pourquoi. detected_country: Mapped[str | None] = mapped_column(Text, nullable=True) fetched_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) keyword_list_id: Mapped[uuid.UUID | None] = mapped_column( UUID(as_uuid=True), ForeignKey("keyword_lists.id", ondelete="SET NULL", name="fk_imports_keyword_list_id"), nullable=True, ) # Pas d'ON DELETE : `jobs` n'est jamais purgée en v1 (socle §1) et un import sans job # serait un rapport sans exécution — l'erreur doit être bruyante. job_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("jobs.id", name="fk_imports_job_id"), nullable=False, ) status: Mapped[str] = mapped_column(Text, nullable=False) # Colonne absente du tableau de la spec mais exigée par le contrat de recopie juste # au-dessus (« recopie `failed` + `error` dans la ligne `imports` », §3.11) : elle est # créée ici parce qu'aucune tranche suivante ne porte de migration (§9). error: Mapped[str | None] = mapped_column(Text, nullable=True) report: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) __table_args__ = ( CheckConstraint("kind IN ('csv', 'api')", name="ck_imports_kind"), CheckConstraint( "status IN ('queued', 'running', 'done', 'failed')", name="ck_imports_status" ), # Warning de doublon au wizard : « ce fichier a déjà été importé sur cette cible ». Index("ix_imports_target_sha256", "site_target_id", "file_sha256"), )