#!/usr/bin/env python3 """ La Voix du Peuple — Vérification des références légales via Légifrance (PISTE) Copyright (C) 2026 billisdead — Licence EUPL-1.2 Usage : python scripts/check_legal_refs.py --init Peuple les legiarti_id (première fois) python scripts/check_legal_refs.py --check Rapport des changements (sans modifier) python scripts/check_legal_refs.py --update Met à jour last_text dans le YAML Codes de sortie : 0 Tout est à jour 1 Erreur (API, fichier, credentials) 2 Changements détectés (utile pour intégration CI ou n8n webhook) Nécessite PISTE_CLIENT_ID et PISTE_CLIENT_SECRET dans .env Inscription gratuite : https://piste.gouv.fr """ import argparse import difflib import os import sys from datetime import UTC, datetime # Ajoute flask-api au path pour importer legifrance_client REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, os.path.join(REPO_ROOT, "artifacts", "flask-api")) from dotenv import load_dotenv load_dotenv(os.path.join(REPO_ROOT, ".env")) try: import yaml except ImportError: print("❌ pyyaml manquant — installez : pip install pyyaml", file=sys.stderr) sys.exit(1) from legifrance_client import LegifranceClient REFS_PATH = os.path.join(REPO_ROOT, "artifacts", "flask-api", "legal_refs.yaml") PRIORITY_LABELS = { "critical": "🔴 CRITIQUE", "high": "🟠 HAUTE", "medium": "🟡 MOYENNE", } def load_refs() -> dict: with open(REFS_PATH, encoding="utf-8") as f: return yaml.safe_load(f) def save_refs(refs: dict): with open(REFS_PATH, "w", encoding="utf-8") as f: yaml.dump(refs, f, allow_unicode=True, default_flow_style=False, sort_keys=False) def cmd_init(client: LegifranceClient, refs: dict): """Peuple les legiarti_id manquants dans le YAML.""" print("Résolution des LEGIARTI IDs manquants via Légifrance…\n") updated = 0 for article in refs.get("articles", []): if article.get("legiarti_id"): print(f" ✓ {article['ref']} — ID déjà connu") continue new_id, text = client.search_article(article["code"], article["article_num"]) if new_id: article["legiarti_id"] = new_id updated += 1 snippet = (text or "")[:120].replace("\n", " ") print(f" ✅ {article['ref']}") print(f" ID : {new_id}") print(f" Ext : {snippet}…") else: print(f" ⚠️ {article['ref']} — introuvable (vérifier code/num)") if updated: refs["meta"]["last_sync"] = datetime.now(UTC).isoformat() save_refs(refs) print(f"\n✅ {updated} ID(s) sauvegardé(s) dans legal_refs.yaml") else: print("\nAucun nouvel ID trouvé.") def cmd_check(client: LegifranceClient, refs: dict, update: bool) -> int: """Compare les textes actuels avec ceux stockés. Retourne le nombre de changements.""" today = datetime.now(UTC).date().isoformat() changes = 0 for article in refs.get("articles", []): key = article["key"] ref = article["ref"] priority = PRIORITY_LABELS.get(article.get("priority", ""), "") legiarti_id = article.get("legiarti_id") last_text = (article.get("last_text") or "").strip() print(f"\n{'─' * 62}") print(f"{priority} {ref} [{key}]") if legiarti_id: live_text = client.fetch_article_by_id(legiarti_id) or "" else: new_id, live_text = client.search_article(article["code"], article["article_num"]) live_text = live_text or "" if new_id and update and not legiarti_id: article["legiarti_id"] = new_id print(f" ↳ LEGIARTI ID résolu : {new_id}") live_text = live_text.strip() if not live_text: print(" ⚠️ Texte non récupéré — API indisponible ou article introuvable") continue if not last_text: print(" ℹ️ Premier enregistrement (pas de baseline)") print(f" Extrait : {live_text[:200].replace(chr(10), ' ')}…") changes += 1 if update: article["last_text"] = live_text article["verified_at"] = today elif live_text == last_text: print(" ✅ Inchangé") else: changes += 1 print(" ⚠️ CHANGEMENT DÉTECTÉ") diff = list( difflib.unified_diff( last_text.splitlines(), live_text.splitlines(), fromfile="ancienne version", tofile="version Légifrance actuelle", lineterm="", n=2, ) ) for line in diff[:40]: prefix = " " if line.startswith("+"): prefix = " +" elif line.startswith("-"): prefix = " -" print(f"{prefix}{line}") if len(diff) > 40: print(f" … ({len(diff) - 40} lignes supplémentaires)") if update: article["last_text"] = live_text article["verified_at"] = today if update and changes: refs["meta"]["last_sync"] = datetime.now(UTC).isoformat() save_refs(refs) print(f"\n✅ legal_refs.yaml mis à jour ({changes} article(s) modifié(s))") return changes def main(): parser = argparse.ArgumentParser( description="Vérification des références légales via API Légifrance (PISTE)" ) parser.add_argument( "--init", action="store_true", help="Peuple les legiarti_id manquants (première exécution)", ) parser.add_argument( "--check", action="store_true", help="Rapport des changements — ne modifie pas le YAML", ) parser.add_argument( "--update", action="store_true", help="Met à jour last_text dans legal_refs.yaml", ) args = parser.parse_args() if not (args.init or args.check or args.update): parser.print_help() sys.exit(0) client = LegifranceClient() if not client.available: print( "❌ PISTE_CLIENT_ID et/ou PISTE_CLIENT_SECRET non définis dans .env\n" " Inscription gratuite : https://piste.gouv.fr\n" " Ajoutez ensuite dans .env :\n" " PISTE_CLIENT_ID=...\n" " PISTE_CLIENT_SECRET=...", file=sys.stderr, ) sys.exit(1) refs = load_refs() if args.init: cmd_init(client, refs) return changes = cmd_check(client, refs, update=args.update) print(f"\n{'═' * 62}") if changes: print(f"⚠️ {changes} article(s) à vérifier / mis à jour.") if not args.update: print(" Relancez avec --update pour sauvegarder les textes actuels.") sys.exit(2) else: print("✅ Toutes les références sont à jour.") sys.exit(0) if __name__ == "__main__": main()