"""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)