Add Légifrance API integration for live legal text injection

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>
This commit is contained in:
2026-06-24 09:19:23 +02:00
parent e8b4bde2f4
commit 67bb0095fa
8 changed files with 764 additions and 2 deletions
+24 -2
View File
@@ -9,15 +9,33 @@ import json
import os
import logging
from openai import OpenAI, BadRequestError
from legal_framework import LEGAL_FILTER_PROMPT, SYNTHESIS_PROMPT, LAW_CHECK_PROMPT
from legal_framework import LEGAL_FILTER_PROMPT, SYNTHESIS_PROMPT, LAW_CHECK_PROMPT, build_filter_prompt
from legifrance_client import get_legifrance_client
logger = logging.getLogger(__name__)
MISTRAL_BASE_URL = "https://api.mistral.ai/v1"
_LEGAL_REFS_PATH = os.path.join(os.path.dirname(__file__), "legal_refs.yaml")
_client: OpenAI | None = None
def _get_live_legal_texts() -> dict[str, str]:
"""
Récupère les textes consolidés des articles critiques depuis Légifrance (PISTE).
Retourne {} si PISTE n'est pas configuré ou en cas d'erreur — sans bloquer le filtrage.
Les résultats sont mis en cache dans Redis (24h).
"""
try:
lf = get_legifrance_client()
if not lf.available:
return {}
return lf.fetch_tracked_articles(_LEGAL_REFS_PATH)
except Exception:
logger.warning("Légifrance: impossible de récupérer les textes live — mode statique")
return {}
def get_client() -> OpenAI:
"""
Supporte deux modes (par ordre de priorité) :
@@ -58,12 +76,16 @@ def filter_idea(content: str) -> dict:
try:
client = get_client()
filter_model = os.environ.get("FILTER_MODEL", os.environ.get("OPENAI_FILTER_MODEL", "mistral-small-latest"))
live_texts = _get_live_legal_texts()
system_prompt = build_filter_prompt(live_texts)
if live_texts:
logger.debug("Légifrance: %d article(s) injecté(s) dans le prompt", len(live_texts))
response = client.chat.completions.create(
model=filter_model,
max_tokens=300,
response_format={"type": "json_object"},
messages=[
{"role": "system", "content": LEGAL_FILTER_PROMPT},
{"role": "system", "content": system_prompt},
{"role": "user", "content": f'Idée soumise : "{content}"'},
],
)
+50
View File
@@ -19,7 +19,57 @@ Sources françaises :
- Code pénal français (partie législative et réglementaire)
- Loi sur la liberté de la presse du 29 juillet 1881
- Code civil français
Injection dynamique :
- Les textes consolidés des articles critiques sont récupérés en temps réel
depuis l'API Légifrance (PISTE) et injectés dans le prompt via build_filter_prompt().
- Le suivi des changements est assuré par scripts/check_legal_refs.py.
- Fichier de configuration : legal_refs.yaml (articles surveillés, LEGIARTI IDs).
"""
from __future__ import annotations
import logging
from datetime import date
logger = logging.getLogger(__name__)
_CRITERIA_ANCHOR = (
"═══════════════════════════════════════════════════════════════════════════════\n"
"CRITÈRES D'ACCEPTATION"
)
def build_filter_prompt(live_texts: dict[str, str] | None = None) -> str:
"""
Construit le prompt de filtrage en injectant les textes légaux consolidés
récupérés depuis Légifrance.
live_texts : {ref_article: texte_en_vigueur} issu de LegifranceClient.
Si vide ou None, retourne LEGAL_FILTER_PROMPT statique.
Les textes injectés ont PRIORITÉ ABSOLUE sur les descriptions statiques
du cadre légal. Ils sont placés juste avant les critères d'acceptation.
"""
if not live_texts:
return LEGAL_FILTER_PROMPT
today = date.today().isoformat()
block = (
"═══════════════════════════════════════════════════════════════════════════════\n"
f"TEXTES LÉGAUX EN VIGUEUR — SOURCE LÉGIFRANCE ({today})\n"
"═══════════════════════════════════════════════════════════════════════════════\n\n"
"Ces extraits sont récupérés depuis l'API officielle Légifrance (PISTE).\n"
"Ils ont PRIORITÉ ABSOLUE sur les descriptions du cadre ci-dessus.\n\n"
)
for ref, text in live_texts.items():
block += f"[{ref}] — texte consolidé en vigueur :\n{text}\n\n"
if _CRITERIA_ANCHOR in LEGAL_FILTER_PROMPT:
parts = LEGAL_FILTER_PROMPT.split(_CRITERIA_ANCHOR, 1)
return parts[0] + block + _CRITERIA_ANCHOR + parts[1]
logger.warning("build_filter_prompt: ancre CRITÈRES D'ACCEPTATION introuvable — fallback statique")
return LEGAL_FILTER_PROMPT
LEGAL_FILTER_PROMPT = """
Tu es un agent de filtrage éthique pour une plateforme démocratique citoyenne.
+153
View File
@@ -0,0 +1,153 @@
# Références légales surveillées via API Légifrance (PISTE)
# Ce fichier est la source de vérité pour le suivi des articles.
#
# Mise à jour :
# python scripts/check_legal_refs.py --init # peuple les legiarti_id
# python scripts/check_legal_refs.py --check # rapport des changements
# python scripts/check_legal_refs.py --update # met à jour last_text
#
# Inscription PISTE (gratuite) : https://piste.gouv.fr
meta:
description: "Articles critiques — filtrage éthique La Voix du Peuple"
last_sync: null
source: "https://www.legifrance.gouv.fr"
piste_docs: "https://piste.gouv.fr/documentation"
# LEGITEXT IDs stables des codes sources
codes:
code_penal: LEGITEXT000006070719
loi_1881: LEGITEXT000006070722
lcen_2004: LEGITEXT000005789847
code_civil: LEGITEXT000006070721
code_sante_pub: LEGITEXT000006072665
articles:
# ── PRIORITÉ CRITIQUE — amendés fréquemment ────────────────────────────────
- key: CP-225-1
ref: "Code pénal, Art. 225-1"
code: LEGITEXT000006070719
article_num: "225-1"
legiarti_id: null # peuplé par --init
priority: critical
reason: >
Liste des critères de discrimination — fréquemment élargie.
Dernière modification significative connue : 2021 (ajout activités syndicales).
Surveiller ajouts éventuels liés à SREN 2024 ou lois ultérieures.
last_text: null # peuplé par --update
verified_at: null
- key: L1881-24
ref: "Loi du 29 juillet 1881, Art. 24"
code: LEGITEXT000006070722
article_num: "24"
legiarti_id: null
priority: critical
reason: >
Provocation à la discrimination et à la haine — fréquemment amendé.
L'apologie du terrorisme en a été retirée en 2014 (transférée CP Art. 421-2-5).
Vérifier toute évolution du périmètre des motifs protégés.
last_text: null
verified_at: null
# ── PRIORITÉ HAUTE ─────────────────────────────────────────────────────────
- key: CP-222-33
ref: "Code pénal, Art. 222-33"
code: LEGITEXT000006070719
article_num: "222-33"
legiarti_id: null
priority: high
reason: >
Harcèlement sexuel — amendé en 2018 (loi n° 2018-703, Schiappa).
Définition élargie, possibles évolutions.
last_text: null
verified_at: null
- key: CP-222-33-2-2
ref: "Code pénal, Art. 222-33-2-2"
code: LEGITEXT000006070719
article_num: "222-33-2-2"
legiarti_id: null
priority: high
reason: >
Cyberharcèlement en meute — introduit par loi n° 2018-703 du 3 août 2018.
Surveiller évolutions liées à SREN 2024 (art. 19 et ss).
last_text: null
verified_at: null
- key: CP-421-2-5
ref: "Code pénal, Art. 421-2-5"
code: LEGITEXT000006070719
article_num: "421-2-5"
legiarti_id: null
priority: high
reason: >
Apologie du terrorisme — introduit par loi n° 2014-1353 du 13 novembre 2014
(anciennement Loi 1881 Art. 24 al. 6). Stable depuis, mais à vérifier.
last_text: null
verified_at: null
- key: CP-211-1
ref: "Code pénal, Art. 211-1"
code: LEGITEXT000006070719
article_num: "211-1"
legiarti_id: null
priority: high
reason: "Génocide — texte fondateur, stable depuis l'entrée en vigueur du code (1994)."
last_text: null
verified_at: null
- key: L1881-24bis
ref: "Loi du 29 juillet 1881, Art. 24 bis"
code: LEGITEXT000006070722
article_num: "24 bis"
legiarti_id: null
priority: high
reason: "Négationnisme — loi Gayssot du 13 juillet 1990, stable."
last_text: null
verified_at: null
- key: CP-226-4-1
ref: "Code pénal, Art. 226-4-1"
code: LEGITEXT000006070719
article_num: "226-4-1"
legiarti_id: null
priority: high
reason: "Usurpation d'identité numérique — LOPPSI 2 (2011). Surveiller extensions."
last_text: null
verified_at: null
# ── PRIORITÉ MOYENNE ───────────────────────────────────────────────────────
- key: CP-212-1
ref: "Code pénal, Art. 212-1"
code: LEGITEXT000006070719
article_num: "212-1"
legiarti_id: null
priority: medium
reason: "Crimes contre l'humanité — stable."
last_text: null
verified_at: null
- key: CP-223-13
ref: "Code pénal, Art. 223-13"
code: LEGITEXT000006070719
article_num: "223-13"
legiarti_id: null
priority: medium
reason: "Provocation au suicide — surveiller évolutions liées aux réseaux sociaux."
last_text: null
verified_at: null
- key: L1881-29
ref: "Loi du 29 juillet 1881, Art. 29"
code: LEGITEXT000006070722
article_num: "29"
legiarti_id: null
priority: medium
reason: "Définition de la diffamation — stable."
last_text: null
verified_at: null
+279
View File
@@ -0,0 +1,279 @@
"""
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
+2
View File
@@ -6,4 +6,6 @@ gunicorn>=23.0.0
openai>=1.77.0
psycopg2-binary>=2.9.10
python-dotenv>=1.0.1
pyyaml>=6.0.2
redis>=5.0.0
requests>=2.32.0