Files

218 lines
7.4 KiB
Python
Raw Permalink Normal View History

"""Lecture des blocs sourcés des rapports de dimension (`research/dim01` → `dim12`).
Le cahier des charges postule un format unique
`Claim / Source / URL / Date / Excerpt / Confidence`. Le corpus réel en compte
**quatre dialectes**, produits par des agents de recherche différents :
- **A — claim plat** (dim01 à dim04, dim07) ::
Claim: … [^24^]
Source: Vie-publique.fr
URL: https://…
Confidence: high
- **B — claim numéroté en gras** (dim05) :: ``**Claim 3 :** … [^3^]``
- **C — claim en liste** (dim10, portions de dim06 et dim08) ::
**Claim 1.2** — …
- **Source** : Conseil d'État
- **URL** : https://…
- **D — prose et notes** (dim06, dim08, dim09, dim11, dim12) : aucun bloc, les
sources vivent dans une liste de notes en fin de fichier (voir
`notes_bas_page`).
Un parseur qui n'aurait vu que le dialecte A perdrait dim05 à dim12, c'est-à-dire
les dimensions associations, entreprises, migration et calendrier.
"""
from __future__ import annotations
import re
from dataclasses import dataclass, field
from datetime import date
from pipeline.journal import logger
from pipeline.modeles import Confiance, Source
from pipeline.parseurs.dates_fr import lire_date
from pipeline.parseurs.markdown import MOTIF_MARQUEUR, nettoyer
log = logger("parseur.claims")
ANNEE_CORPUS = 2026
# Une ligne de claim, tous dialectes : « Claim: », « **Claim 3 :** »,
# « **Claim 1.2** — », « - **Claim**: », et la forme de la validation de
# phase 5 qui glisse une parenthèse entre le numéro et le deux-points :
# « **Claim 1.1 (décisif, source primaire) : … » — sans quoi ce fichier ne
# rend que 2 blocs sur 24.
_LIGNE_CLAIM = re.compile(
r"^\s*(?:[-*]\s*)?\*{0,2}\s*Claim\s*([\d.]*)\s*(?:\([^)]*\))?\s*\*{0,2}\s*[:—–-]\s*(.*)$",
re.IGNORECASE,
)
# Un champ du bloc, avec ou sans puce, avec ou sans gras, avec le deux-points
# à l'intérieur ou à l'extérieur des astérisques.
_LIGNE_CHAMP = re.compile(
r"^\s*(?:[-*]\s*)?\*{0,2}\s*"
r"(Source|URL|Date|Excerpt|Extrait|Contexte|Context|Confidence|Confiance)"
r"\s*\*{0,2}\s*:\s*\*{0,2}\s*(.*?)\s*\*{0,2}\s*$",
re.IGNORECASE,
)
_CHAMPS = {
"source": "source",
"url": "url",
"date": "date",
"excerpt": "excerpt",
"extrait": "excerpt",
"context": "contexte",
"contexte": "contexte",
"confidence": "confiance",
"confiance": "confiance",
}
# Formes relevées dans le corpus, du plus fort au plus faible.
_CONFIANCES: tuple[tuple[str, Confiance], ...] = (
("tres elevee", Confiance.HIGH),
("très élevée", Confiance.HIGH),
("medium-high", Confiance.MEDIUM),
("moyenne-haute", Confiance.MEDIUM),
("high", Confiance.HIGH),
("haute", Confiance.HIGH),
("elevee", Confiance.HIGH),
("élevée", Confiance.HIGH),
("medium", Confiance.MEDIUM),
("moyenne", Confiance.MEDIUM),
("low", Confiance.LOW),
("faible", Confiance.LOW),
)
_URL = re.compile(r"https?://[^\s;,)\]]+")
@dataclass(slots=True)
class BlocClaim:
"""Une affirmation sourcée extraite d'un rapport de dimension."""
dimension: str # « dim07 »
affirmation: str
marqueurs: list[str] = field(default_factory=list) # numéros locaux : ["24"]
editeur: str | None = None
urls: list[str] = field(default_factory=list)
date_publication: date | None = None
extrait: str | None = None
contexte: str | None = None
confiance: Confiance = Confiance.MEDIUM
ligne: int = 0
def en_sources(self) -> list[Source]:
"""Convertit le bloc en sources exploitables — une par URL citée."""
sources: list[Source] = []
for url in self.urls:
try:
sources.append(
Source(
url=url,
titre=self.affirmation[:280] or None,
editeur=self.editeur,
date_publication=self.date_publication,
extrait_verbatim=self.extrait,
contexte=self.contexte,
confiance=self.confiance,
marqueur=f"{self.dimension}-{self.marqueurs[0]}"
if self.marqueurs
else self.dimension,
fichier_origine=f"research/lois-2026_{self.dimension}.md",
)
)
except ValueError as erreur: # URL inexploitable : on ignore, sans bruit
log.debug("URL rejetée", url=url, motif=str(erreur))
return sources
def parser(markdown: str, dimension: str) -> list[BlocClaim]:
"""Extrait tous les blocs sourcés d'un rapport de dimension."""
blocs: list[BlocClaim] = []
courant: BlocClaim | None = None
for numero, ligne in enumerate(markdown.splitlines(), start=1):
if debut := _LIGNE_CLAIM.match(ligne):
courant = _ouvrir_bloc(debut.group(2), dimension, numero)
blocs.append(courant)
continue
if courant is None:
continue
if champ := _LIGNE_CHAMP.match(ligne):
_remplir(courant, _CHAMPS[champ.group(1).lower()], champ.group(2))
continue
# Un titre de section referme le bloc en cours ; le reste (prose,
# lignes vides) est ignoré sans le refermer, car certains blocs sont
# entrecoupés de commentaires du rédacteur.
if ligne.lstrip().startswith("#"):
courant = None
complets = [b for b in blocs if b.urls]
log.debug(
"blocs analysés",
dimension=dimension,
blocs=len(blocs),
avec_url=len(complets),
)
return blocs
def _ouvrir_bloc(affirmation: str, dimension: str, ligne: int) -> BlocClaim:
marqueurs = MOTIF_MARQUEUR.findall(affirmation)
return BlocClaim(
dimension=dimension,
affirmation=nettoyer(affirmation),
marqueurs=[m for m in marqueurs if m.isdigit() or "." in m],
ligne=ligne,
)
def _remplir(bloc: BlocClaim, champ: str, valeur: str) -> None:
valeur = valeur.strip()
if not valeur:
return
match champ:
case "source":
bloc.editeur = nettoyer(valeur)[:200] or None
case "url":
# Une ligne peut citer plusieurs URLs séparées par « ; ».
bloc.urls.extend(u for u in _URL.findall(valeur) if u not in bloc.urls)
case "date":
if lue := lire_date(valeur, annee_defaut=ANNEE_CORPUS):
bloc.date_publication = lue
case "excerpt":
bloc.extrait = _nettoyer_extrait(valeur)
case "contexte":
bloc.contexte = nettoyer(valeur)[:1000] or None
case "confiance":
bloc.confiance = lire_confiance(valeur)
def _nettoyer_extrait(valeur: str) -> str | None:
"""Conserve l'extrait verbatim, sans les guillemets ni l'italique du corpus."""
extrait = valeur.strip()
for ouvrant, fermant in (("«", "»"), ('"', '"'), ("“", "”")):
extrait = extrait.strip().removeprefix(ouvrant).removesuffix(fermant).strip()
extrait = extrait.replace("*", "").strip()
extrait = MOTIF_MARQUEUR.sub("", extrait)
return " ".join(extrait.split())[:2000] or None
def lire_confiance(valeur: str) -> Confiance:
"""Traduit les huit formes de confiance rencontrées dans le corpus."""
plat = valeur.lower().replace("*", "").strip()
for forme, confiance in _CONFIANCES:
if plat.startswith(forme):
return confiance
return Confiance.MEDIUM