131 lines
4.6 KiB
Python
131 lines
4.6 KiB
Python
 
|
"""Collecteur du Sénat — liste chronologique des lois promulguées.
|
||
|
|
|
||
|
|
C'est le repli principal lorsque l'API Légifrance est indisponible, et un
|
||
|
|
contrôle croisé lorsqu'elle fonctionne. La page rend, pour chaque loi
|
||
|
|
promulguée de la session, son numéro, sa date, son intitulé complet et le lien
|
||
|
|
vers son dossier législatif.
|
||
|
|
|
||
|
|
L'adresse indiquée au §5.1 du cahier des charges — `senat.fr/lois/index.html` —
|
||
|
|
renvoie une erreur 404 depuis la refonte du site. L'équivalent fonctionnel est
|
||
|
|
`senat.fr/dossiers-legislatifs/lois-promulguees.html`, vérifié le 25 juillet
|
||
|
|
2026 : 51 lois y figuraient, dont les 27 de la fenêtre de veille.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import re
|
||
|
|
|
||
|
|
from selectolax.parser import HTMLParser
|
||
|
|
|
||
|
|
from pipeline.collecteurs.base import Collecteur, ResultatCollecte, TexteCollecte
|
||
|
|
from pipeline.collecteurs.http import ClientHttp
|
||
|
|
from pipeline.journal import logger
|
||
|
|
from pipeline.modeles import Confiance, Evenement, Source, Statut, Tier, TypeEtape, TypeTexte
|
||
|
|
from pipeline.parseurs.dates_fr import lire_date
|
||
|
|
|
||
|
|
log = logger("collecteur.senat")
|
||
|
|
|
||
|
|
BASE = "https://www.senat.fr"
|
||
|
|
URL_LOIS_PROMULGUEES = f"{BASE}/dossiers-legislatifs/lois-promulguees.html"
|
||
|
|
|
||
|
|
# « Loi n° 2026-650 du 23 juillet 2026 relative au renforcement… »
|
||
|
|
# « Loi organique n° 2026-410 du 28 mai 2026 … »
|
||
|
|
_INTITULE = re.compile(
|
||
|
|
r"Loi\s+(organique\s+)?n°\s*(\d{4}-\d{1,4})\s+du\s+"
|
||
|
|
r"(\d{1,2}(?:er)?\s+\w+\.?\s+\d{4})\s+(.*)",
|
||
|
|
re.IGNORECASE | re.DOTALL,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class CollecteurSenat(Collecteur):
|
||
|
|
nom = "senat"
|
||
|
|
|
||
|
|
def __init__(self, client: ClientHttp, *, annees: tuple[int, ...] = (2026,)) -> None:
|
||
|
|
self.client = client
|
||
|
|
self.annees = annees
|
||
|
|
|
||
|
|
def _collecter(self) -> ResultatCollecte:
|
||
|
|
resultat = ResultatCollecte(collecteur=self.nom)
|
||
|
|
|
||
|
|
reponse = self.client.get(URL_LOIS_PROMULGUEES)
|
||
|
|
if not reponse.a_reussi:
|
||
|
|
raise RuntimeError(f"HTTP {reponse.code} sur {URL_LOIS_PROMULGUEES}")
|
||
|
|
|
||
|
|
document = HTMLParser(reponse.texte)
|
||
|
|
vus: set[str] = set()
|
||
|
|
|
||
|
|
for ancre in document.css("a"):
|
||
|
|
libelle = " ".join(ancre.text(separator=" ", strip=True).split())
|
||
|
|
texte = self._lire_intitule(libelle, ancre.attributes.get("href"))
|
||
|
|
if texte is None or texte.numero_officiel in vus:
|
||
|
|
continue
|
||
|
|
vus.add(texte.numero_officiel)
|
||
|
|
resultat.textes.append(texte)
|
||
|
|
|
||
|
|
log.debug("lois promulguées relevées", nombre=len(resultat.textes))
|
||
|
|
return resultat
|
||
|
|
|
||
|
|
def _lire_intitule(self, libelle: str, href: str | None) -> TexteCollecte | None:
|
||
|
|
correspondance = _INTITULE.search(libelle)
|
||
|
|
if not correspondance:
|
||
|
|
return None
|
||
|
|
|
||
|
|
organique = bool(correspondance.group(1))
|
||
|
|
numero = correspondance.group(2)
|
||
|
|
if int(numero.split("-")[0]) not in self.annees:
|
||
|
|
return None
|
||
|
|
|
||
|
|
date_promulgation = lire_date(correspondance.group(3))
|
||
|
|
if date_promulgation is None:
|
||
|
|
return None
|
||
|
|
|
||
|
|
# Le libellé se termine par un état (« parue », « en cours »)
|
||
|
|
# qu'il ne faut pas confondre avec l'objet de la loi.
|
||
|
|
objet = re.split(r"\s+(?:parue|en cours)\b", correspondance.group(4))[0].strip(" .;")
|
||
|
|
titre = (
|
||
|
|
f"LOI {'organique ' if organique else ''}n° {numero} "
|
||
|
|
f"du {correspondance.group(3)} {objet}"
|
||
|
|
)
|
||
|
|
|
||
|
|
url = _absolu(href) or URL_LOIS_PROMULGUEES
|
||
|
|
|
||
|
|
return TexteCollecte(
|
||
|
|
titre=" ".join(titre.split()),
|
||
|
|
source_url=url,
|
||
|
|
collecteur=self.nom,
|
||
|
|
numero_officiel=numero,
|
||
|
|
type=TypeTexte.LOI_ORGANIQUE if organique else TypeTexte.LOI,
|
||
|
|
statut=Statut.PROMULGUEE,
|
||
|
|
date_promulgation=date_promulgation,
|
||
|
|
evenements=[
|
||
|
|
Evenement(
|
||
|
|
date_evenement=date_promulgation,
|
||
|
|
type_etape=TypeEtape.PROMULGATION,
|
||
|
|
description=f"Loi promulguée n° {numero}, relevée sur la liste du Sénat",
|
||
|
|
source_url=url,
|
||
|
|
)
|
||
|
|
],
|
||
|
|
sources=[
|
||
|
|
Source(
|
||
|
|
url=url,
|
||
|
|
titre=" ".join(titre.split())[:280],
|
||
|
|
editeur="Sénat",
|
||
|
|
date_publication=date_promulgation,
|
||
|
|
tier=Tier.T1,
|
||
|
|
confiance=Confiance.HIGH,
|
||
|
|
marqueur=f"senat-{numero}",
|
||
|
|
fichier_origine="collecteur:senat",
|
||
|
|
)
|
||
|
|
],
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _absolu(href: str | None) -> str | None:
|
||
|
|
if not href:
|
||
|
|
return None
|
||
|
|
if href.startswith("http"):
|
||
|
|
return href
|
||
|
|
if href.startswith("/"):
|
||
|
|
return BASE + href
|
||
|
|
return None
|