veye-lalwa/pipeline/modeles.py

"""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 RegimeLibertes(StrEnum):
    """Cotation globale d'un texte par l'annexe du rapport (`sec10.md`)."""

    DROITS_PLUS = "droits_plus"        # création ou extension de droits
    CONTROLE_PLUS = "controle_plus"    # restriction, ou extension des contrôles
    MIXTE = "mixte"
    NEUTRE = "neutre"


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)

    regime_libertes: RegimeLibertes | None = None
    regime_libertes_note: str | None = None

    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,
            ),
            "regime_libertes": str(self.regime_libertes) if self.regime_libertes else None,
            "regime_libertes_note": self.regime_libertes_note,
            "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