veye-lalwa/pipeline/collecteurs/legifrance.py

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