Files
veye-lalwa/pipeline/modeles.py
T
Cyber MawonajandClaude Opus 5 31d3935bce Phase 1 : socle Python, schéma SQLite + FTS5, modèles pydantic
- pyproject/uv sur Python 3.12, dépendances épinglées
- migrations versionnées (PRAGMA user_version) ; 001_initial pose textes,
  sources, evenements, decisions_cc, insights, veille_log, veille_changements
- index FTS5 à contenu externe, tokenizer unicode61 remove_diacritics 2 :
  « chlordecone » retrouve « chlordécone », highlight() disponible
- vue v_textes : devant_cc combine le statut et les affaires CC en instance,
  ce qui réconcilie le §7 (Riposte reste adoptee_non_promulguee) et le §8.7
  (la facette saisie_cc doit remonter ≥ 7 textes)
- modèles pydantic refusant les incohérences de statut : promulguée sans
  numéro, navette avec numéro officiel, slug malformé
- Makefile, .env.example, journalisation structlog

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-07-25 17:49:11 -04:00

381 lines
12 KiB
Python

"""Modèles de domaine validés par pydantic.
Le vocabulaire est celui du corpus de recherche, en français : un `Texte` a un
`statut`, une `pertinence` guadeloupéenne et des `impacts` sur trois publics.
Les énumérations reprennent exactement les valeurs du §4 du prompt de mission
et les contraintes CHECK de la migration 001.
"""
from __future__ import annotations
import json
import re
from datetime import date
from enum import StrEnum
from typing import Any
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
# ─────────────────────────────────────────────────────────────────────────────
# Énumérations
# ─────────────────────────────────────────────────────────────────────────────
class TypeTexte(StrEnum):
LOI = "loi"
LOI_ORGANIQUE = "loi_organique"
PJL = "pjl"
PPL = "ppl"
ORDONNANCE = "ordonnance"
DECISION_CC = "decision_cc"
DECRET = "decret"
ACCORD_INTERNATIONAL = "accord_international"
class Statut(StrEnum):
"""Règle d'or du rapport : le statut prime sur l'intitulé."""
PROMULGUEE = "promulguee"
ADOPTEE_NON_PROMULGUEE = "adoptee_non_promulguee"
SAISIE_CC = "saisie_cc"
VALIDEE_CC = "validee_cc"
CENSUREE_PARTIELLEMENT = "censuree_partiellement"
NAVETTE = "navette"
DEPOSEE_NON_EXAMINEE = "deposee_non_examinee"
ANNONCEE = "annoncee"
REJETEE = "rejetee"
class Theme(StrEnum):
JUSTICE = "justice"
SECURITE = "securite"
NUMERIQUE = "numerique"
SOCIAL = "social"
FISCAL = "fiscal"
ENVIRONNEMENT = "environnement"
AGRICULTURE = "agriculture"
SANTE = "sante"
MEMOIRE_PATRIMOINE = "memoire_patrimoine"
OUTRE_MER = "outre_mer"
ECONOMIE = "economie"
MIGRATION = "migration"
INSTITUTIONS = "institutions"
class Public(StrEnum):
"""Les trois publics dont la mission suit les libertés."""
INDIVIDUS = "individus"
ASSOCIATIONS = "associations"
ENTREPRISES = "entreprises"
class SensImpact(StrEnum):
POSITIF = "positif"
NEGATIF = "negatif"
MIXTE = "mixte"
NEUTRE = "neutre"
class Pertinence(StrEnum):
FORTE = "forte"
MOYENNE = "moyenne"
FAIBLE = "faible"
class Confiance(StrEnum):
HIGH = "high"
MEDIUM = "medium"
LOW = "low"
class Tier(StrEnum):
"""T1 = source primaire institutionnelle ; T2 = presse et analyses."""
T1 = "T1"
T2 = "T2"
class TypeEtape(StrEnum):
DEPOT = "depot"
ADOPTION_1RE_LECTURE = "adoption_1re_lecture"
ADOPTION_DEFINITIVE = "adoption_definitive"
COMMISSION_MIXTE_PARITAIRE = "commission_mixte_paritaire"
TRANSMISSION = "transmission"
SAISINE_CC = "saisine_cc"
DECISION_CC = "decision_cc"
PROMULGATION = "promulgation"
PUBLICATION_JO = "publication_jo"
ENTREE_VIGUEUR = "entree_vigueur"
REJET = "rejet"
ANNONCE = "annonce"
AUTRE = "autre"
class ResultatCC(StrEnum):
CONFORME = "conforme"
CONFORME_AVEC_RESERVES = "conforme_avec_reserves"
NON_CONFORMITE_PARTIELLE = "non_conformite_partielle"
NON_CONFORMITE_TOTALE = "non_conformite_totale"
EN_INSTANCE = "en_instance"
# ─────────────────────────────────────────────────────────────────────────────
# Modèles
# ─────────────────────────────────────────────────────────────────────────────
MOTIF_SLUG = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
MOTIF_NUMERO = re.compile(r"^\d{4}-\d{1,4}$")
MOTIF_AFFAIRE_CC = re.compile(r"^\d{4}-\d{1,4} DC$")
class Impact(BaseModel):
"""Effet d'un texte sur les libertés d'un public donné."""
model_config = ConfigDict(extra="forbid")
sens: SensImpact
note: str = Field(default="", max_length=400)
class Source(BaseModel):
"""Citation vérifiable : sans URL, un fait n'entre pas en base."""
model_config = ConfigDict(extra="forbid")
url: str
titre: str | None = None
editeur: str | None = None
date_publication: date | None = None
tier: Tier = Tier.T2
extrait_verbatim: str | None = None
contexte: str | None = None
confiance: Confiance | None = None
marqueur: str | None = None
fichier_origine: str | None = None
@field_validator("url")
@classmethod
def _url_plausible(cls, v: str) -> str:
v = v.strip().rstrip(".,;)")
if not v.startswith(("http://", "https://")):
raise ValueError(f"URL non exploitable : {v!r}")
return v
@model_validator(mode="after")
def _deduire_tier(self) -> Source:
"""Classe automatiquement en T1 les hôtes institutionnels connus."""
if self.tier == Tier.T2 and est_source_primaire(self.url):
object.__setattr__(self, "tier", Tier.T1)
return self
HOTES_PRIMAIRES = (
"legifrance.gouv.fr",
"assemblee-nationale.fr",
"senat.fr",
"conseil-constitutionnel.fr",
"conseil-etat.fr",
"vie-publique.fr",
"journal-officiel.gouv.fr",
"elysee.fr",
"gouvernement.fr",
"defenseurdesdroits.fr",
"cnil.fr",
"insee.fr",
"data.gouv.fr",
)
def est_source_primaire(url: str) -> bool:
"""Vrai pour les hôtes institutionnels et les sous-domaines en `.gouv.fr`."""
hote = url.split("//", 1)[-1].split("/", 1)[0].lower().removeprefix("www.")
if any(hote == h or hote.endswith("." + h) for h in HOTES_PRIMAIRES):
return True
return hote.endswith(".gouv.fr")
class Evenement(BaseModel):
"""Une étape datée du parcours d'un texte."""
model_config = ConfigDict(extra="forbid")
date_evenement: date
type_etape: TypeEtape
description: str
source_url: str | None = None
previsionnel: bool = False
class DecisionCC(BaseModel):
"""Affaire portée devant le Conseil constitutionnel."""
model_config = ConfigDict(extra="forbid")
numero_affaire: str
date_saisine: date | None = None
date_decision: date | None = None
resultat: ResultatCC | None = None
saisissants: str | None = None
resume: str | None = None
url: str | None = None
date_decision_attendue: date | None = None
@field_validator("numero_affaire")
@classmethod
def _format_affaire(cls, v: str) -> str:
v = " ".join(v.split()).upper().replace("DC.", "DC")
if not MOTIF_AFFAIRE_CC.match(v):
raise ValueError(f"Numéro d'affaire attendu au format « 2026-915 DC », reçu {v!r}")
return v
@model_validator(mode="after")
def _coherence_instance(self) -> DecisionCC:
if self.date_decision is None and self.resultat is None:
object.__setattr__(self, "resultat", ResultatCC.EN_INSTANCE)
return self
class Texte(BaseModel):
"""Un texte législatif suivi par la veille."""
model_config = ConfigDict(extra="forbid")
id: str
numero_officiel: str | None = None
type: TypeTexte
titre_court: str
titre_officiel: str | None = None
statut: Statut
statut_date: date | None = None
date_depot: date | None = None
date_adoption: date | None = None
date_promulgation: date | None = None
date_entree_vigueur: date | None = None
prochaine_echeance: date | None = None
prochaine_echeance_label: str | None = None
themes: list[Theme] = Field(default_factory=list)
impacts: dict[Public, Impact] = Field(default_factory=dict)
guadeloupe_pertinence: Pertinence | None = None
guadeloupe_note: str | None = None
resume: str | None = None
points_cles: list[str] = Field(default_factory=list)
confiance: Confiance = Confiance.MEDIUM
source_seed: str | None = None
a_verifier: bool = False
motif_verification: str | None = None
sources: list[Source] = Field(default_factory=list)
evenements: list[Evenement] = Field(default_factory=list)
decisions_cc: list[DecisionCC] = Field(default_factory=list)
@field_validator("id")
@classmethod
def _slug_valide(cls, v: str) -> str:
if not MOTIF_SLUG.match(v):
raise ValueError(f"Identifiant non conforme au format slug : {v!r}")
return v
@field_validator("numero_officiel")
@classmethod
def _numero_valide(cls, v: str | None) -> str | None:
if v is None:
return None
v = v.strip()
if not MOTIF_NUMERO.match(v):
raise ValueError(f"Numéro officiel attendu au format « 2026-491 », reçu {v!r}")
return v
@field_validator("themes")
@classmethod
def _themes_uniques(cls, v: list[Theme]) -> list[Theme]:
vus: list[Theme] = []
for t in v:
if t not in vus:
vus.append(t)
return vus
@model_validator(mode="after")
def _coherence_statut(self) -> Texte:
"""Un texte promulgué porte forcément un numéro et une date de promulgation."""
if self.statut is Statut.PROMULGUEE:
if not self.numero_officiel:
raise ValueError(f"{self.id} : statut « promulguee » sans numéro officiel")
if not self.date_promulgation:
raise ValueError(f"{self.id} : statut « promulguee » sans date de promulgation")
elif self.statut in (
Statut.ADOPTEE_NON_PROMULGUEE,
Statut.NAVETTE,
Statut.DEPOSEE_NON_EXAMINEE,
Statut.ANNONCEE,
) and self.numero_officiel:
raise ValueError(
f"{self.id} : un texte au statut « {self.statut} » ne peut pas porter "
f"le numéro officiel {self.numero_officiel!r} — un numéro n'est attribué "
"qu'à la promulgation"
)
if self.statut_date is None:
object.__setattr__(
self,
"statut_date",
self.date_promulgation or self.date_adoption or self.date_depot,
)
return self
# ── Sérialisation vers SQLite ────────────────────────────────────────────
def en_ligne_sql(self) -> dict[str, Any]:
"""Aplatit le modèle en un dictionnaire directement insérable."""
def iso(d: date | None) -> str | None:
return d.isoformat() if d else None
return {
"id": self.id,
"numero_officiel": self.numero_officiel,
"type": str(self.type),
"titre_court": self.titre_court,
"titre_officiel": self.titre_officiel,
"statut": str(self.statut),
"statut_date": iso(self.statut_date),
"date_depot": iso(self.date_depot),
"date_adoption": iso(self.date_adoption),
"date_promulgation": iso(self.date_promulgation),
"date_entree_vigueur": iso(self.date_entree_vigueur),
"prochaine_echeance": iso(self.prochaine_echeance),
"prochaine_echeance_label": self.prochaine_echeance_label,
"themes": json.dumps([str(t) for t in self.themes], ensure_ascii=False),
"impacts": json.dumps(
{str(p): {"sens": str(i.sens), "note": i.note} for p, i in self.impacts.items()},
ensure_ascii=False,
),
"guadeloupe_pertinence": (
str(self.guadeloupe_pertinence) if self.guadeloupe_pertinence else None
),
"guadeloupe_note": self.guadeloupe_note,
"resume": self.resume,
"points_cles": json.dumps(self.points_cles, ensure_ascii=False),
"confiance": str(self.confiance),
"source_seed": self.source_seed,
"a_verifier": int(self.a_verifier),
"motif_verification": self.motif_verification,
}
class Insight(BaseModel):
"""Analyse transversale exposée dans l'encart « point de veille »."""
model_config = ConfigDict(extra="forbid")
numero: int
titre: str
corps: str
implications: str | None = None
confiance: str = "medium"
derive_de: list[str] = Field(default_factory=list)
fichier_origine: str | None = None