"""Accès SQLite : migrations versionnées, écriture des textes, requêtes de veille. Une seule règle : rien n'entre en base sans passer par un modèle pydantic validé. Les fonctions de ce module reçoivent des `Texte`, jamais des dictionnaires bruts venus d'un parseur. """ from __future__ import annotations import json import sqlite3 from collections.abc import Iterable, Iterator from contextlib import contextmanager from pathlib import Path from typing import Any from pipeline import chemins from pipeline.journal import logger from pipeline.modeles import DecisionCC, Evenement, Insight, Source, Texte log = logger("db") # ───────────────────────────────────────────────────────────────────────────── # Connexion et migrations # ───────────────────────────────────────────────────────────────────────────── def connecter(base: Path | None = None, *, lecture_seule: bool = False) -> sqlite3.Connection: """Ouvre la base avec les réglages attendus par le reste du pipeline.""" base = base or chemins.BASE_SQLITE base.parent.mkdir(parents=True, exist_ok=True) if lecture_seule: cx = sqlite3.connect(f"file:{base}?mode=ro", uri=True) else: cx = sqlite3.connect(base) cx.row_factory = sqlite3.Row cx.execute("PRAGMA foreign_keys = ON") if not lecture_seule: cx.execute("PRAGMA journal_mode = WAL") cx.execute("PRAGMA synchronous = NORMAL") return cx @contextmanager def transaction(cx: sqlite3.Connection) -> Iterator[sqlite3.Connection]: """Valide à la sortie, annule intégralement à la moindre exception.""" try: yield cx except Exception: cx.rollback() raise else: cx.commit() def _version_schema(cx: sqlite3.Connection) -> int: return int(cx.execute("PRAGMA user_version").fetchone()[0]) def migrer(cx: sqlite3.Connection, dossier: Path | None = None) -> list[str]: """Applique les migrations non encore jouées, dans l'ordre des noms. Retourne la liste des migrations appliquées lors de cet appel. """ dossier = dossier or chemins.MIGRATIONS fichiers = sorted(dossier.glob("*.sql")) version = _version_schema(cx) appliquees: list[str] = [] for fichier in fichiers: numero = int(fichier.name.split("_", 1)[0]) if numero <= version: continue log.info("migration", fichier=fichier.name, numero=numero) cx.executescript(fichier.read_text(encoding="utf-8")) cx.execute(f"PRAGMA user_version = {numero}") cx.commit() appliquees.append(fichier.name) return appliquees def initialiser(base: Path | None = None, *, repartir_de_zero: bool = False) -> sqlite3.Connection: """Ouvre la base et s'assure que le schéma est à jour.""" base = base or chemins.BASE_SQLITE if repartir_de_zero and base.exists(): for suffixe in ("", "-wal", "-shm"): Path(str(base) + suffixe).unlink(missing_ok=True) log.info("base supprimée avant réinitialisation", base=str(base)) cx = connecter(base) migrer(cx) return cx # ───────────────────────────────────────────────────────────────────────────── # Écriture # ───────────────────────────────────────────────────────────────────────────── _CHAMPS_TEXTE = ( "id", "numero_officiel", "type", "titre_court", "titre_officiel", "statut", "statut_date", "date_depot", "date_adoption", "date_promulgation", "date_entree_vigueur", "prochaine_echeance", "prochaine_echeance_label", "themes", "impacts", "regime_libertes", "regime_libertes_note", "guadeloupe_pertinence", "guadeloupe_note", "resume", "points_cles", "confiance", "source_seed", "a_verifier", "motif_verification", ) def enregistrer_texte(cx: sqlite3.Connection, texte: Texte) -> None: """Insère ou met à jour un texte, avec ses sources, événements et décisions.""" ligne = texte.en_ligne_sql() colonnes = ", ".join(_CHAMPS_TEXTE) valeurs = ", ".join(f":{c}" for c in _CHAMPS_TEXTE) maj = ", ".join(f"{c} = excluded.{c}" for c in _CHAMPS_TEXTE if c != "id") cx.execute( f"INSERT INTO textes ({colonnes}) VALUES ({valeurs}) " f"ON CONFLICT (id) DO UPDATE SET {maj}, maj_le = datetime('now')", ligne, ) for source in texte.sources: enregistrer_source(cx, texte.id, source) for evenement in texte.evenements: enregistrer_evenement(cx, texte.id, evenement) for decision in texte.decisions_cc: enregistrer_decision_cc(cx, decision, texte_id=texte.id) def enregistrer_source(cx: sqlite3.Connection, texte_id: str, source: Source) -> None: cx.execute( """ INSERT INTO sources (texte_id, url, titre, editeur, date_publication, tier, extrait_verbatim, contexte, confiance, marqueur, fichier_origine) VALUES (:texte_id, :url, :titre, :editeur, :date_publication, :tier, :extrait_verbatim, :contexte, :confiance, :marqueur, :fichier_origine) ON CONFLICT (texte_id, url, COALESCE(marqueur, '')) DO UPDATE SET titre = COALESCE(excluded.titre, titre), editeur = COALESCE(excluded.editeur, editeur), date_publication = COALESCE(excluded.date_publication, date_publication), tier = excluded.tier, extrait_verbatim = COALESCE(excluded.extrait_verbatim, extrait_verbatim), contexte = COALESCE(excluded.contexte, contexte), confiance = COALESCE(excluded.confiance, confiance) """, { "texte_id": texte_id, "url": source.url, "titre": source.titre, "editeur": source.editeur, "date_publication": source.date_publication.isoformat() if source.date_publication else None, "tier": str(source.tier), "extrait_verbatim": source.extrait_verbatim, "contexte": source.contexte, "confiance": str(source.confiance) if source.confiance else None, "marqueur": source.marqueur, "fichier_origine": source.fichier_origine, }, ) def enregistrer_evenement(cx: sqlite3.Connection, texte_id: str, evenement: Evenement) -> None: cx.execute( """ INSERT INTO evenements (texte_id, date_evenement, type_etape, description, source_url, previsionnel) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT (texte_id, date_evenement, type_etape, description) DO UPDATE SET source_url = COALESCE(excluded.source_url, source_url), previsionnel = excluded.previsionnel """, ( texte_id, evenement.date_evenement.isoformat(), str(evenement.type_etape), evenement.description, evenement.source_url, int(evenement.previsionnel), ), ) def enregistrer_decision_cc( cx: sqlite3.Connection, decision: DecisionCC, *, texte_id: str | None = None ) -> None: cx.execute( """ INSERT INTO decisions_cc (texte_id, numero_affaire, date_saisine, date_decision, resultat, saisissants, resume, url, date_decision_attendue) VALUES (:texte_id, :numero_affaire, :date_saisine, :date_decision, :resultat, :saisissants, :resume, :url, :date_decision_attendue) ON CONFLICT (numero_affaire) DO UPDATE SET texte_id = COALESCE(excluded.texte_id, texte_id), date_saisine = COALESCE(excluded.date_saisine, date_saisine), date_decision = COALESCE(excluded.date_decision, date_decision), resultat = COALESCE(excluded.resultat, resultat), saisissants = COALESCE(excluded.saisissants, saisissants), resume = COALESCE(excluded.resume, resume), url = COALESCE(excluded.url, url), date_decision_attendue = COALESCE(excluded.date_decision_attendue, date_decision_attendue), maj_le = datetime('now') """, { "texte_id": texte_id, "numero_affaire": decision.numero_affaire, "date_saisine": decision.date_saisine.isoformat() if decision.date_saisine else None, "date_decision": decision.date_decision.isoformat() if decision.date_decision else None, "resultat": str(decision.resultat) if decision.resultat else None, "saisissants": decision.saisissants, "resume": decision.resume, "url": decision.url, "date_decision_attendue": decision.date_decision_attendue.isoformat() if decision.date_decision_attendue else None, }, ) def enregistrer_insight(cx: sqlite3.Connection, insight: Insight) -> None: cx.execute( """ INSERT INTO insights (numero, titre, corps, implications, confiance, derive_de, fichier_origine) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT (numero) DO UPDATE SET titre = excluded.titre, corps = excluded.corps, implications = excluded.implications, confiance = excluded.confiance, derive_de = excluded.derive_de, fichier_origine = excluded.fichier_origine """, ( insight.numero, insight.titre, insight.corps, insight.implications, insight.confiance, json.dumps(insight.derive_de, ensure_ascii=False), insight.fichier_origine, ), ) def enregistrer_textes(cx: sqlite3.Connection, textes: Iterable[Texte]) -> int: """Écrit un lot de textes dans une transaction unique.""" total = 0 with transaction(cx): for texte in textes: enregistrer_texte(cx, texte) total += 1 return total # ───────────────────────────────────────────────────────────────────────────── # Journal des exécutions # ───────────────────────────────────────────────────────────────────────────── def ouvrir_run(cx: sqlite3.Connection, mode: str) -> int: curseur = cx.execute("INSERT INTO veille_log (mode) VALUES (?)", (mode,)) cx.commit() return int(curseur.lastrowid) def cloturer_run( cx: sqlite3.Connection, run_id: int, *, duree_s: float, ajouts: int = 0, modifications: int = 0, echecs: int = 0, alertes: list[str] | None = None, rapport: str | None = None, ) -> None: cx.execute( """ UPDATE veille_log SET duree_s = ?, ajouts = ?, modifications = ?, echecs = ?, alertes = ?, rapport = ? WHERE id = ? """, ( duree_s, ajouts, modifications, echecs, json.dumps(alertes or [], ensure_ascii=False), rapport, run_id, ), ) cx.commit() def enregistrer_changement( cx: sqlite3.Connection, run_id: int, *, texte_id: str | None, nature: str, champ: str | None = None, ancienne_valeur: str | None = None, nouvelle_valeur: str | None = None, description: str | None = None, ) -> None: cx.execute( """ INSERT INTO veille_changements (run_id, texte_id, nature, champ, ancienne_valeur, nouvelle_valeur, description) VALUES (?, ?, ?, ?, ?, ?, ?) """, (run_id, texte_id, nature, champ, ancienne_valeur, nouvelle_valeur, description), ) # ───────────────────────────────────────────────────────────────────────────── # Lecture # ───────────────────────────────────────────────────────────────────────────── def statistiques(cx: sqlite3.Connection) -> dict[str, Any]: """Compteurs affichés en fin de seed et sur le tableau de bord.""" def un(requete: str) -> int: return int(cx.execute(requete).fetchone()[0]) return { "textes": un("SELECT COUNT(*) FROM textes"), "sources": un("SELECT COUNT(*) FROM sources"), "evenements": un("SELECT COUNT(*) FROM evenements"), "decisions_cc": un("SELECT COUNT(*) FROM decisions_cc"), "insights": un("SELECT COUNT(*) FROM insights"), "textes_sans_source": un( "SELECT COUNT(*) FROM textes t " "WHERE NOT EXISTS (SELECT 1 FROM sources s WHERE s.texte_id = t.id)" ), "a_verifier": un("SELECT COUNT(*) FROM textes WHERE a_verifier = 1"), "par_statut": { r["statut"]: r["n"] for r in cx.execute( "SELECT statut, COUNT(*) AS n FROM textes GROUP BY statut ORDER BY n DESC" ) }, "par_pertinence_gpe": { (r["guadeloupe_pertinence"] or "non_cotee"): r["n"] for r in cx.execute( "SELECT guadeloupe_pertinence, COUNT(*) AS n FROM textes " "GROUP BY guadeloupe_pertinence" ) }, "devant_cc": un("SELECT COUNT(*) FROM v_textes WHERE devant_cc = 1"), } def main() -> None: """`make db-init` — crée la base si besoin et applique les migrations.""" import argparse from pipeline.journal import configurer analyseur = argparse.ArgumentParser(description="Initialise ou migre la base SQLite.") analyseur.add_argument( "--repartir-de-zero", action="store_true", help="supprime la base existante avant de recréer le schéma", ) arguments = analyseur.parse_args() configurer() cx = initialiser(repartir_de_zero=arguments.repartir_de_zero) log.info( "base prête", chemin=str(chemins.BASE_SQLITE), version_schema=_version_schema(cx), **statistiques(cx) | {"par_statut": None, "par_pertinence_gpe": None}, ) if __name__ == "__main__": main()