billisdead
67bb0095fa
Implements a two-layer approach to ensure legal texts are accurate and current: Option A — runtime injection (legifrance_client.py): - LegifranceClient: OAuth2 via PISTE (oauth.piste.gouv.fr), fetches consolidated article text via api.piste.gouv.fr/dila/legifrance/lf-engine-app - Redis cache with 24h TTL (in-memory fallback if no Redis) - build_filter_prompt() injects live texts into the prompt with explicit priority over static descriptions — model uses official Légifrance text, not training memory - Graceful fallback to static prompt if API is unavailable — filtering never blocks Option B — weekly sync script (scripts/check_legal_refs.py): - Reads legal_refs.yaml (11 tracked articles, priority critical/high/medium) - Fetches current text from Légifrance, diffs against stored baseline - --init: resolves LEGIARTI IDs on first run - --check: report only; --update: saves new texts to YAML - Exit code 2 when changes detected (CI/n8n-compatible) Supporting changes: - legal_refs.yaml: tracked articles with code LEGITEXT IDs, priorities, rationale - requirements.txt: add pyyaml>=6.0.2 and requests>=2.32.0 - .env.example: document PISTE_CLIENT_ID, PISTE_CLIENT_SECRET, LEGIFRANCE_CACHE_TTL - transparence.tsx: document the Légifrance injection for public transparency - Fixed portal URL: piste.gouv.fr (not piste.api.gouv.fr — verified June 2026) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
280 lines
9.9 KiB
Python
280 lines
9.9 KiB
Python
"""
|
|
La Voix du Peuple — Client API Légifrance (PISTE)
|
|
Copyright (C) 2026 billisdead — Licence EUPL-1.2
|
|
|
|
Récupère les textes des articles légaux en vigueur via l'API officielle PISTE.
|
|
Cache les résultats dans Redis (24h) ou en mémoire si Redis n'est pas configuré.
|
|
|
|
Prérequis :
|
|
Créer un compte sur https://piste.gouv.fr et obtenir
|
|
PISTE_CLIENT_ID + PISTE_CLIENT_SECRET (inscription gratuite).
|
|
"""
|
|
import json
|
|
import logging
|
|
import os
|
|
import time
|
|
|
|
import requests
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
PISTE_AUTH_URL = "https://oauth.piste.gouv.fr/api/oauth/token" # vérifié juin 2026
|
|
PISTE_API_BASE = "https://api.piste.gouv.fr/dila/legifrance/lf-engine-app" # vérifié juin 2026
|
|
DEFAULT_CACHE_TTL = int(os.environ.get("LEGIFRANCE_CACHE_TTL", 86400)) # 24h
|
|
|
|
|
|
class LegifranceClient:
|
|
"""
|
|
Client PISTE avec cache Redis (fallback mémoire).
|
|
Utilisation :
|
|
client = LegifranceClient()
|
|
if client.available:
|
|
legiarti_id, text = client.search_article("LEGITEXT000006070719", "225-1")
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.client_id = os.environ.get("PISTE_CLIENT_ID")
|
|
self.client_secret = os.environ.get("PISTE_CLIENT_SECRET")
|
|
self._token: str | None = None
|
|
self._token_exp: float = 0.0
|
|
self._redis = None
|
|
self._mem: dict[str, tuple[str, float]] = {}
|
|
self._init_redis()
|
|
|
|
def _init_redis(self):
|
|
url = os.environ.get("REDIS_URL")
|
|
if url:
|
|
try:
|
|
import redis as _r
|
|
r = _r.from_url(url)
|
|
r.ping()
|
|
self._redis = r
|
|
except Exception:
|
|
pass
|
|
|
|
@property
|
|
def available(self) -> bool:
|
|
return bool(self.client_id and self.client_secret)
|
|
|
|
# ── Cache ────────────────────────────────────────────────────────────────
|
|
|
|
def _get(self, key: str) -> str | None:
|
|
if self._redis:
|
|
try:
|
|
v = self._redis.get(key)
|
|
if v:
|
|
return v.decode()
|
|
except Exception:
|
|
pass
|
|
entry = self._mem.get(key)
|
|
if entry and time.time() < entry[1]:
|
|
return entry[0]
|
|
return None
|
|
|
|
def _set(self, key: str, value: str, ttl: int = DEFAULT_CACHE_TTL):
|
|
if self._redis:
|
|
try:
|
|
self._redis.setex(key, ttl, value)
|
|
return
|
|
except Exception:
|
|
pass
|
|
self._mem[key] = (value, time.time() + ttl)
|
|
|
|
# ── Auth ─────────────────────────────────────────────────────────────────
|
|
|
|
def _token_headers(self) -> dict[str, str]:
|
|
if not (self._token and time.time() < self._token_exp):
|
|
resp = requests.post(
|
|
PISTE_AUTH_URL,
|
|
data={
|
|
"grant_type": "client_credentials",
|
|
"client_id": self.client_id,
|
|
"client_secret": self.client_secret,
|
|
"scope": "openid",
|
|
},
|
|
timeout=10,
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
self._token = data["access_token"]
|
|
self._token_exp = time.time() + data.get("expires_in", 3600) - 60
|
|
return {"Authorization": f"Bearer {self._token}"}
|
|
|
|
# ── API calls ─────────────────────────────────────────────────────────────
|
|
|
|
def fetch_article_by_id(self, legiarti_id: str) -> str | None:
|
|
"""Récupère le texte brut d'un article via son identifiant LEGIARTI."""
|
|
cache_key = f"lgf:art:{legiarti_id}"
|
|
if (cached := self._get(cache_key)):
|
|
return cached
|
|
try:
|
|
resp = requests.post(
|
|
f"{PISTE_API_BASE}/consult/getArticle",
|
|
json={"id": legiarti_id},
|
|
headers=self._token_headers(),
|
|
timeout=12,
|
|
)
|
|
resp.raise_for_status()
|
|
article = resp.json().get("article", {})
|
|
# L'API peut retourner texte brut ou HTML selon la version
|
|
text = article.get("texte") or article.get("texteHtml") or ""
|
|
text = text.strip()
|
|
if text:
|
|
self._set(cache_key, text)
|
|
return text or None
|
|
except Exception:
|
|
logger.warning("Légifrance: échec fetch article %s", legiarti_id)
|
|
return None
|
|
|
|
def search_article(
|
|
self, code_legitext: str, article_num: str
|
|
) -> tuple[str | None, str | None]:
|
|
"""
|
|
Cherche un article par code LEGITEXT + numéro d'article.
|
|
Retourne (legiarti_id, texte) ou (None, None).
|
|
|
|
Le format article_num accepté : "225-1", "24 bis", "222-33-2-2", etc.
|
|
La recherche tente plusieurs variantes pour maximiser les chances de match.
|
|
"""
|
|
cache_key = f"lgf:search:{code_legitext}:{article_num}"
|
|
if (cached := self._get(cache_key)):
|
|
data = json.loads(cached)
|
|
return data.get("id"), data.get("text")
|
|
|
|
try:
|
|
headers = self._token_headers()
|
|
|
|
# Variantes du numéro d'article pour la recherche
|
|
variants = [article_num]
|
|
if not article_num[0].isalpha():
|
|
variants += [f"L{article_num}", f"R{article_num}"]
|
|
|
|
for variant in variants:
|
|
result = self._search_variant(code_legitext, variant, headers)
|
|
if result:
|
|
legiarti_id, text = result
|
|
if legiarti_id and text:
|
|
self._set(
|
|
cache_key,
|
|
json.dumps({"id": legiarti_id, "text": text}),
|
|
)
|
|
return legiarti_id, text
|
|
|
|
return None, None
|
|
|
|
except Exception:
|
|
logger.warning(
|
|
"Légifrance: échec search %s %s", code_legitext, article_num
|
|
)
|
|
return None, None
|
|
|
|
def _search_variant(
|
|
self, code_legitext: str, article_num: str, headers: dict
|
|
) -> tuple[str, str] | None:
|
|
"""Tente une recherche avec un numéro d'article particulier."""
|
|
resp = requests.post(
|
|
f"{PISTE_API_BASE}/search",
|
|
json={
|
|
"recherche": {
|
|
"champs": [
|
|
{
|
|
"typeChamp": "NUM_ARTICLE",
|
|
"criteres": [
|
|
{"typeRecherche": "EXACTE", "valeur": article_num}
|
|
],
|
|
"operateur": "ET",
|
|
}
|
|
],
|
|
"filtres": [
|
|
{"facette": "CODE_ID", "valeurs": [code_legitext]},
|
|
{
|
|
"facette": "TEXT_LEGAL_STATUS",
|
|
"valeurs": ["VIGUEUR"],
|
|
},
|
|
],
|
|
"pageNumber": 1,
|
|
"pageSize": 3,
|
|
"operateur": "ET",
|
|
"sort": "PERTINENCE",
|
|
"typePagination": "DEFAUT",
|
|
},
|
|
"fond": "CODE_DATE",
|
|
},
|
|
headers=headers,
|
|
timeout=15,
|
|
)
|
|
resp.raise_for_status()
|
|
payload = resp.json()
|
|
|
|
# L'API PISTE peut envoyer les résultats sous différentes structures
|
|
results = payload.get("results", [])
|
|
if not results:
|
|
return None
|
|
|
|
first = results[0]
|
|
|
|
# Cas 1 : résultat direct avec id + texte
|
|
legiarti_id = first.get("id")
|
|
text = first.get("texte") or first.get("extract") or ""
|
|
|
|
# Cas 2 : résultat imbriqué (sections → articles)
|
|
if not legiarti_id:
|
|
for section in first.get("sections", []):
|
|
for art in section.get("articles", []):
|
|
legiarti_id = art.get("id")
|
|
text = art.get("texte") or ""
|
|
if legiarti_id:
|
|
break
|
|
if legiarti_id:
|
|
break
|
|
|
|
if not legiarti_id:
|
|
return None
|
|
|
|
# Si le texte est vide dans les résultats, on le récupère par ID
|
|
if not text.strip():
|
|
text = self.fetch_article_by_id(legiarti_id) or ""
|
|
|
|
return (legiarti_id, text.strip()) if text.strip() else None
|
|
|
|
def fetch_tracked_articles(self, refs_path: str) -> dict[str, str]:
|
|
"""
|
|
Charge legal_refs.yaml et récupère les textes des articles critical/high.
|
|
Retourne {ref: texte} pour injection dans le prompt.
|
|
"""
|
|
if not self.available:
|
|
return {}
|
|
try:
|
|
import yaml # pyyaml
|
|
with open(refs_path) as f:
|
|
refs = yaml.safe_load(f)
|
|
except Exception:
|
|
logger.warning("Légifrance: impossible de charger %s", refs_path)
|
|
return {}
|
|
|
|
live: dict[str, str] = {}
|
|
for article in refs.get("articles", []):
|
|
if article.get("priority") not in ("critical", "high"):
|
|
continue
|
|
ref = article["ref"]
|
|
legiarti_id = article.get("legiarti_id")
|
|
if legiarti_id:
|
|
text = self.fetch_article_by_id(legiarti_id)
|
|
else:
|
|
_, text = self.search_article(article["code"], article["article_num"])
|
|
if text:
|
|
live[ref] = text
|
|
|
|
return live
|
|
|
|
|
|
# Singleton : un client par worker Gunicorn, cache partagé via Redis
|
|
_client: LegifranceClient | None = None
|
|
|
|
|
|
def get_legifrance_client() -> LegifranceClient:
|
|
global _client
|
|
if _client is None:
|
|
_client = LegifranceClient()
|
|
return _client
|