""" Reference d'extraction, phase 0 : mesurer AVANT de corriger quoi que ce soit. POURQUOI CE SCRIPT EXISTE. Le pipeline d'extraction va etre modifie (pagination, normalisation avant decoupage, largeur des colonnes). Sans une reference figee prise avant les corrections, aucun gain ne sera demontrable ensuite : on aura seulement l'intuition que "c'est mieux". Ce script fige les trois chiffres qui comptent, dans data/eval_runs/, et ne touche a rien d'autre. STRICTEMENT EN LECTURE. Aucune ecriture en base, aucun appel d'API, aucun credit consomme. Les seules ecritures sont le fichier JSON de sortie. CE QU'IL MESURE, et pourquoi chaque mesure est la : 1. Le degrade silencieux. Quand aucun motif d'article n'est reconnu, extract_articles ne leve pas : il bascule sur _extract_paragraph_chunks et fabrique des pseudo-articles PARA_1, PARA_2... La loi est publiee, son compte d'articles est non nul, et rien ne signale l'echec. Le taux de PARA_ est donc le vrai indicateur de sante, pas articles_count > 0. 2. Le gain de normalize_for_chunking(). Elle EST desormais appelee en production (app/tasks/process_law.py, avant extract_articles) — cette docstring affirmait le contraire, ce qui etait vrai a l'ecriture et ne l'est plus. On rejoue la detection de motif sur les caches deja payes, avec et sans elle, pour chiffrer ce que son cablage rapporte. ATTENTION : la metrique depend de _detect_article_pattern, qui a change depuis. Rejouee aujourd'hui sur les MEMES caches, elle rend x1,05 la ou phase0 relevait x3,52. Deux mesures ne se comparent qu'au meme commit. 3. La couverture de pages. _pages_from renvoie une seule page quand la reponse de LlamaParse est une chaine, alors que le markdown contient les vraies coupures. On compare donc les pages declarees par le cache, les pages reellement presentes dans le texte, et le nombre de pages du PDF source. Usage : DATABASE_URL=postgresql+asyncpg://jurix:jurix@localhost:5433/jurix_dev \\ python -m scripts.eval.extraction.baseline python -m scripts.eval.extraction.baseline --label phase0_avant_correctifs Author: JuriX Team """ from __future__ import annotations import argparse import json import logging import re import sys import warnings from contextlib import contextmanager from decimal import Decimal from pathlib import Path from typing import Any, Dict, List, Optional, Sequence sys.path.insert(0, ".") from sqlalchemy import text from app.core.database import SyncSessionLocal from app.utils.chunk_refiner import normalize_for_chunking from app.utils.text_chunker import _detect_article_pattern logger = logging.getLogger("eval.extraction.baseline") CACHE_DIR = Path("data/ocr_cache") UPLOADS_DIR = Path("data/uploads") RUNS_DIR = Path("data/eval_runs") # Separateur de page emis par LlamaParse dans le markdown. _pages_from l'ignore # aujourd'hui, d'ou l'ecart entre pages declarees et pages reelles. PAGE_SEPARATOR = "\n---\n" def _parse_args(argv: Optional[Sequence[str]] = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--label", default="phase0_baseline", help="nom du fichier ecrit dans data/eval_runs/") parser.add_argument("--cache-dir", type=Path, default=CACHE_DIR) parser.add_argument("--skip-db", action="store_true", help="ne mesurer que les caches (aucune connexion base)") return parser.parse_args(argv) # ==================== 1. CACHES OCR DEJA PAYES ==================== def _count_article_headers(body: str) -> int: """Nombre d'en-tetes d'article que le chunker reconnaitrait dans ce texte.""" pattern = _detect_article_pattern(body) return len(re.findall(pattern, body)) if pattern else 0 @contextmanager def _quiet_chunker(): """ Fait taire le chunker le temps de la mesure. _detect_article_pattern journalise le motif retenu a chaque appel. On l'appelle deux fois par document : sur 61 caches, cela noie le rapport sous 122 lignes sans interet ici. """ chunker_log = logging.getLogger("app.utils.text_chunker") previous = chunker_log.level chunker_log.setLevel(logging.ERROR) try: yield finally: chunker_log.setLevel(previous) def measure_cache(cache_dir: Path) -> Dict[str, Any]: """ Rejoue la detection de motif sur les caches, avec et sans normalisation. Le gain mesure ici est disponible immediatement et gratuitement : il ne demande aucun changement de moteur OCR, seulement de brancher une fonction qui existe deja. """ files = sorted(cache_dir.glob("*.json")) if not files: logger.warning("Aucun cache dans %s", cache_dir) declared_pages = real_pages = 0 headers_before = headers_after = 0 degraded_before = degraded_after = 0 with_table = 0 per_doc: List[Dict[str, Any]] = [] for path in files: try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: logger.warning("Cache illisible %s : %s", path.name, exc) continue pages = payload.get("pages") or [] body = "\n\n".join(pages) # Les vraies coupures sont dans le markdown, pas dans la liste `pages`. n_real = body.count(PAGE_SEPARATOR) + 1 if body else 0 with _quiet_chunker(): before = _count_article_headers(body) after = _count_article_headers(normalize_for_chunking(body)) declared_pages += len(pages) real_pages += n_real headers_before += before headers_after += after degraded_before += before == 0 degraded_after += after == 0 with_table += " Any: """ Rend un scalaire SQLAlchemy serialisable en JSON. round() cote Postgres renvoie un NUMERIC, que psycopg2 mappe sur Decimal, que json.dumps refuse. Convertir ici plutot qu'au moment d'ecrire evite de disperser le probleme sur chaque champ ajoute plus tard. """ return float(value) if isinstance(value, Decimal) else value def measure_db() -> Dict[str, Any]: """Compte les symptomes observables en base. Lecture seule.""" with SyncSessionLocal() as session: out: Dict[str, Any] = { key: _plain(session.execute(text(sql)).scalar()) for key, sql in _DB_QUERIES.items() } # Part des caracteres de laws.content retrouves dans les articles : ce # qui manque a ete supprime entre l'extraction et le decoupage. out["char_retention_pct"] = _plain(session.execute(text( "SELECT round(100.0 * sum(a_len) / NULLIF(sum(l_len), 0)) FROM (" " SELECT length(l.content) AS l_len," " COALESCE((SELECT sum(length(content)) FROM articles" " WHERE law_id = l.id), 0) AS a_len" " FROM laws l) t" )).scalar()) rows = session.execute(text( "SELECT l.id, l.file_id, count(a.id) AS articles," " count(DISTINCT a.page_number) AS pages_extracted " "FROM laws l LEFT JOIN articles a ON a.law_id = l.id " "GROUP BY l.id, l.file_id ORDER BY l.id" )).all() total = out["articles_total"] or 0 out["articles_para_pct"] = round(100.0 * (out["articles_para"] or 0) / total) if total else None out["page_coverage"] = _measure_page_coverage(rows) return out def _measure_page_coverage(rows: Sequence[Any]) -> Dict[str, Any]: """ Compare les pages extraites au nombre de pages du PDF source. C'est le garde-fou le moins cher du projet et il n'existe pas : un PDF de 400 pages tronque a une seule passe aujourd'hui le seul controle en place (plus de 50 caracteres) et est publie sans le moindre signal. """ try: with warnings.catch_warnings(): warnings.simplefilter("ignore") from pypdf import PdfReader except ImportError: logger.warning("pypdf absent : couverture de pages non mesuree") return {"available": False} per_law: List[Dict[str, Any]] = [] for law_id, file_id, articles, pages_extracted in rows: if not file_id: continue pdf = UPLOADS_DIR / f"{file_id}.pdf" if not pdf.exists(): continue try: with warnings.catch_warnings(): warnings.simplefilter("ignore") pdf_pages = len(PdfReader(str(pdf)).pages) except Exception as exc: # PDF corrompu, chiffre, tronque logger.warning("PDF illisible pour la loi %s : %s", law_id, exc) continue per_law.append({ "law_id": law_id, "pdf_pages": pdf_pages, "pages_extracted": pages_extracted, "articles": articles, "coverage": round(pages_extracted / pdf_pages, 3) if pdf_pages else None, }) measured = [r for r in per_law if r["coverage"] is not None] below = [r for r in measured if r["coverage"] < 0.95] return { "available": True, "laws_measured": len(measured), "pdf_pages_total": sum(r["pdf_pages"] for r in measured), "pages_extracted_total": sum(r["pages_extracted"] for r in measured), "mean_coverage": round( sum(r["coverage"] for r in measured) / len(measured), 3 ) if measured else None, "laws_below_95pct": len(below), "per_law": per_law, } # ==================== 3. RAPPORT ==================== def _print_report(cache: Dict[str, Any], db: Optional[Dict[str, Any]]) -> None: print() print("=" * 68) print(" REFERENCE D'EXTRACTION - PHASE 0 (avant tout correctif)") print("=" * 68) print(f"\nCaches OCR deja payes ({cache['documents']} documents)") print(f" pages declarees / reelles {cache['declared_pages']} / {cache['real_pages']}") if cache["pages_lost_ratio"] is not None: print(f" structure de pages perdue {cache['pages_lost_ratio']:.1%}") print(f" en-tetes avant normalisation {cache['article_headers_before_normalize']}") print(f" en-tetes apres normalisation {cache['article_headers_after_normalize']}" f" (x{cache['headers_gain_factor']})") print(f" documents sans motif detecte {cache['degraded_docs_before']}" f" -> {cache['degraded_docs_after']} sur {cache['documents']}") print(f" documents avec tableau {cache['docs_with_table']}") if db is None: print("\n(base non mesuree : --skip-db)") return print("\nBase de donnees") print(f" lois {db['laws_total']}" f" (publiees {db['laws_published']}, refusees {db['laws_refused']})") print(f" articles {db['articles_total']}") print(f" dont pseudo-articles PARA_ {db['articles_para']}" f" ({db['articles_para_pct']}%) sur {db['laws_degraded']} lois") print(f" dont FULL_TEXT / LEGAL_BASIS {db['articles_full_text']} / {db['articles_legal_basis']}") print(f" page_number = 1 {db['articles_page_1']} / {db['articles_total']}") print(f" caracteres conserves {db['char_retention_pct']}%") print(f" lois publiees sans article {db['laws_published_without_article']} (doit valoir 0)") cov = db["page_coverage"] if cov.get("available"): print(f"\nCouverture de pages ({cov['laws_measured']} lois mesurees)") print(f" pages PDF / pages extraites {cov['pdf_pages_total']} / {cov['pages_extracted_total']}") print(f" couverture moyenne {cov['mean_coverage']}") print(f" lois sous 95 % {cov['laws_below_95pct']} / {cov['laws_measured']}") print() def main(argv: Optional[Sequence[str]] = None) -> int: logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") args = _parse_args(argv) cache = measure_cache(args.cache_dir) db = None if args.skip_db else measure_db() _print_report(cache, db) RUNS_DIR.mkdir(parents=True, exist_ok=True) out = RUNS_DIR / f"{args.label}.json" out.write_text(json.dumps({ "label": args.label, "phase": 0, "note": "Reference figee AVANT les correctifs de pipeline. Ne pas regenerer apres.", # Meme convention que scripts/eval/run_eval.py : les chiffres ne veulent # rien dire hors de l'etat de corpus qui les a produits. "corpus_snapshot": { "alembic_head": db["alembic_head"] if db else None, "articles": db["articles_total"] if db else None, }, "cache": cache, "database": db, }, ensure_ascii=False, indent=2), encoding="utf-8") logger.info("Reference ecrite dans %s", out) return 0 if __name__ == "__main__": sys.exit(main())