229 lines
8.5 KiB
Python
229 lines
8.5 KiB
Python
 
|
"""Collecteur Légifrance, via l'API PISTE de la DILA.
|
||
|
|
|
||
|
|
Authentification OAuth 2.0 en `client_credentials`. Deux pièges vérifiés le
|
||
|
|
25 juillet 2026 sur le compte réel :
|
||
|
|
|
||
|
|
- une application PISTE expose **deux** couples de valeurs, et seul le couple
|
||
|
|
OAuth (« Client ID » / « Client secret ») ouvre le jeton ; la clé d'API
|
||
|
|
(« API key » / « API key secret ») produit un `invalid_client` ;
|
||
|
|
- les identifiants de production ne sont **pas** acceptés par le bac à sable,
|
||
|
|
qui exige une application distincte.
|
||
|
|
|
||
|
|
Sans identifiants, le collecteur se déclare indisponible et le pipeline bascule
|
||
|
|
sur le repli documenté : promulgations de l'Assemblée nationale, liste des lois
|
||
|
|
du Sénat, décisions du Conseil constitutionnel.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import os
|
||
|
|
import time
|
||
|
|
from datetime import date
|
||
|
|
|
||
|
|
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.legifrance")
|
||
|
|
|
||
|
|
URL_JETON = {
|
||
|
|
"prod": "https://oauth.piste.gouv.fr/api/oauth/token",
|
||
|
|
"sandbox": "https://sandbox-oauth.piste.gouv.fr/api/oauth/token",
|
||
|
|
}
|
||
|
|
URL_API = {
|
||
|
|
"prod": "https://api.piste.gouv.fr/dila/legifrance/lf-engine-app",
|
||
|
|
"sandbox": "https://sandbox-api.piste.gouv.fr/dila/legifrance/lf-engine-app",
|
||
|
|
}
|
||
|
|
|
||
|
|
# Le lien public d'un texte se déduit de son identifiant JORF.
|
||
|
|
GABARIT_PUBLIC = "https://www.legifrance.gouv.fr/jorf/id/{identifiant}"
|
||
|
|
|
||
|
|
|
||
|
|
class CollecteurLegifrance(Collecteur):
|
||
|
|
"""Interroge le fonds LODA (lois, ordonnances, décrets, arrêtés)."""
|
||
|
|
|
||
|
|
nom = "legifrance"
|
||
|
|
|
||
|
|
def __init__(
|
||
|
|
self,
|
||
|
|
client: ClientHttp,
|
||
|
|
*,
|
||
|
|
numeros_a_verifier: list[str] | None = None,
|
||
|
|
annee: int = 2026,
|
||
|
|
) -> None:
|
||
|
|
self.client = client
|
||
|
|
self.numeros_a_verifier = numeros_a_verifier or []
|
||
|
|
self.annee = annee
|
||
|
|
self._jeton: str | None = None
|
||
|
|
self._jeton_expire_a: float = 0.0
|
||
|
|
|
||
|
|
# ── Disponibilité ────────────────────────────────────────────────────────
|
||
|
|
def est_disponible(self) -> tuple[bool, str | None]:
|
||
|
|
if not os.environ.get("LEGIFRANCE_CLIENT_ID"):
|
||
|
|
return False, (
|
||
|
|
"LEGIFRANCE_CLIENT_ID absent de .env — repli sur l'Assemblée "
|
||
|
|
"nationale, le Sénat et le Conseil constitutionnel"
|
||
|
|
)
|
||
|
|
if not os.environ.get("LEGIFRANCE_CLIENT_SECRET"):
|
||
|
|
return False, "LEGIFRANCE_CLIENT_SECRET absent de .env"
|
||
|
|
return True, None
|
||
|
|
|
||
|
|
# ── Authentification ─────────────────────────────────────────────────────
|
||
|
|
@property
|
||
|
|
def environnement(self) -> str:
|
||
|
|
valeur = os.environ.get("LEGIFRANCE_ENV", "prod").lower()
|
||
|
|
return valeur if valeur in URL_JETON else "prod"
|
||
|
|
|
||
|
|
def _obtenir_jeton(self) -> str:
|
||
|
|
"""Jeton OAuth, renouvelé une minute avant son expiration."""
|
||
|
|
if self._jeton and time.monotonic() < self._jeton_expire_a:
|
||
|
|
return self._jeton
|
||
|
|
|
||
|
|
reponse = self.client.post(
|
||
|
|
URL_JETON[self.environnement],
|
||
|
|
donnees={
|
||
|
|
"grant_type": "client_credentials",
|
||
|
|
"client_id": os.environ["LEGIFRANCE_CLIENT_ID"],
|
||
|
|
"client_secret": os.environ["LEGIFRANCE_CLIENT_SECRET"],
|
||
|
|
"scope": "openid",
|
||
|
|
},
|
||
|
|
utiliser_cache=False,
|
||
|
|
)
|
||
|
|
if not reponse.a_reussi:
|
||
|
|
raise RuntimeError(
|
||
|
|
f"authentification PISTE refusée (HTTP {reponse.code}) : "
|
||
|
|
f"{reponse.texte[:200]} — vérifier qu'il s'agit bien du couple "
|
||
|
|
"OAuth et non de la clé d'API"
|
||
|
|
)
|
||
|
|
|
||
|
|
charge = reponse.json()
|
||
|
|
self._jeton = charge["access_token"]
|
||
|
|
self._jeton_expire_a = time.monotonic() + max(int(charge.get("expires_in", 3600)) - 60, 60)
|
||
|
|
log.debug("jeton PISTE obtenu", scope=charge.get("scope"))
|
||
|
|
return self._jeton
|
||
|
|
|
||
|
|
def _entetes(self) -> dict[str, str]:
|
||
|
|
return {
|
||
|
|
"Authorization": f"Bearer {self._obtenir_jeton()}",
|
||
|
|
"Content-Type": "application/json",
|
||
|
|
"Accept": "application/json",
|
||
|
|
}
|
||
|
|
|
||
|
|
# ── Collecte ─────────────────────────────────────────────────────────────
|
||
|
|
def _collecter(self) -> ResultatCollecte:
|
||
|
|
resultat = ResultatCollecte(collecteur=self.nom)
|
||
|
|
|
||
|
|
for numero in self.numeros_a_verifier:
|
||
|
|
try:
|
||
|
|
if texte := self.chercher_par_numero(numero):
|
||
|
|
resultat.textes.append(texte)
|
||
|
|
except Exception as erreur: # noqa: BLE001
|
||
|
|
resultat.erreurs.append(f"{numero} : {erreur}")
|
||
|
|
|
||
|
|
return resultat
|
||
|
|
|
||
|
|
def chercher_par_numero(self, numero: str) -> TexteCollecte | None:
|
||
|
|
"""Retrouve une loi par son numéro officiel (« 2026-491 »)."""
|
||
|
|
charge = {
|
||
|
|
"recherche": {
|
||
|
|
"champs": [
|
||
|
|
{
|
||
|
|
"typeChamp": "NUM",
|
||
|
|
"criteres": [
|
||
|
|
{"typeRecherche": "EXACTE", "valeur": numero, "operateur": "ET"}
|
||
|
|
],
|
||
|
|
"operateur": "ET",
|
||
|
|
}
|
||
|
|
],
|
||
|
|
"filtres": [],
|
||
|
|
"pageNumber": 1,
|
||
|
|
"pageSize": 5,
|
||
|
|
"operateur": "ET",
|
||
|
|
"sort": "PERTINENCE",
|
||
|
|
"typePagination": "DEFAUT",
|
||
|
|
},
|
||
|
|
"fond": "LODA_DATE",
|
||
|
|
}
|
||
|
|
|
||
|
|
reponse = self.client.post(
|
||
|
|
f"{URL_API[self.environnement]}/search",
|
||
|
|
json_corps=charge,
|
||
|
|
entetes=self._entetes(),
|
||
|
|
)
|
||
|
|
if not reponse.a_reussi:
|
||
|
|
raise RuntimeError(f"recherche Légifrance HTTP {reponse.code}: {reponse.texte[:160]}")
|
||
|
|
|
||
|
|
donnees = reponse.json()
|
||
|
|
for element in donnees.get("results") or []:
|
||
|
|
for titre in element.get("titles") or []:
|
||
|
|
if texte := _lire_titre(titre, numero):
|
||
|
|
return texte
|
||
|
|
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def _lire_titre(titre: dict, numero_attendu: str) -> TexteCollecte | None:
|
||
|
|
"""Convertit une entrée de résultat Légifrance en texte collecté."""
|
||
|
|
intitule = (titre.get("title") or "").strip()
|
||
|
|
if not intitule or numero_attendu not in intitule:
|
||
|
|
return None
|
||
|
|
|
||
|
|
identifiant = titre.get("cid") or titre.get("id") or ""
|
||
|
|
url = GABARIT_PUBLIC.format(identifiant=identifiant) if identifiant else None
|
||
|
|
if url is None:
|
||
|
|
return None
|
||
|
|
|
||
|
|
signature = _date_du_titre(titre, intitule)
|
||
|
|
organique = "LOI organique" in intitule or "LOI ORGANIQUE" in intitule.upper()
|
||
|
|
|
||
|
|
evenements = []
|
||
|
|
if signature:
|
||
|
|
evenements.append(
|
||
|
|
Evenement(
|
||
|
|
date_evenement=signature,
|
||
|
|
type_etape=TypeEtape.PROMULGATION,
|
||
|
|
description=f"Promulgation constatée sur Légifrance : {intitule[:200]}",
|
||
|
|
source_url=url,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
|
||
|
|
return TexteCollecte(
|
||
|
|
titre=intitule,
|
||
|
|
source_url=url,
|
||
|
|
collecteur="legifrance",
|
||
|
|
numero_officiel=numero_attendu,
|
||
|
|
type=TypeTexte.LOI_ORGANIQUE if organique else TypeTexte.LOI,
|
||
|
|
statut=Statut.PROMULGUEE,
|
||
|
|
date_promulgation=signature,
|
||
|
|
identifiant_externe=identifiant,
|
||
|
|
evenements=evenements,
|
||
|
|
sources=[
|
||
|
|
Source(
|
||
|
|
url=url,
|
||
|
|
titre=intitule[:280],
|
||
|
|
editeur="Légifrance",
|
||
|
|
date_publication=signature,
|
||
|
|
tier=Tier.T1,
|
||
|
|
confiance=Confiance.HIGH,
|
||
|
|
marqueur=f"legifrance-{identifiant}",
|
||
|
|
fichier_origine="collecteur:legifrance",
|
||
|
|
)
|
||
|
|
],
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _date_du_titre(titre: dict, intitule: str) -> date | None:
|
||
|
|
"""Date de signature : champ dédié quand il existe, sinon lue dans l'intitulé.
|
||
|
|
|
||
|
|
Les résultats de recherche laissent souvent `dateSignature` à `None` ;
|
||
|
|
l'intitulé officiel, lui, porte toujours la date (« LOI n° 2026-491 du
|
||
|
|
12 juin 2026 … »).
|
||
|
|
"""
|
||
|
|
for champ in ("dateSignature", "dateDebut", "datePublication"):
|
||
|
|
if valeur := titre.get(champ):
|
||
|
|
if lue := lire_date(str(valeur)):
|
||
|
|
return lue
|
||
|
|
return lire_date(intitule)
|