feat(extraction): remplacer LlamaParse par Gemini multimodal

La cle LlamaParse ayant ete retiree, l'ingestion de tout nouveau PDF levait et
marquait le document `refused` — le bon echec, mais l'ingestion etait a l'arret.

LE MOTIF DU CHANGEMENT EST LA PAGINATION, PAS LE PRIX. LlamaParse rend le
document en un bloc sans signaler ses coupures de page ; `_split_markdown_pages`
les reconstruisait en coupant sur les lignes de trois tirets, et sa propre
docstring admettait la limite : « la pagination ne devient exacte qu'avec un
moteur qui renvoie un numero de page explicite ». Mesure en base : les 710
articles portent tous `page_number = 1`, et sur les 27 lois, 271 pages de PDF ne
donnent que 27 pages extraites — couverture moyenne 0,60. Un lien `?article=35`
ouvre donc le PDF a la premiere page.

MESURE AVANT D'ECRIRE, sur le Code Minier dont la sortie LlamaParse est en base :

densite 2352 car/page contre 2405 — equivalent, rien n'est perdu
marqueurs de page 1 par page contre 1 pour 69 pages
lot pages 21 a 32 numerote 21..32 exactement

Le dernier point n'est acquis que si la consigne annonce au modele le numero de
la premiere page fournie. Sans elle il renumerote chaque lot a partir de 1, et
chaque lot ecrase la pagination du precedent.

UN MODE D'ECHEC PROPRE A CE MOTEUR. Gemini rend parfois
`finish_reason=RECITATION` et refuse de transcrire une page d'acte officiel dont
la forme lui est familiere. Mesure sur sept documents : une page de garde
refusee, soit ~14 %. Ce n'est pas transitoire — trois formulations de consigne
essayees, toutes refusees — mais c'est PAR PAGE : dans le document mesure, la
page 1 refuse et les pages 2 et 3 s'extraient. Le lot est donc rejoue page par
page, seule la page refusee est perdue, et elle est nommee dans
`processing_error`. Perdre une page de garde vaut mieux que perdre les
soixante-huit autres.

CE QUI EST CONSERVE. Une centaine de lignes de `llama_parse_service.py` ne
dependaient d'aucun fournisseur et avaient ete calibrees sur ce corpus : le
retrait du cachet « COPIE CERTIFIEE CONFORME » avec sa regle des deux
occurrences — qui epargne un en-tete « PRESIDENCE DE LA REPUBLIQUE » legitime —,
le filigrane `www.prc.cm` eclate caractere par caractere, et le retrait de
`<sup>` sans jamais toucher `<table>`. Elles vivent dans
`app/utils/markdown_cleanup.py` avec leurs six tests.

N'est PAS conserve `_split_markdown_pages` : reconstruire par heuristique ce que
le nouveau moteur donne exactement reviendrait a deviner ce qu'on connait.

Le cache disque est repris tel quel — indexe par sha256 du fichier, ecriture
atomique, une extraction vide n'est jamais gravee — avec `tier` devenu `engine`.
Les 61 entrees LlamaParse ne sont donc pas relues, ce qui est correct : elles
contiennent la sortie non paginee qu'on remplace. Elles ne sont pas detruites.

Un defaut recopie de l'ancien service puis corrige : `api_key=""` passe
explicitement retombait sur le reglage global, rendant intestable le cas du
service non configure.

Quota : le palier gratuit plafonne a 20 generations par jour. Les 27 lois en
base coutent 35 appels, soit deux jours. `reextract_corpus.py` est reprenable
par construction — le cache par sha256 fait qu'un document deja extrait n'est
jamais repaye — s'arrete proprement sur 429 et sort en code 2.

Tests : 725 -> 752, dont 21 sur le nouveau service et 6 migres avec le
nettoyage. Le reseau est simule par monkeypatch et non par respx : le SDK
google-genai a sa propre pile, qu'un mock respx ne verrait pas.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

.env.example

@@ -23,14 +23,13 @@ GEMINI_API_KEY=your-gemini-api-key-here
 GEMINI_MODEL=gemini-3.8-flash
 GEMINI_EMBEDDING_MODEL=models/gemini-embedding-001
 
-# ---- LlamaParse (LlamaCloud) — extraction OCR des PDF scannes ----
-# Cle a creer sur https://cloud.llamaindex.ai
-LLAMA_CLOUD_API_KEY=llx-your-llama-parse-key-here
-# Tier : fast(1cr/page) | cost_effective(3cr) | agentic(10cr) | agentic_plus(45cr)
-# 1000 credits = 1,25 $. cost_effective mesure a 100% de rappel sur le corpus prc.cm ;
-# fast relit la couche texte existante (50% seulement) — ne pas l'utiliser ici.
-LLAMA_PARSE_TIER=cost_effective
-# Cache OCR par sha256 : un fichier deja extrait n'est jamais repaye
+# ---- Extraction des PDF ----
+# L'extraction passe par Gemini multimodal : meme cle que la generation.
+# Pages envoyees par appel. La limite du modele est de 65 536 jetons EN SORTIE ;
+# a ~2400 caracteres par page, 20 pages en produisent ~13 000. Le palier gratuit
+# plafonne a 20 appels par jour : agrandir le lot economise des appels.
+# PDF_EXTRACTION_PAGES_PER_CALL=20
+# Cache d'extraction par sha256 : un fichier deja traite n'est jamais repaye
 OCR_CACHE_DIR=./data/ocr_cache
 
 # ---- Securite / JWT ----

README.md

@@ -16,7 +16,7 @@ article, réponses citées via Gemini.
 | Recherche sémantique | `pgvector` + index HNSW, embeddings `gemini-embedding-001` tronqués à 1536 dim. |
 | Cache | tables `query_cache` et `embedding_cache` |
 | LLM | Gemini (`google-genai`) |
-| OCR | LlamaParse v2 |
+| Extraction PDF | Gemini multimodal |
 | Tâches de fond | `BackgroundTasks` FastAPI |
 
 Pas de Redis, pas de Meilisearch, pas de Celery : recherche, cache et files
@@ -82,7 +82,7 @@ Toutes les variables de `.env.example` sont réellement lues par
 |---|---|
 | `DATABASE_URL` | `postgresql+asyncpg://…` — le driver asyncpg est obligatoire |
 | `GEMINI_API_KEY` / `GEMINI_MODEL` | LLM des réponses. Modèles Flash actuels : `gemini-3.8-flash`, `3.7`, `3.6`, `3.5` |
-| `LLAMA_CLOUD_API_KEY` / `LLAMA_PARSE_TIER` | OCR. `cost_effective` est le tier recommandé |
+| `PDF_EXTRACTION_PAGES_PER_CALL` | Pages envoyées par appel d'extraction (défaut 20) |
 | `SECRET_KEY` | Signature JWT. **Obligatoire hors développement** : l'application refuse de démarrer si la valeur du dépôt est conservée |
 | `ADMIN_EMAIL` / `ADMIN_PASSWORD` | Valeurs par défaut de `scripts/create_admin.py` |
 
@@ -103,7 +103,7 @@ créés par `POST /api/v1/admin/users` ou par `scripts/create_admin.py`.
 ## Chaîne d'ingestion
 
 ```
-upload → LlamaParse → normalisation → découpage par article
+upload → extraction Gemini → normalisation → découpage par article
        → embeddings → tsvector (trigger) → published
 ```
 
@@ -111,14 +111,28 @@ Un document est créé en `processing`. Il passe à `published` en cas de succè
 à `refused` avec `processing_error` en cas d'échec — jamais de contenu
 partiellement extrait dans le corpus public.
 
-**Aucun repli dégradé.** LlamaParse réessaie 4 fois avec un délai croissant ;
-au-delà, l'ingestion échoue. La couche texte des PDF scannés du corpus a été
-mesurée à environ 20 % de rappel (filigrane inséré au milieu des phrases,
-cachets lus comme du charabia, un document sur cinq sans aucun texte
-exploitable) : la publier serait pire que de ne rien publier.
-
-Les extractions sont mises en cache par `sha256` : un fichier déjà traité n'est
-jamais repayé.
+**Aucun repli dégradé.** L'extraction réessaie trois fois sur une saturation du
+fournisseur (503), jamais sur un quota épuisé (429) ; au-delà, l'ingestion
+échoue. La couche texte des PDF scannés du corpus a été mesurée à environ 20 %
+de rappel (filigrane inséré au milieu des phrases, cachets lus comme du
+charabia, un document sur cinq sans aucun texte exploitable) : la publier serait
+pire que de ne rien publier.
+
+**Pagination.** Gemini rend un marqueur `<<PAGE:n>>` par page, ce que LlamaParse
+ne faisait pas — d'où des articles portant tous `page_number = 1`. C'est la
+raison du changement de moteur, pas le prix.
+
+**Refus de transcription.** Le modèle rend parfois `finish_reason=RECITATION`
+sur une page d'acte officiel dont la forme lui est familière : mesuré à une page
+de garde sur sept documents. Le refus est **par page** — le lot est alors rejoué
+page par page, et seule la page refusée est perdue, signalée dans
+`processing_error`. Reformuler la consigne n'y change rien (trois variantes
+essayées), réduire le lot non plus.
+
+**Quota.** Le palier gratuit plafonne à 20 appels de génération par jour et par
+modèle. Un document d'une page coûte un appel ; les 72 PDF locaux (481 pages) en
+coûtent 83. Le script de ré-extraction est reprenable : le cache `sha256` fait
+qu'un fichier déjà traité n'est jamais repayé.
 
 ## Découpage
 

app/core/config.py

@@ -71,13 +71,17 @@ class Settings(BaseSettings):
     TEXT_WEIGHT: float = 0.4
     SEMANTIC_WEIGHT: float = 0.6
 
-    # LlamaParse (PDF extraction OCR)
-    LLAMA_CLOUD_API_KEY: str = ""
-    # Tier de parsing : fast(1cr) | cost_effective(3cr) | agentic(10cr) | agentic_plus(45cr)
-    # cost_effective est le meilleur rapport qualite/prix mesure sur le corpus prc.cm
-    LLAMA_PARSE_TIER: str = "cost_effective"
-    # Cache OCR par sha256 — evite de repayer l'extraction d'un fichier deja traite
+    # LLAMA_CLOUD_API_KEY et LLAMA_PARSE_TIER ont ete retires : l'extraction
+    # passe par Gemini (app/services/pdf_extraction_service.py), qui rend un
+    # numero de page explicite la ou LlamaParse n'en donnait aucun.
+    # Cache d'extraction par sha256 — evite de repayer un fichier deja traite
     OCR_CACHE_DIR: str = "./data/ocr_cache"
+    # Pages envoyees par appel a Gemini. La limite du modele est de 65 536
+    # jetons EN SORTIE ; a ~2400 caracteres par page, 20 pages produisent
+    # ~13 000 jetons, avec de la marge pour la reflexion interne. Le palier
+    # gratuit plafonnant a 20 appels par jour, agrandir le lot economise des
+    # appels — au risque de tronquer si les pages sont denses.
+    PDF_EXTRACTION_PAGES_PER_CALL: int = 20
 
     # CORS — comma-separated list of extra allowed origins for production
     # Example: https://jurix.vercel.app,https://www.jurix.cm

app/services/llama_parse_service.py

@@ -1,594 +0,0 @@
-"""
-LlamaParseService - Extraction OCR des documents juridiques via LlamaParse v2.
-
-Remplace l'extraction par couche texte (pdfplumber/pypdf), inexploitable sur le
-corpus prc.cm : 21% des documents n'ont aucune couche texte utilisable, et les
-79% restants sont pollués par le filigrane www.prc.cm et les cachets officiels.
-
-Banc d'essai sur 6 pages representatives (verite terrain relevee visuellement) :
-
-    moteur                      prose    tableau budgetaire
-    couche texte actuelle        20%           0/9
-    tesseract 5.5 local          74%           0/9
-    mistral OCR                 100%           8/9   (chiffre faux, reproductible)
-    llamaparse cost_effective   100%           9/9
-
-Architecture :
-- API REST v2 : POST /parse/upload -> poll GET /parse/{id} -> markdown
-- Cache disque par sha256 : une page n'est jamais payee deux fois
-- Retry exponentiel : 1 echec reseau observe sur 6 appels au banc
-- Decoupe automatique des PDF volumineux (5 fichiers >50 Mo dans le corpus)
-- Marqueurs <<PAGE:n>> pour que text_chunker renseigne Article.page_number
-- Nettoyage des cachets officiels (regle 7 du chunking)
-
-Tiers disponibles (credits/page, 1000 credits = 1,25 $) :
-    fast            1    relit la couche texte -> 50% de rappel, inutilisable ici
-    cost_effective  3    recommande : 100% de rappel, 9/9 sur les tableaux
-    agentic        10    aucun gain mesure sur ce corpus
-    agentic_plus   45    aucun gain mesure sur ce corpus
-
-Author: JuriX Team
-"""
-
-import asyncio
-import hashlib
-import json
-import logging
-import re
-import time
-from functools import lru_cache
-from pathlib import Path
-from typing import Any, Dict, List, Optional
-
-import httpx
-
-from app.core.config import settings
-
-logger = logging.getLogger(__name__)
-
-
-class LlamaParseError(Exception):
-    """Erreur d'extraction LlamaParse."""
-    pass
-
-
-# ==================== NETTOYAGE DES CACHETS ====================
-
-# Vocabulaire du cachet "COPIE CERTIFIEE CONFORME" appose sur chaque page du
-# corpus prc.cm. LlamaParse le restitue en texte (contrairement a Mistral OCR
-# qui l'ignore), il faut donc le retirer avant chunking sinon il pollue le FTS
-# et les embeddings de chaque document.
-_STAMP_VOCAB = re.compile(
-    r"(copie\s+certifi\w*\s+conforme"
-    r"|certified\s+true\s+copy"
-    r"|service\s+du\s+fichier\s+l[ée]gislatif"
-    r"|legislative\s+and\s+statutory\s+affairs"
-    r"|presidency\s+of\s+the\s+republic"
-    r"|secr[ée]tariat[\s-]+g[ée]n[ée]ral"
-    r"|pr[ée]sidence\s+de\s+la\s+r[ée]publique)",
-    re.IGNORECASE,
-)
-
-# Fragments du filigrane diagonal "www.prc.cm" eclate par l'OCR
-_WATERMARK = re.compile(
-    r"^\s*(?:w\s*){1,3}$|^\s*\.?\s*(?:p\s*r\s*c|c\s*m|prc\.cm|www\.prc\.cm)\s*\.?\s*$",
-    re.IGNORECASE,
-)
-
-# Balisage HTML inline emis par LlamaParse (les <table> sont conserves : la
-# structure tabulaire porte du sens, cf. les annexes budgetaires).
-# <sup>/<sub> sont critiques : "ARTICLE 1<sup>ER</sup>" doit redevenir
-# "ARTICLE 1ER" pour que ARTICLE_PATTERNS de text_chunker le reconnaisse.
-_INLINE_TAGS = re.compile(
-    r"</?(?:u|mark|b|i|em|strong|span|sup|sub|small)\b[^>]*>", re.IGNORECASE
-)
-
-
-def _strip_stamp_blocks(text: str) -> str:
-    """
-    Retire les blocs de cachet officiel du markdown.
-
-    Un bloc = au moins 2 lignes consecutives (blancs autorises) appartenant au
-    vocabulaire du cachet. Le seuil de 2 evite de supprimer un en-tete legitime :
-    beaucoup de documents portent "PRESIDENCE DE LA REPUBLIQUE" comme autorite
-    emettrice, ce qui est du contenu, pas du tampon.
-
-    Args:
-        text: Markdown brut renvoye par LlamaParse
-
-    Returns:
-        Markdown nettoye
-    """
-    lines = text.splitlines()
-    drop = [False] * len(lines)
-    i = 0
-
-    while i < len(lines):
-        if not _STAMP_VOCAB.search(lines[i]):
-            i += 1
-            continue
-
-        # Cas 1 : cachet entier restitue sur une seule ligne, par ex.
-        # "[signature: PRESIDENCE ... COPIE CERTIFIEE CONFORME CERTIFIED TRUE COPY]"
-        # ou "stamp: ..." / "logo: ...". Deux phrases distinctes du vocabulaire sur
-        # la meme ligne ne peuvent pas etre du texte juridique legitime.
-        if len(set(m.group(0).lower() for m in _STAMP_VOCAB.finditer(lines[i]))) >= 2:
-            drop[i] = True
-            i += 1
-            continue
-
-        # Etendre le bloc tant qu'on reste dans le vocabulaire du cachet
-        j = i
-        hits = 0
-        while j < len(lines):
-            stripped = lines[j].strip(" >*|-\t")
-            if _STAMP_VOCAB.search(lines[j]):
-                hits += 1
-                j += 1
-            elif not stripped:
-                j += 1
-            else:
-                break
-
-        if hits >= 2:
-            for k in range(i, j):
-                drop[k] = True
-        i = max(j, i + 1)
-
-    kept = [ln for ln, d in zip(lines, drop) if not d]
-    kept = [ln for ln in kept if not _WATERMARK.match(ln)]
-    cleaned = "\n".join(kept)
-    cleaned = _INLINE_TAGS.sub("", cleaned)
-    return re.sub(r"\n{4,}", "\n\n\n", cleaned).strip()
-
-
-# ==================== DECOUPE EN PAGES ====================
-
-# Saut de page emis par LlamaParse quand la reponse arrive en un seul bloc :
-# une ligne contenant EXACTEMENT trois tirets.
-#
-# Les trois contraintes sont mesurees, pas supposees (64 documents apparies a
-# leur PDF, 473 pages de verite terrain) :
-#   - ligne entiere         : `|---|---|`, separateur de tableau markdown, ne
-#                             peut pas declencher de coupure.
-#   - exactement trois      : `-{3,}` donnait 10 sur-decoupes et 44/64 exacts ;
-#                             `-{3}` en donne 3 et 51/64. Les suites plus
-#                             longues sont du contenu (soulignements, bordures
-#                             ASCII), jamais des sauts de page.
-_PAGE_BREAK = re.compile(r"\n-{3}[ \t]*(?=\n)")
-
-# Frontmatter YAML en tete de document. Ses deux delimiteurs sont des lignes de
-# trois tirets : sans ce retrait prealable, le delimiteur FERMANT serait pris
-# pour un saut de page et couperait le document en deux.
-_FRONTMATTER = re.compile(r"\A---[ \t]*\n.*?\n---[ \t]*(?=\n|\Z)", re.DOTALL)
-
-
-def _split_markdown_pages(markdown: str) -> List[str]:
-    """
-    Restitue la structure de pages d'un markdown recu en un seul bloc.
-
-    POURQUOI. `_pages_from` renvoyait `[md]` des que la reponse etait une
-    chaine, ce qui ecrasait la pagination de TOUT le document : les 849 articles
-    en base portaient tous `page_number = 1`. Les vraies coupures etaient
-    pourtant presentes dans le texte, simplement jamais exploitees.
-
-    RESULTAT MESURE sur 64 documents apparies a leur PDF (473 pages reelles) :
-    64 pages restituees avant, 411 apres, dont 51/64 documents au compte exact
-    et 58/64 a une page pres. Les 10 documents restants sont sous-decoupes
-    parce que LlamaParse n'a emis aucun separateur pour ces pages-la : c'est une
-    reconstruction heuristique, pas une source d'autorite. La pagination ne
-    devient exacte qu'avec un moteur qui renvoie un numero de page explicite.
-
-    Les pages vides sont CONSERVEES : `extract_text` numerote avant de filtrer
-    (`enumerate(..., start=1)` puis `if md.strip()`), donc supprimer une page
-    blanche ici decalerait tous les numeros suivants.
-
-    Repli prudent : si la coupure ne produit pas au moins deux pages non vides,
-    le document est rendu tel quel. Un texte sans saut de page reste donc une
-    page unique, comme avant.
-    """
-    # Le frontmatter est mis de cote AVANT la coupure : son delimiteur fermant
-    # est lui aussi une ligne de trois tirets, et couperait la premiere page.
-    front = _FRONTMATTER.match(markdown)
-    head, body = (front.group(0), markdown[front.end():]) if front else ("", markdown)
-
-    parts = _PAGE_BREAK.split(body)
-    if sum(1 for p in parts if p.strip()) < 2:
-        return [markdown]
-
-    # Un saut en tete ou en pied de document produit une page vide parasite,
-    # qui n'en est pas une. Les pages vides INTERIEURES, elles, sont gardees.
-    while parts and not parts[0].strip():
-        parts.pop(0)
-    while parts and not parts[-1].strip():
-        parts.pop()
-
-    pages = [p.strip("\n") for p in parts]
-    if head and pages:
-        pages[0] = f"{head}\n{pages[0]}".strip("\n")
-    return pages
-
-
-class LlamaParseService:
-    """
-    Client LlamaParse v2 pour l'extraction des PDF juridiques scannes.
-
-    Usage:
-        >>> service = LlamaParseService()
-        >>> if service.is_available():
-        ...     text = await service.extract_text(Path("decret.pdf"))
-    """
-
-    BASE_URL = "https://api.cloud.llamaindex.ai/api/v2/parse"
-    DEFAULT_TIER = "cost_effective"
-
-    # Decoupe defensive : seuil conservateur, en-deca de la limite annoncee.
-    # Le corpus contient 5 fichiers >50 Mo (max observe : 124,6 Mo / 48 pages).
-    # Version du format d'entree de cache. A incrementer des qu'un changement
-    # rend les entrees existantes trompeuses plutot que simplement vieilles.
-    # v2 : `_pages_from` decoupe desormais un markdown recu en un seul bloc ;
-    # les entrees v1 rendaient tout le document sur une page unique.
-    CACHE_SCHEMA = 2
-
-    MAX_FILE_MB = 45
-    SPLIT_PAGES = 100
-
-    MAX_RETRIES = 4
-    RETRY_BASE_DELAY = 3.0
-    POLL_INTERVAL = 3.0
-    POLL_TIMEOUT = 900.0
-    UPLOAD_TIMEOUT = 300.0
-
-    def __init__(
-        self,
-        api_key: Optional[str] = None,
-        tier: Optional[str] = None,
-        cache_dir: Optional[Path] = None,
-    ):
-        """
-        Initialise le service.
-
-        Args:
-            api_key: Cle LlamaCloud (defaut: settings.LLAMA_CLOUD_API_KEY)
-            tier: Tier de parsing (defaut: settings.LLAMA_PARSE_TIER)
-            cache_dir: Repertoire du cache sha256 (defaut: ./data/ocr_cache)
-        """
-        self.api_key = api_key or settings.LLAMA_CLOUD_API_KEY
-        self.tier = tier or getattr(settings, "LLAMA_PARSE_TIER", self.DEFAULT_TIER)
-
-        self.cache_dir = Path(
-            cache_dir or getattr(settings, "OCR_CACHE_DIR", "./data/ocr_cache")
-        )
-        try:
-            self.cache_dir.mkdir(parents=True, exist_ok=True)
-        except OSError as e:
-            logger.warning(f"⚠️ Cache OCR indisponible ({e}), desactive")
-            self.cache_dir = None
-
-        if self.api_key:
-            logger.info(f"✅ LlamaParseService pret (tier={self.tier})")
-        else:
-            logger.warning("⚠️ LLAMA_CLOUD_API_KEY absente — LlamaParse desactive")
-
-    # ==================== API PUBLIQUE ====================
-
-    def is_available(self) -> bool:
-        """Le service est-il configure ?"""
-        return bool(self.api_key)
-
-    async def extract_text(self, file_path: Path) -> str:
-        """
-        Extrait le texte complet d'un PDF, avec marqueurs de page.
-
-        Point d'entree utilise par app/tasks/process_law.py.
-
-        Args:
-            file_path: Chemin du PDF
-
-        Returns:
-            Markdown nettoye, pages separees par <<PAGE:n>>
-
-        Raises:
-            LlamaParseError: Si l'extraction echoue apres tous les retries
-        """
-        pages = await self.extract_pages(file_path)
-        if not pages:
-            raise LlamaParseError(f"Aucune page extraite de {file_path.name}")
-
-        parts = [f"<<PAGE:{i}>>\n{md}" for i, md in enumerate(pages, start=1) if md.strip()]
-        text = "\n\n".join(parts)
-        logger.info(
-            f"✅ LlamaParse: {len(pages)} page(s), {len(text)} caracteres "
-            f"depuis {file_path.name}"
-        )
-        return text
-
-    async def extract_pages(self, file_path: Path) -> List[str]:
-        """
-        Extrait le markdown page par page (cache-aware).
-
-        Args:
-            file_path: Chemin du PDF
-
-        Returns:
-            Liste de markdown, un element par page, dans l'ordre
-        """
-        assert file_path is not None, "file_path requis"
-        if not self.is_available():
-            raise LlamaParseError("LLAMA_CLOUD_API_KEY non configuree")
-        if not file_path.exists():
-            raise LlamaParseError(f"Fichier introuvable: {file_path}")
-
-        cache_key = self._sha256(file_path)
-        cached = self._read_cache(cache_key)
-        if cached is not None:
-            logger.info(f"🎯 Cache OCR HIT: {file_path.name} ({len(cached)} pages)")
-            return cached
-
-        chunks = self._split_if_needed(file_path)
-        pages: List[str] = []
-        try:
-            for idx, chunk_path in enumerate(chunks, start=1):
-                if len(chunks) > 1:
-                    logger.info(f"📄 Segment {idx}/{len(chunks)}: {chunk_path.name}")
-                pages.extend(await self._parse_one(chunk_path))
-        finally:
-            for c in chunks:
-                if c != file_path:
-                    c.unlink(missing_ok=True)
-
-        pages = [_strip_stamp_blocks(p) for p in pages]
-        self._write_cache(cache_key, pages)
-        return pages
-
-    async def health_check(self) -> Dict[str, Any]:
-        """Verifie la configuration et la joignabilite de l'API."""
-        if not self.is_available():
-            return {"service": "LlamaParse", "status": "unconfigured", "tier": self.tier}
-        try:
-            async with httpx.AsyncClient(timeout=15.0) as client:
-                r = await client.get(
-                    f"{self.BASE_URL}/00000000-0000-0000-0000-000000000000",
-                    headers=self._headers(),
-                )
-            reachable = r.status_code in (200, 401, 403, 404, 422)
-            return {
-                "service": "LlamaParse",
-                "status": "healthy" if reachable else "degraded",
-                "tier": self.tier,
-                "cache_dir": str(self.cache_dir) if self.cache_dir else None,
-            }
-        except Exception as e:
-            return {"service": "LlamaParse", "status": "unreachable", "error": str(e)}
-
-    # ==================== APPEL API ====================
-
-    def _headers(self) -> Dict[str, str]:
-        return {"Authorization": f"Bearer {self.api_key}"}
-
-    async def _parse_one(self, file_path: Path) -> List[str]:
-        """Soumet un fichier, attend le job, renvoie le markdown par page."""
-        last_error: Optional[Exception] = None
-
-        for attempt in range(1, self.MAX_RETRIES + 1):
-            try:
-                job_id = await self._upload(file_path)
-                return await self._poll(job_id)
-            except LlamaParseError:
-                raise
-            except Exception as e:
-                last_error = e
-                if attempt < self.MAX_RETRIES:
-                    delay = self.RETRY_BASE_DELAY * (2 ** (attempt - 1))
-                    logger.warning(
-                        f"⚠️ Tentative {attempt}/{self.MAX_RETRIES} echouee "
-                        f"({type(e).__name__}: {e}), retry dans {delay:.0f}s"
-                    )
-                    await asyncio.sleep(delay)
-
-        raise LlamaParseError(
-            f"Echec apres {self.MAX_RETRIES} tentatives sur {file_path.name}: {last_error}"
-        )
-
-    async def _upload(self, file_path: Path) -> str:
-        """Soumet le fichier et renvoie l'id du job."""
-        config = json.dumps({"tier": self.tier, "version": "latest"})
-
-        async with httpx.AsyncClient(timeout=self.UPLOAD_TIMEOUT) as client:
-            with file_path.open("rb") as fh:
-                resp = await client.post(
-                    f"{self.BASE_URL}/upload",
-                    headers=self._headers(),
-                    files={"file": (file_path.name, fh, "application/pdf")},
-                    data={"configuration": config},
-                )
-
-        if resp.status_code in (401, 403):
-            raise LlamaParseError(f"Cle LlamaCloud refusee (HTTP {resp.status_code})")
-        if resp.status_code not in (200, 201):
-            raise RuntimeError(f"upload HTTP {resp.status_code}: {resp.text[:300]}")
-
-        payload = resp.json()
-        job_id = (payload.get("job") or {}).get("id") or payload.get("id")
-        if not job_id:
-            raise RuntimeError(f"Pas de job id dans la reponse: {str(payload)[:300]}")
-        return job_id
-
-    async def _poll(self, job_id: str) -> List[str]:
-        """Interroge le job jusqu'a completion, renvoie le markdown par page."""
-        started = time.monotonic()
-
-        async with httpx.AsyncClient(timeout=120.0) as client:
-            while True:
-                if time.monotonic() - started > self.POLL_TIMEOUT:
-                    raise LlamaParseError(
-                        f"Timeout ({self.POLL_TIMEOUT:.0f}s) sur le job {job_id}"
-                    )
-                await asyncio.sleep(self.POLL_INTERVAL)
-
-                resp = await client.get(
-                    f"{self.BASE_URL}/{job_id}",
-                    headers=self._headers(),
-                    params={"expand": "markdown,markdown_full"},
-                )
-                if resp.status_code != 200:
-                    raise RuntimeError(f"poll HTTP {resp.status_code}: {resp.text[:300]}")
-
-                data = resp.json()
-                job = data.get("job") or {}
-                status = job.get("status") or data.get("status")
-
-                if status in ("COMPLETED", "SUCCESS"):
-                    return self._pages_from(data)
-                if status in ("FAILED", "CANCELLED", "ERROR"):
-                    raise LlamaParseError(
-                        f"Job {status}: {job.get('error_message') or 'raison inconnue'}"
-                    )
-
-    @staticmethod
-    def _pages_from(data: Dict[str, Any]) -> List[str]:
-        """Normalise la reponse v2 en liste de markdown par page."""
-        md = data.get("markdown")
-
-        if isinstance(md, list):
-            return [
-                (p.get("markdown", "") if isinstance(p, dict) else str(p)) for p in md
-            ]
-        if isinstance(md, str) and md.strip():
-            return _split_markdown_pages(md)
-
-        pages = data.get("pages")
-        if isinstance(pages, list):
-            return [
-                (p.get("markdown", "") if isinstance(p, dict) else str(p)) for p in pages
-            ]
-
-        full = data.get("markdown_full")
-        return _split_markdown_pages(full) if isinstance(full, str) and full.strip() else []
-
-    # ==================== DECOUPE DES GROS FICHIERS ====================
-
-    def _split_if_needed(self, file_path: Path) -> List[Path]:
-        """
-        Decoupe un PDF trop volumineux en segments temporaires.
-
-        Returns:
-            [file_path] si aucune decoupe n'est necessaire, sinon les segments.
-        """
-        size_mb = file_path.stat().st_size / 1_000_000
-        if size_mb <= self.MAX_FILE_MB:
-            return [file_path]
-
-        try:
-            from pypdf import PdfReader, PdfWriter
-        except ImportError:
-            logger.warning(
-                f"⚠️ pypdf absent, {file_path.name} ({size_mb:.0f} Mo) envoye tel quel"
-            )
-            return [file_path]
-
-        reader = PdfReader(str(file_path))
-        total = len(reader.pages)
-        logger.info(
-            f"✂️ {file_path.name}: {size_mb:.0f} Mo / {total} pages -> "
-            f"segments de {self.SPLIT_PAGES} pages"
-        )
-
-        parts: List[Path] = []
-        for start in range(0, total, self.SPLIT_PAGES):
-            writer = PdfWriter()
-            for page in reader.pages[start:start + self.SPLIT_PAGES]:
-                writer.add_page(page)
-            part = file_path.with_name(f"{file_path.stem}__part{start // self.SPLIT_PAGES}.pdf")
-            with part.open("wb") as fh:
-                writer.write(fh)
-            parts.append(part)
-
-        return parts
-
-    # ==================== CACHE SHA256 ====================
-
-    @staticmethod
-    def _sha256(file_path: Path) -> str:
-        h = hashlib.sha256()
-        with file_path.open("rb") as fh:
-            for block in iter(lambda: fh.read(1024 * 1024), b""):
-                h.update(block)
-        return h.hexdigest()
-
-    def _cache_path(self, key: str) -> Optional[Path]:
-        return self.cache_dir / f"{key}.json" if self.cache_dir else None
-
-    def _read_cache(self, key: str) -> Optional[List[str]]:
-        path = self._cache_path(key)
-        if not path or not path.exists():
-            return None
-        try:
-            payload = json.loads(path.read_text(encoding="utf-8"))
-            if payload.get("tier") != self.tier:
-                return None
-
-            pages = payload.get("pages")
-
-            # Entree ecrite avant le decoupage en pages : elle rend tout le
-            # document sur une seule page. Son CONTENU reste valide et a deja
-            # ete paye — on le redecoupe sur place plutot que de rappeler l'API.
-            # Re-extraire aurait coute des credits pour un texte qu'on possede
-            # deja.
-            if payload.get("schema") != self.CACHE_SCHEMA and pages:
-                pages = _split_markdown_pages("\n\n".join(pages))
-                logger.info(f"♻️ Cache OCR v1 redecoupe sur place : {len(pages)} page(s)")
-                self._write_cache(key, pages)
-
-            # Une liste VIDE etait un HIT : une extraction ratee se gravait sur
-            # disque et se rejouait indefiniment, sans jamais rappeler l'API. Le
-            # seul moyen d'en sortir etait d'effacer le fichier a la main.
-            if not pages:
-                logger.warning("⚠️ Entree de cache OCR vide, ignoree")
-                return None
-            return pages
-        except Exception as e:
-            logger.warning(f"⚠️ Cache OCR illisible ({e}), ignore")
-            return None
-
-    def _write_cache(self, key: str, pages: List[str]) -> None:
-        path = self._cache_path(key)
-        if not path:
-            return
-
-        # Ne jamais graver un echec. Sans ce garde-fou, `pages == []` — ce que
-        # `_pages_from` rend des que la reponse n'a aucune forme reconnue, y
-        # compris sur un job COMPLETED au markdown nul — devenait un resultat
-        # permanent.
-        if not any(p.strip() for p in pages):
-            logger.warning("⚠️ Extraction vide, non mise en cache")
-            return
-
-        try:
-            # Ecriture atomique : un plantage en cours d'ecriture laissait un
-            # JSON tronque. Il etait relu comme un miss, donc non fatal, mais il
-            # occupait la place de l'entree valide.
-            tmp = path.with_suffix(".json.tmp")
-            tmp.write_text(
-                json.dumps(
-                    {
-                        "schema": self.CACHE_SCHEMA,
-                        "tier": self.tier,
-                        "pages": pages,
-                        "cached_at": time.time(),
-                    },
-                    ensure_ascii=False,
-                ),
-                encoding="utf-8",
-            )
-            tmp.replace(path)
-        except OSError as e:
-            logger.warning(f"⚠️ Ecriture cache OCR echouee ({e})")
-
-
-@lru_cache()
-def get_llama_parse_service() -> LlamaParseService:
-    """Singleton LlamaParseService."""
-    return LlamaParseService()
-
-

app/services/pdf_extraction_service.py

@@ -0,0 +1,529 @@
+"""
+Extraction de PDF en markdown par Gemini multimodal.
+
+Remplace LlamaParse. La raison n'est pas le prix mais la PAGINATION.
+
+LlamaParse rend le document en un bloc et ne signale pas ses coupures de page ;
+`_split_markdown_pages` les reconstruisait a l'heuristique, en coupant sur les
+lignes de trois tirets, et sa propre docstring admettait la limite : « la
+pagination ne devient exacte qu'avec un moteur qui renvoie un numero de page
+explicite ». Resultat mesure sur la base : les 710 articles portaient tous
+`page_number = 1`, et sur les 27 lois, 271 pages de PDF ne donnaient que 27
+pages extraites — couverture moyenne 0,60.
+
+Gemini est ce moteur. Mesure sur le Code Minier, dont la sortie LlamaParse est
+en base pour comparaison :
+
+    densite            2352 caracteres/page  contre 2405 — equivalent
+    marqueurs de page  1 par page            contre 1 pour 69 pages
+    lot pages 21 a 32  numerote 21..32       exactement
+
+CONTRAT DE SORTIE, impose par `text_chunker.py` et non negociable :
+`<<PAGE:n>>\\n<contenu>` par page, pages jointes par `\\n\\n`, numerotation a
+partir de 1 sur la page PHYSIQUE du PDF. Sans espace apres le deux-points :
+`text_chunker` compile `r'<<PAGE:(\\d+)>>'`, et un `<<PAGE: 12>>` serait
+invisible pour lui tout en etant retire ailleurs.
+
+Author: JuriX Team
+"""
+
+import asyncio
+import hashlib
+import io
+import json
+import logging
+import time
+from functools import lru_cache
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from google import genai
+from google.genai import types
+
+from app.core.config import settings
+from app.services.gemini_service import (
+    _finish_reason,
+    _is_overloaded,
+    _is_quota_exhausted,
+    _visible_text,
+    retry_after_seconds,
+)
+from app.utils.markdown_cleanup import strip_stamp_blocks
+
+logger = logging.getLogger(__name__)
+
+
+class PdfExtractionError(Exception):
+    """
+    Extraction impossible.
+
+    Son `str()` atterrit tel quel dans `laws.processing_error`, colonne visible
+    par l'administrateur : le message doit se lire, pas se decoder.
+    """
+
+
+class PdfExtractionQuotaError(PdfExtractionError):
+    """
+    Quota de generation epuise chez le fournisseur.
+
+    Type distinct pour que les scripts d'ingestion s'arretent proprement et
+    reprennent le lendemain, au lieu de compter le document comme un echec
+    definitif. Sous-classe de PdfExtractionError pour que le pipeline, lui,
+    n'ait rien a savoir de cette nuance.
+    """
+
+
+class PdfExtractionRefusedError(PdfExtractionError):
+    """
+    Le modele refuse de transcrire une page, la jugeant recitee.
+
+    Gemini rend `finish_reason=RECITATION` sur certaines pages d'actes officiels
+    — vraisemblablement celles dont la forme lui est familiere. Mesure sur un
+    echantillon de 7 documents : UNE page de garde refusee, soit environ 14 %.
+
+    Ce n'est PAS transitoire : la meme page refuse a l'identique, quelle que
+    soit la formulation de la consigne (trois variantes essayees, toutes
+    refusees) et quelle que soit la taille du lot. En revanche c'est PAR PAGE :
+    dans le document mesure, la page 1 refuse et les pages 2 et 3 s'extraient
+    normalement.
+    """
+
+
+# Consigne d'extraction. Ecrite en francais parce que le corpus l'est, et
+# formulee en regles numerotees : les consignes en prose donnaient des
+# preambules du genre « Voici la transcription du document ».
+_CONSIGNE = """Transcris ce document juridique en Markdown, fidelement et integralement.
+
+{pagination}
+
+Regles :
+1. Commence CHAQUE page par une ligne `<<PAGE:n>>` seule, sans espace apres le deux-points.
+2. Conserve les numeros d'article exactement tels qu'ils apparaissent, en debut de ligne.
+3. Restitue les tableaux en Markdown, sans en omettre de ligne.
+4. N'ajoute aucun commentaire, aucune introduction, aucune conclusion, aucune note.
+5. Ignore les filigranes, les cachets et les mentions de copie certifiee conforme.
+6. Si une page est vide, emets quand meme son marqueur, suivi de rien."""
+
+
+class GeminiPdfExtractor:
+    """
+    Extracteur PDF -> markdown pagine.
+
+    Interface volontairement identique a celle de l'ancien service : le
+    pipeline (`app/tasks/process_law.py:_extract_pdf_text`) ne fait que
+    `is_available()` puis `await extract_text(path)`, et convertit toute
+    exception en `status='refused'` + `processing_error`.
+    """
+
+    CACHE_SCHEMA = 3
+
+    # Nombre de tentatives sur une saturation passagere (503). Le palier gratuit
+    # en rend par salves de quelques secondes ; un 429, lui, n'est JAMAIS
+    # retente — cela brulerait des appels sur un quota deja epuise.
+    OVERLOAD_MAX_ATTEMPTS = 3
+    OVERLOAD_BASE_DELAY_S = 2.0
+
+    # Budget de sortie par appel. La limite du modele est de 65 536 jetons ;
+    # a ~2400 caracteres par page, un lot de 20 pages produit ~48 Ko, soit
+    # ~13 000 jetons. La marge couvre la reflexion interne du modele.
+    MAX_OUTPUT_TOKENS = 60_000
+
+    def __init__(
+        self,
+        api_key: Optional[str] = None,
+        model_name: Optional[str] = None,
+        pages_per_call: Optional[int] = None,
+        cache_dir: Optional[Path] = None,
+    ):
+        # `if api_key is None` et non `api_key or ...` : une chaine vide passee
+        # EXPLICITEMENT veut dire « pas de cle », et retombait sinon sur le
+        # reglage global — rendant intestable le cas du service non configure.
+        self.api_key = settings.GEMINI_API_KEY if api_key is None else api_key
+        self.model_name = model_name or settings.GEMINI_MODEL
+        self.pages_per_call = pages_per_call or settings.PDF_EXTRACTION_PAGES_PER_CALL
+
+        self.cache_dir: Optional[Path] = Path(
+            cache_dir or getattr(settings, "OCR_CACHE_DIR", "./data/ocr_cache")
+        )
+        try:
+            self.cache_dir.mkdir(parents=True, exist_ok=True)
+        except OSError as e:
+            logger.warning(f"⚠️ Cache d'extraction indisponible ({e}), desactive")
+            self.cache_dir = None
+
+        # Client construit paresseusement : `is_available()` doit pouvoir
+        # repondre sans cle, et le pipeline l'interroge avant tout travail.
+        self._client: Optional[genai.Client] = None
+
+        # Pages que le modele a refuse de transcrire lors de la derniere
+        # extraction. Lue par le script de re-extraction pour signaler les
+        # documents incomplets plutot que de les faire passer pour entiers.
+        self.pages_refusees: List[int] = []
+
+    # ==================== SURFACE PUBLIQUE ====================
+
+    def is_available(self) -> bool:
+        return bool(self.api_key)
+
+    async def extract_text(self, file_path: Path) -> str:
+        """
+        Rend le markdown pagine du PDF.
+
+        Raises:
+            PdfExtractionError: extraction impossible
+            PdfExtractionQuotaError: quota du fournisseur epuise
+        """
+        pages = await self.extract_pages(file_path)
+        if not pages:
+            raise PdfExtractionError(f"Aucune page extraite de {file_path.name}")
+
+        # Numerotation AVANT filtrage : une page blanche consomme son numero,
+        # ce qui garde les suivantes alignees sur la page physique du PDF.
+        parts = [f"<<PAGE:{i}>>\n{md}" for i, md in enumerate(pages, start=1) if md.strip()]
+        texte = "\n\n".join(parts)
+        logger.info(
+            f"✅ Gemini : {len(pages)} page(s), {len(texte)} caracteres "
+            f"depuis {file_path.name}"
+        )
+        return texte
+
+    async def extract_pages(self, file_path: Path) -> List[str]:
+        """Markdown page par page, index 0 = page 1 du PDF."""
+        if not self.is_available():
+            raise PdfExtractionError(
+                "GEMINI_API_KEY absente : extraction impossible. "
+                "Aucun repli degrade n'est utilise."
+            )
+        if not file_path.exists():
+            raise PdfExtractionError(f"Fichier introuvable: {file_path}")
+
+        # Remis a zero a chaque document : cette liste decrit CE document.
+        self.pages_refusees: List[int] = []
+
+        cle = await asyncio.to_thread(self._sha256, file_path)
+        en_cache = self._read_cache(cle)
+        if en_cache is not None:
+            logger.info(f"♻️ Cache d'extraction : {file_path.name} ({len(en_cache)} page(s))")
+            return en_cache
+
+        lots = await asyncio.to_thread(self._decouper_en_lots, file_path)
+        pages: List[str] = []
+        for premiere, octets in lots:
+            try:
+                pages.extend(await self._extraire_lot(octets, premiere, file_path.name))
+            except PdfExtractionRefusedError:
+                # Le refus est PAR PAGE, pas par document : mesure sur un
+                # document reel, la page 1 refuse et les pages 2 et 3 passent.
+                # On rejoue donc le lot page par page pour ne perdre que ce qui
+                # est reellement refuse.
+                logger.warning(
+                    f"⚠️ Lot refuse a partir de la page {premiere}, reprise page par page"
+                )
+                pages.extend(
+                    await self._extraire_lot_page_par_page(
+                        file_path, premiere, len(octets), file_path.name
+                    )
+                )
+
+        pages = [strip_stamp_blocks(p) for p in pages]
+        self._write_cache(cle, pages)
+        return pages
+
+    async def _extraire_lot_page_par_page(
+        self, file_path: Path, premiere_page: int, _taille: int, nom: str
+    ) -> List[str]:
+        """
+        Rejoue un lot refuse, une page a la fois.
+
+        Une page qui refuse encore est rendue VIDE plutot que de faire echouer
+        tout le document : perdre une page de garde vaut mieux que perdre les
+        soixante-huit autres. Le refus est journalise en ERREUR — il ne doit pas
+        se noyer — et la liste des pages perdues remonte a l'appelant, qui
+        decide.
+        """
+        pages_du_lot = await asyncio.to_thread(
+            self._pages_du_lot, file_path, premiere_page
+        )
+        resultat: List[str] = []
+        perdues: List[int] = []
+
+        for decalage, octets in enumerate(pages_du_lot):
+            numero = premiere_page + decalage
+            try:
+                extrait = await self._extraire_lot(octets, numero, nom)
+                resultat.extend(extrait)
+            except PdfExtractionRefusedError:
+                logger.error(f"❌ Page {numero} de {nom} refusee : contenu perdu")
+                perdues.append(numero)
+                resultat.append("")
+
+        if perdues:
+            self.pages_refusees.extend(perdues)
+        return resultat
+
+    def _pages_du_lot(self, file_path: Path, premiere_page: int) -> List[bytes]:
+        """Chaque page du lot, isolee, en memoire."""
+        from pypdf import PdfReader, PdfWriter
+
+        lecteur = PdfReader(str(file_path))
+        debut = premiere_page - 1
+        fin = min(debut + self.pages_per_call, len(lecteur.pages))
+        sortie = []
+        for index in range(debut, fin):
+            ecrivain = PdfWriter()
+            ecrivain.add_page(lecteur.pages[index])
+            tampon = io.BytesIO()
+            ecrivain.write(tampon)
+            sortie.append(tampon.getvalue())
+        return sortie
+
+    async def health_check(self) -> Dict[str, Any]:
+        if not self.is_available():
+            return {"service": "GeminiPdfExtractor", "status": "unconfigured"}
+        return {
+            "service": "GeminiPdfExtractor",
+            "status": "healthy",
+            "model": self.model_name,
+            "pages_per_call": self.pages_per_call,
+            "cache_dir": str(self.cache_dir) if self.cache_dir else None,
+        }
+
+    def compter_appels(self, file_path: Path) -> int:
+        """
+        Nombre d'appels que coutera ce document, cache non compris.
+
+        Sert au `--dry-run` du script de ré-extraction : sur un palier limite a
+        20 appels par jour, savoir AVANT de commencer vaut mieux que decouvrir
+        a mi-parcours.
+        """
+        total = self._compter_pages(file_path)
+        return (total + self.pages_per_call - 1) // self.pages_per_call
+
+    # ==================== APPEL AU MODELE ====================
+
+    @property
+    def client(self) -> genai.Client:
+        if self._client is None:
+            if not self.api_key:
+                raise PdfExtractionError("GEMINI_API_KEY absente")
+            # Meme delai que les autres services Gemini : sans http_options, un
+            # appel peut bloquer indefiniment sur une lecture de socket. Un
+            # incident de quarante minutes l'a etabli.
+            self._client = genai.Client(
+                api_key=self.api_key,
+                http_options=types.HttpOptions(timeout=settings.GEMINI_TIMEOUT_S * 1000),
+            )
+        return self._client
+
+    async def _extraire_lot(
+        self, octets: bytes, premiere_page: int, nom: str
+    ) -> List[str]:
+        """Envoie un lot de pages et rend son markdown, page par page."""
+        if premiere_page == 1:
+            pagination = "Ce document commence a sa premiere page."
+        else:
+            # SANS cette phrase, le modele renumerote chaque lot a partir de 1
+            # et chaque lot ecrase la pagination du precedent. Verifie : lot des
+            # pages 21 a 32 rendu « 21..32 » avec la phrase, « 1..12 » sans.
+            pagination = (
+                f"ATTENTION : ces pages sont un EXTRAIT. La premiere page fournie "
+                f"est la page {premiere_page} du document complet. Numerote donc "
+                f"les marqueurs a partir de {premiere_page}, en numerotation ABSOLUE."
+            )
+
+        contenu = [
+            types.Part.from_bytes(data=octets, mime_type="application/pdf"),
+            _CONSIGNE.format(pagination=pagination),
+        ]
+        config = types.GenerateContentConfig(
+            max_output_tokens=self.MAX_OUTPUT_TOKENS,
+            # Une transcription n'a pas a etre creative. La temperature nulle
+            # rend aussi le resultat reproductible d'un appel a l'autre, ce qui
+            # est indispensable pour comparer deux extractions.
+            temperature=0.0,
+        )
+
+        for essai in range(1, self.OVERLOAD_MAX_ATTEMPTS + 1):
+            try:
+                # client.aio : surface asynchrone native du SDK. La surface
+                # synchrone bloquerait la boucle pendant tout l'aller-retour,
+                # soit des dizaines de secondes par lot.
+                reponse = await self.client.aio.models.generate_content(
+                    model=self.model_name, contents=contenu, config=config
+                )
+                break
+            except Exception as err:
+                if _is_quota_exhausted(err):
+                    delai = retry_after_seconds(err)
+                    raise PdfExtractionQuotaError(
+                        f"Quota d'extraction epuise. Reprise possible dans {delai} secondes."
+                    ) from err
+                if not _is_overloaded(err) or essai == self.OVERLOAD_MAX_ATTEMPTS:
+                    raise PdfExtractionError(
+                        f"Extraction de {nom} impossible : {err}"
+                    ) from err
+                attente = self.OVERLOAD_BASE_DELAY_S * (2 ** (essai - 1))
+                logger.warning(
+                    f"⚠️ Modele sature (essai {essai}/{self.OVERLOAD_MAX_ATTEMPTS}), "
+                    f"nouvelle tentative dans {attente:.1f}s"
+                )
+                await asyncio.sleep(attente)
+
+        # `_visible_text` et non `reponse.text` : sur un modele a raisonnement,
+        # `.text` concatene la reflexion interne avec la reponse.
+        texte = _visible_text(reponse)
+        if not texte.strip():
+            raison = _finish_reason(reponse)
+            if raison == "RECITATION":
+                raise PdfExtractionRefusedError(
+                    f"Le modele refuse de transcrire {nom}, pages {premiere_page} et "
+                    f"suivantes (recitation)."
+                )
+            raise PdfExtractionError(
+                f"Reponse vide du modele sur {nom}, pages a partir de "
+                f"{premiere_page} (finish_reason={raison})"
+            )
+        return self._decouper_par_marqueur(texte, premiere_page)
+
+    @staticmethod
+    def _decouper_par_marqueur(texte: str, premiere_page: int) -> List[str]:
+        """
+        Separe le markdown sur les marqueurs `<<PAGE:n>>`.
+
+        Le marqueur lui-meme est RETIRE : `extract_text` le repose ensuite a
+        partir de la position dans la liste. C'est ce qui garantit que la
+        numerotation finale est continue meme si le modele en saute un.
+        """
+        import re
+
+        morceaux = re.split(r"<<PAGE:\s*(\d+)\s*>>", texte)
+        if len(morceaux) < 3:
+            # Aucun marqueur : le lot entier compte pour une page. Mieux vaut
+            # une page unique qu'une exception — le contenu, lui, est bon.
+            return [texte.strip()]
+
+        pages: Dict[int, str] = {}
+        for i in range(1, len(morceaux) - 1, 2):
+            numero = int(morceaux[i])
+            pages[numero] = morceaux[i + 1].strip()
+
+        # Reconstitution dense : de la premiere page annoncee au maximum vu,
+        # les numeros manquants deviennent des pages vides plutot que des trous.
+        if not pages:
+            return [texte.strip()]
+        fin = max(pages)
+        return [pages.get(n, "") for n in range(premiere_page, fin + 1)]
+
+    # ==================== DECOUPAGE DU PDF ====================
+
+    @staticmethod
+    def _compter_pages(file_path: Path) -> int:
+        from pypdf import PdfReader
+
+        try:
+            return len(PdfReader(str(file_path)).pages)
+        except Exception as e:
+            raise PdfExtractionError(f"PDF illisible ({file_path.name}) : {e}") from e
+
+    def _decouper_en_lots(self, file_path: Path) -> List[tuple]:
+        """
+        Rend [(numero_de_premiere_page, octets_du_lot), ...].
+
+        Decoupage EN MEMOIRE : l'ancien service ecrivait des fichiers
+        temporaires a cote de l'original et les effacait dans un `finally`. Un
+        `BytesIO` ne laisse rien a nettoyer et ne peut pas polluer le repertoire
+        d'upload si le processus meurt.
+        """
+        from pypdf import PdfReader, PdfWriter
+
+        lecteur = PdfReader(str(file_path))
+        total = len(lecteur.pages)
+        if total <= self.pages_per_call:
+            return [(1, file_path.read_bytes())]
+
+        lots = []
+        for debut in range(0, total, self.pages_per_call):
+            ecrivain = PdfWriter()
+            for page in lecteur.pages[debut:debut + self.pages_per_call]:
+                ecrivain.add_page(page)
+            tampon = io.BytesIO()
+            ecrivain.write(tampon)
+            lots.append((debut + 1, tampon.getvalue()))
+        logger.info(
+            f"✂️ {file_path.name} : {total} pages -> {len(lots)} lot(s) "
+            f"de {self.pages_per_call}"
+        )
+        return lots
+
+    # ==================== CACHE SHA256 ====================
+
+    @staticmethod
+    def _sha256(file_path: Path) -> str:
+        h = hashlib.sha256()
+        with file_path.open("rb") as fh:
+            for block in iter(lambda: fh.read(1024 * 1024), b""):
+                h.update(block)
+        return h.hexdigest()
+
+    def _cache_path(self, key: str) -> Optional[Path]:
+        return self.cache_dir / f"{key}.json" if self.cache_dir else None
+
+    def _read_cache(self, key: str) -> Optional[List[str]]:
+        path = self._cache_path(key)
+        if not path or not path.exists():
+            return None
+        try:
+            payload = json.loads(path.read_text(encoding="utf-8"))
+            # `engine` remplace l'ancien `tier` : une entree produite par un
+            # autre moteur — ou par un autre modele — n'est pas la meme
+            # extraction, et surtout pas la meme pagination.
+            if payload.get("engine") != self.model_name:
+                return None
+            if payload.get("schema") != self.CACHE_SCHEMA:
+                return None
+
+            pages = payload.get("pages")
+            # Une liste vide comptait autrefois comme un succes : une extraction
+            # ratee se gravait et se rejouait indefiniment, et le seul moyen
+            # d'en sortir etait d'effacer le fichier a la main.
+            if not pages:
+                logger.warning("⚠️ Entree de cache vide, ignoree")
+                return None
+            return pages
+        except Exception as e:
+            logger.warning(f"⚠️ Cache d'extraction illisible ({e}), ignore")
+            return None
+
+    def _write_cache(self, key: str, pages: List[str]) -> None:
+        path = self._cache_path(key)
+        if not path:
+            return
+        # Ne jamais graver un echec.
+        if not any(p.strip() for p in pages):
+            logger.warning("⚠️ Extraction vide, non mise en cache")
+            return
+        try:
+            # Ecriture atomique : un plantage en cours d'ecriture laissait un
+            # JSON tronque qui occupait la place de l'entree valide.
+            tmp = path.with_suffix(".json.tmp")
+            tmp.write_text(
+                json.dumps(
+                    {
+                        "schema": self.CACHE_SCHEMA,
+                        "engine": self.model_name,
+                        "pages": pages,
+                        "cached_at": time.time(),
+                    },
+                    ensure_ascii=False,
+                ),
+                encoding="utf-8",
+            )
+            tmp.replace(path)
+        except OSError as e:
+            logger.warning(f"⚠️ Ecriture du cache d'extraction echouee ({e})")
+
+
+@lru_cache()
+def get_pdf_extractor() -> GeminiPdfExtractor:
+    """Instance partagee. Le client Gemini et le cache disque sont reutilises."""
+    return GeminiPdfExtractor()

app/tasks/process_law.py

@@ -294,44 +294,53 @@ def _ingest_file_content(law_id: int, law, file_id: str):
 
 def _extract_pdf_text(file_path) -> tuple:
     """
-    Extrait le texte d'un PDF via LlamaParse. Aucun repli degrade.
+    Extrait le texte d'un PDF via Gemini multimodal. Aucun repli degrade.
 
-    Le repli precedent — LlamaParse -> Tesseract -> pypdf — etait dangereux :
-    pypdf renvoie la couche texte deja presente dans les PDF prc.cm, mesuree a
-    ~20% de rappel (filigrane injecte au milieu des phrases, cachets lus comme
-    du charabia, 21% des documents sans aucun texte exploitable). Le document
-    etait alors publie avec ce contenu et marque "completed", sans aucun signal.
+    Le repli LlamaParse -> Tesseract -> pypdf a ete supprime avant celui-ci, et
+    la raison tient toujours : pypdf renvoie la couche texte deja presente dans
+    les PDF prc.cm, mesuree a ~20% de rappel (filigrane injecte au milieu des
+    phrases, cachets lus comme du charabia, 21% des documents sans aucun texte
+    exploitable). Le document etait alors publie avec ce contenu et marque
+    "completed", sans aucun signal.
 
-    Desormais : LlamaParse fait ses 4 tentatives avec backoff exponentiel ; s'il
-    echoue, on leve. L'appelant marque le document "refused" avec l'erreur, et
-    il reste retraitable. Rien de degrade n'entre dans l'index.
+    Le moteur a change pour la PAGINATION. LlamaParse ne signalait pas ses
+    coupures de page : les 710 articles en base portaient tous
+    `page_number = 1`, et 271 pages de PDF ne donnaient que 27 pages extraites.
+    Gemini rend un numero de page explicite, verifie exact sur un lot median.
 
     Returns:
-        (texte, []) — la liste d'erreurs est conservee pour compatibilite d'appel
+        (texte, erreurs) — `erreurs` nomme les pages que le modele a refuse de
+        transcrire, le cas echeant. Le document reste publie sans elles : perdre
+        une page de garde vaut mieux que perdre les soixante-huit autres.
 
     Raises:
-        LlamaParseError: extraction impossible
+        PdfExtractionError: extraction impossible
     """
 
-    from app.services.llama_parse_service import (
-        LlamaParseError,
-        get_llama_parse_service,
+    from app.services.pdf_extraction_service import (
+        PdfExtractionError,
+        get_pdf_extractor,
     )
 
-    # get_llama_parse_service() et non LlamaParseService() : le singleton
-    # lru_cache existait mais etait contourne par une construction directe,
-    # ce qui recreait le client et perdait le cache a chaque document.
-    service = get_llama_parse_service()
+    service = get_pdf_extractor()
     if not service.is_available():
-        raise LlamaParseError(
-            "LLAMA_CLOUD_API_KEY absente : extraction impossible. "
+        raise PdfExtractionError(
+            "GEMINI_API_KEY absente : extraction impossible. "
             "Aucun repli degrade n'est utilise."
         )
 
-    logger.info(f"📄 Extraction LlamaParse : {file_path.name}")
+    logger.info(f"📄 Extraction Gemini : {file_path.name}")
     text = _run_blocking(service.extract_text(file_path))
-    logger.info(f"✅ LlamaParse : {len(text)} caracteres")
-    return text, []
+    logger.info(f"✅ Gemini : {len(text)} caracteres")
+
+    erreurs = []
+    if service.pages_refusees:
+        erreurs.append(
+            "Pages non transcrites (refus du modele) : "
+            + ", ".join(str(n) for n in service.pages_refusees)
+        )
+        logger.error(f"❌ {file_path.name} : {erreurs[0]}")
+    return text, erreurs
 
 
 def _extract_docx_text(file_path) -> tuple:

app/utils/markdown_cleanup.py

@@ -0,0 +1,108 @@
+"""
+Nettoyage du markdown produit par l'extraction de PDF.
+
+Ces regles ne dependent d'AUCUN fournisseur : elles decrivent ce que le corpus
+prc.cm contient et qu'il ne faut pas indexer. Elles vivaient dans
+`llama_parse_service.py` et auraient disparu avec lui ; elles ont ete calibrees
+sur ce corpus et servent tout autant a l'extraction par Gemini.
+
+Trois regles, chacune payee par une observation :
+
+- Le cachet « COPIE CERTIFIEE CONFORME » est appose sur CHAQUE page. Restitue en
+  texte, il pollue le plein texte et les vecteurs de tous les documents.
+- Le filigrane diagonal `www.prc.cm` ressort eclate caractere par caractere.
+- `ARTICLE 1<sup>ER</sup>` doit redevenir `ARTICLE 1ER`, faute de quoi
+  `text_chunker.ARTICLE_PATTERNS` ne reconnait pas le marqueur — mais `<table>`
+  n'est JAMAIS retire : les annexes budgetaires portent leur sens dans leur
+  structure tabulaire.
+
+Author: JuriX Team
+"""
+
+import re
+
+# Vocabulaire du cachet "COPIE CERTIFIEE CONFORME" appose sur chaque page du
+# corpus prc.cm. L'extracteur le restitue en texte, il faut donc le retirer avant chunking sinon il pollue le FTS
+# et les embeddings de chaque document.
+_STAMP_VOCAB = re.compile(
+    r"(copie\s+certifi\w*\s+conforme"
+    r"|certified\s+true\s+copy"
+    r"|service\s+du\s+fichier\s+l[ée]gislatif"
+    r"|legislative\s+and\s+statutory\s+affairs"
+    r"|presidency\s+of\s+the\s+republic"
+    r"|secr[ée]tariat[\s-]+g[ée]n[ée]ral"
+    r"|pr[ée]sidence\s+de\s+la\s+r[ée]publique)",
+    re.IGNORECASE,
+)
+
+# Fragments du filigrane diagonal "www.prc.cm" eclate par l'OCR
+_WATERMARK = re.compile(
+    r"^\s*(?:w\s*){1,3}$|^\s*\.?\s*(?:p\s*r\s*c|c\s*m|prc\.cm|www\.prc\.cm)\s*\.?\s*$",
+    re.IGNORECASE,
+)
+
+# Balisage HTML inline emis par l'extracteur (les <table> sont conserves : la
+# structure tabulaire porte du sens, cf. les annexes budgetaires).
+# <sup>/<sub> sont critiques : "ARTICLE 1<sup>ER</sup>" doit redevenir
+# "ARTICLE 1ER" pour que ARTICLE_PATTERNS de text_chunker le reconnaisse.
+_INLINE_TAGS = re.compile(
+    r"</?(?:u|mark|b|i|em|strong|span|sup|sub|small)\b[^>]*>", re.IGNORECASE
+)
+
+
+def strip_stamp_blocks(text: str) -> str:
+    """
+    Retire les blocs de cachet officiel du markdown.
+
+    Un bloc = au moins 2 lignes consecutives (blancs autorises) appartenant au
+    vocabulaire du cachet. Le seuil de 2 evite de supprimer un en-tete legitime :
+    beaucoup de documents portent "PRESIDENCE DE LA REPUBLIQUE" comme autorite
+    emettrice, ce qui est du contenu, pas du tampon.
+
+    Args:
+        text: Markdown brut d'extraction
+
+    Returns:
+        Markdown nettoye
+    """
+    lines = text.splitlines()
+    drop = [False] * len(lines)
+    i = 0
+
+    while i < len(lines):
+        if not _STAMP_VOCAB.search(lines[i]):
+            i += 1
+            continue
+
+        # Cas 1 : cachet entier restitue sur une seule ligne, par ex.
+        # "[signature: PRESIDENCE ... COPIE CERTIFIEE CONFORME CERTIFIED TRUE COPY]"
+        # ou "stamp: ..." / "logo: ...". Deux phrases distinctes du vocabulaire sur
+        # la meme ligne ne peuvent pas etre du texte juridique legitime.
+        if len(set(m.group(0).lower() for m in _STAMP_VOCAB.finditer(lines[i]))) >= 2:
+            drop[i] = True
+            i += 1
+            continue
+
+        # Etendre le bloc tant qu'on reste dans le vocabulaire du cachet
+        j = i
+        hits = 0
+        while j < len(lines):
+            stripped = lines[j].strip(" >*|-\t")
+            if _STAMP_VOCAB.search(lines[j]):
+                hits += 1
+                j += 1
+            elif not stripped:
+                j += 1
+            else:
+                break
+
+        if hits >= 2:
+            for k in range(i, j):
+                drop[k] = True
+        i = max(j, i + 1)
+
+    kept = [ln for ln, d in zip(lines, drop) if not d]
+    kept = [ln for ln in kept if not _WATERMARK.match(ln)]
+    cleaned = "\n".join(kept)
+    cleaned = _INLINE_TAGS.sub("", cleaned)
+    return re.sub(r"\n{4,}", "\n\n\n", cleaned).strip()

scripts/eval/extraction/baseline.py

@@ -18,10 +18,15 @@ CE QU'IL MESURE, et pourquoi chaque mesure est la :
      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 latent de normalize_for_chunking(). La fonction existe dans
-     app/utils/chunk_refiner.py, elle est testee, et elle n'est appelee nulle
-     part en production. On rejoue la detection de motif sur les caches OCR
-     deja payes, avec et sans elle, pour chiffrer ce que son cablage rapporte.
+  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

scripts/maintenance/reextract_corpus.py

@@ -0,0 +1,206 @@
+#!/usr/bin/env python3
+"""
+Re-extrait les documents deja en base avec le moteur Gemini.
+
+POURQUOI. L'extraction precedente ne signalait pas les coupures de page : les
+710 articles portent tous `page_number = 1`, et sur les 27 lois, 271 pages de
+PDF ne donnent que 27 pages extraites — couverture moyenne 0,60. La consequence
+visible est qu'un lien `?article=35` ouvre le PDF a la premiere page.
+
+REPRENABLE PAR CONSTRUCTION. Le cache est indexe par sha256 du fichier : un
+document deja extrait n'est jamais repaye, donc relancer le script reprend ou le
+quota l'avait arrete. Sur 429, il s'arrete proprement, dit combien de documents
+restent, et sort en code 2 — ce qui distingue « quota atteint » de « ca a
+plante ».
+
+Le palier gratuit plafonne a 20 appels de generation par jour et par modele.
+Un document d'une page coute un appel ; les 72 PDF locaux (481 pages) en
+coutent 83.
+
+Usage:
+    python scripts/maintenance/reextract_corpus.py --dry-run
+    python scripts/maintenance/reextract_corpus.py --apply --max-calls 18
+    python scripts/maintenance/reextract_corpus.py --apply --law-id 16
+
+Enchainement complet :
+    reextract_corpus.py --apply        # laws.content recoit les vrais <<PAGE:n>>
+    rechunk_laws.py --apply            # articles.page_number devient exact
+    regenerate_embeddings.py --all --force
+
+Codes de sortie:
+    0  termine
+    1  au moins un document en echec
+    2  quota epuise, relancer plus tard
+"""
+
+import argparse
+import asyncio
+import logging
+import sys
+from pathlib import Path
+from typing import List, Optional
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+
+from app.core.database import SyncSessionLocal
+from app.models.law import Law
+from app.services.pdf_extraction_service import (
+    PdfExtractionError,
+    PdfExtractionQuotaError,
+    get_pdf_extractor,
+)
+from app.utils.file_utils import resolve_upload_path
+
+logging.basicConfig(level=logging.INFO, format="%(message)s")
+logger = logging.getLogger("reextract")
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        description="Re-extrait les documents en base avec Gemini.",
+    )
+    mode = parser.add_mutually_exclusive_group()
+    mode.add_argument("--dry-run", action="store_true",
+                      help="n'appelle rien, chiffre le cout en appels (defaut)")
+    mode.add_argument("--apply", action="store_true",
+                      help="re-extrait et ecrit laws.content")
+    parser.add_argument("--law-id", type=int, action="append", dest="law_ids",
+                        help="ne traiter que cette loi (repetable)")
+    parser.add_argument("--max-calls", type=int, default=None,
+                        help="plafond dur d'appels pour cette session — le garde-fou du quota")
+    parser.add_argument("--only-multipage", action="store_true",
+                        help="ignorer les documents d'une seule page")
+    return parser
+
+
+def _documents(session, law_ids: Optional[List[int]]):
+    """Rend [(loi, chemin_du_pdf)] pour les lois dont le fichier existe."""
+    query = session.query(Law).order_by(Law.id)
+    if law_ids:
+        query = query.filter(Law.id.in_(law_ids))
+
+    sortie = []
+    for law in query.all():
+        if not law.file_id:
+            logger.warning("  #%-5s sans file_id, ignoree", law.id)
+            continue
+        try:
+            sortie.append((law, resolve_upload_path(law.file_id)))
+        except (ValueError, FileNotFoundError) as exc:
+            logger.warning("  #%-5s fichier introuvable (%s)", law.id, exc)
+    return sortie
+
+
+async def _traiter(documents, extracteur, appliquer: bool, max_calls: Optional[int]) -> dict:
+    bilan = {"traites": 0, "echecs": 0, "appels": 0, "pages_refusees": 0, "quota": False}
+
+    for law, chemin in documents:
+        cout = extracteur.compter_appels(chemin)
+        if max_calls is not None and bilan["appels"] + cout > max_calls:
+            logger.info(
+                "  #%-5s %s appel(s) — plafond de session atteint, arret ici",
+                law.id, cout,
+            )
+            break
+
+        try:
+            texte = await extracteur.extract_text(chemin)
+        except PdfExtractionQuotaError as exc:
+            logger.error("\n⛔ %s", exc)
+            bilan["quota"] = True
+            break
+        except PdfExtractionError as exc:
+            logger.error("  #%-5s ECHEC : %s", law.id, exc)
+            bilan["echecs"] += 1
+            continue
+
+        bilan["appels"] += cout
+        bilan["traites"] += 1
+        refusees = list(extracteur.pages_refusees)
+        bilan["pages_refusees"] += len(refusees)
+
+        marqueurs = texte.count("<<PAGE:")
+        detail = f" — pages refusees : {refusees}" if refusees else ""
+        logger.info(
+            "  #%-5s %5d car, %3d page(s), %s appel(s)%s  %s",
+            law.id, len(texte), marqueurs, cout, detail, (law.title or "")[:38],
+        )
+
+        if appliquer:
+            # Une transaction PAR document : un echec en cours de route ne doit
+            # pas laisser la moitie du corpus dans un etat intermediaire.
+            with SyncSessionLocal() as session:
+                ligne = session.get(Law, law.id)
+                if ligne:
+                    ligne.content = texte
+                    ligne.processing_error = (
+                        f"Pages non transcrites (refus du modele) : "
+                        f"{', '.join(str(n) for n in refusees)}"
+                        if refusees else None
+                    )
+                    session.commit()
+
+    return bilan
+
+
+def main(argv: Optional[List[str]] = None) -> int:
+    args = build_parser().parse_args(argv)
+    extracteur = get_pdf_extractor()
+
+    if not extracteur.is_available():
+        logger.error("❌ GEMINI_API_KEY absente : rien a faire.")
+        return 1
+
+    with SyncSessionLocal() as session:
+        documents = _documents(session, args.law_ids)
+        # Detacher les objets : la session se ferme, et le traitement ouvre la
+        # sienne par document.
+        documents = [
+            (session.merge(law), chemin) for law, chemin in documents
+        ]
+        session.expunge_all()
+
+    if args.only_multipage:
+        documents = [
+            (law, chemin) for law, chemin in documents
+            if extracteur._compter_pages(chemin) > 1
+        ]
+
+    total_appels = sum(extracteur.compter_appels(chemin) for _, chemin in documents)
+    logger.info("Documents            : %d", len(documents))
+    logger.info("Appels necessaires   : %d  (cache non compris)", total_appels)
+    logger.info("Soit, a 20 par jour  : %.1f jour(s)", total_appels / 20)
+    if args.max_calls:
+        logger.info("Plafond de session   : %d appel(s)", args.max_calls)
+    logger.info("%s", "-" * 72)
+
+    if not args.apply:
+        logger.info("(dry-run : aucun appel emis. Relancez avec --apply.)")
+        return 0
+
+    bilan = asyncio.run(_traiter(documents, extracteur, True, args.max_calls))
+
+    logger.info("%s", "=" * 72)
+    logger.info("Documents traites    : %d", bilan["traites"])
+    logger.info("Appels consommes     : %d", bilan["appels"])
+    if bilan["pages_refusees"]:
+        logger.info("Pages refusees       : %d  (voir laws.processing_error)",
+                    bilan["pages_refusees"])
+    if bilan["echecs"]:
+        logger.error("Echecs               : %d", bilan["echecs"])
+
+    restants = len(documents) - bilan["traites"] - bilan["echecs"]
+    if bilan["quota"]:
+        logger.info("\n⛔ Quota epuise. %d document(s) restent.", restants)
+        logger.info("   Relancez demain : le cache reprend ou il s'est arrete.")
+        return 2
+    if restants:
+        logger.info("\n%d document(s) restent (plafond de session).", restants)
+
+    logger.info("\nEtape suivante : python scripts/maintenance/rechunk_laws.py --apply")
+    logger.info("%s", "=" * 72)
+    return 1 if bilan["echecs"] else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/test_services/test_llama_parse_service.py

@@ -1,305 +0,0 @@
-"""
-Tests du service d'extraction LlamaParse.
-
-Aucun appel réseau : `respx` intercepte httpx. Sans cela ces tests
-consommeraient des crédits LlamaParse à chaque exécution.
-"""
-
-import json
-
-import httpx
-import pytest
-import respx
-
-from app.services.llama_parse_service import (
-    LlamaParseError,
-    LlamaParseService,
-    _split_markdown_pages,
-    _strip_stamp_blocks,
-)
-
-BASE = "https://api.cloud.llamaindex.ai/api/v2/parse"
-
-
-@pytest.fixture
-def service(tmp_path):
-    return LlamaParseService(api_key="cle-de-test", cache_dir=tmp_path / "cache")
-
-
-@pytest.fixture
-def pdf(tmp_path):
-    p = tmp_path / "doc.pdf"
-    p.write_bytes(b"%PDF-1.4\ncontenu\n%%EOF")
-    return p
-
-
-class TestNettoyageDesCachets:
-    """
-    Retrait du cachet officiel apposé sur chaque page du corpus prc.cm.
-
-    LlamaParse restitue ce cachet en texte ; sans nettoyage il pollue le FTS et
-    les embeddings de tous les documents.
-    """
-
-    def test_bloc_multiligne_retire(self):
-        md = (
-            "Article 1er. Contenu juridique réel.\n\n"
-            "PRESIDENCE DE LA REPUBLIQUE\n"
-            "SERVICE DU FICHIER LEGISLATIF ET REGLEMENTAIRE\n"
-            "COPIE CERTIFIEE CONFORME\n"
-            "CERTIFIED TRUE COPY\n\n"
-            "Article 2. Autre contenu."
-        )
-        out = _strip_stamp_blocks(md)
-        assert "COPIE CERTIFIEE CONFORME" not in out
-        assert "Article 1er. Contenu juridique réel." in out
-        assert "Article 2. Autre contenu." in out
-
-    def test_cachet_sur_une_seule_ligne_retire(self):
-        """LlamaParse condense parfois tout le cachet sur une ligne."""
-        md = (
-            "Article 1er. Contenu.\n"
-            "[signature: PRESIDENCE DE LA REPUBLIQUE SECRETARIAT GENERAL "
-            "COPIE CERTIFIEE CONFORME CERTIFIED TRUE COPY]\n"
-        )
-        out = _strip_stamp_blocks(md)
-        assert "CERTIFIED TRUE COPY" not in out
-        assert "Article 1er. Contenu." in out
-
-    def test_en_tete_legitime_conserve(self):
-        """
-        Beaucoup de décrets portent « PRESIDENCE DE LA REPUBLIQUE » comme
-        autorité émettrice : c'est du contenu, pas un tampon. Une seule phrase
-        du vocabulaire ne doit donc rien déclencher.
-        """
-        md = "PRESIDENCE DE LA REPUBLIQUE\n\nDECRET N° 2024/191 du 4 juin 2024"
-        out = _strip_stamp_blocks(md)
-        assert "PRESIDENCE DE LA REPUBLIQUE" in out
-        assert "DECRET N° 2024/191" in out
-
-    def test_filigrane_retire(self):
-        md = "Article 1er.\nw\nw\nw\n.prc\n.cm\nContenu."
-        out = _strip_stamp_blocks(md)
-        assert "Article 1er." in out and "Contenu." in out
-        assert "\nw\n" not in f"\n{out}\n"
-
-    def test_tableaux_conserves(self):
-        """Les <table> portent du sens (annexes budgétaires) — jamais retirés."""
-        md = "<table><tr><td>Recettes</td><td>66 900 000</td></tr></table>"
-        assert "<table>" in _strip_stamp_blocks(md)
-
-    def test_balises_inline_retirees(self):
-        """<sup> casse la détection d'article : ARTICLE 1<sup>ER</sup>."""
-        assert _strip_stamp_blocks("**ARTICLE 1<sup>ER</sup>**") == "**ARTICLE 1ER**"
-
-
-class TestDisponibilite:
-    def test_indisponible_sans_cle(self, tmp_path, monkeypatch):
-        # api_key="" retombe volontairement sur settings.LLAMA_CLOUD_API_KEY :
-        # il faut donc neutraliser aussi la configuration pour tester ce cas.
-        monkeypatch.setattr(
-            "app.services.llama_parse_service.settings.LLAMA_CLOUD_API_KEY", ""
-        )
-        assert not LlamaParseService(api_key="", cache_dir=tmp_path).is_available()
-
-    def test_disponible_avec_cle(self, service):
-        assert service.is_available()
-
-
-class TestExtraction:
-    @pytest.mark.asyncio
-    @respx.mock
-    async def test_extraction_et_marqueurs_de_page(self, service, pdf):
-        respx.post(f"{BASE}/upload").mock(
-            return_value=httpx.Response(200, json={"job": {"id": "job-1"}})
-        )
-        respx.get(f"{BASE}/job-1").mock(
-            return_value=httpx.Response(
-                200,
-                json={
-                    "job": {"status": "COMPLETED"},
-                    "markdown": [{"markdown": "Page une"}, {"markdown": "Page deux"}],
-                },
-            )
-        )
-        service.POLL_INTERVAL = 0
-        text = await service.extract_text(pdf)
-        # Les marqueurs alimentent Article.page_number via text_chunker
-        assert "<<PAGE:1>>" in text and "<<PAGE:2>>" in text
-        assert "Page une" in text and "Page deux" in text
-
-    @pytest.mark.asyncio
-    @respx.mock
-    async def test_cache_evite_un_second_appel(self, service, pdf):
-        upload = respx.post(f"{BASE}/upload").mock(
-            return_value=httpx.Response(200, json={"job": {"id": "job-1"}})
-        )
-        respx.get(f"{BASE}/job-1").mock(
-            return_value=httpx.Response(
-                200,
-                json={"job": {"status": "COMPLETED"}, "markdown": [{"markdown": "Texte"}]},
-            )
-        )
-        service.POLL_INTERVAL = 0
-        first = await service.extract_text(pdf)
-        second = await service.extract_text(pdf)
-        assert first == second
-        assert upload.call_count == 1, "le second appel aurait dû être servi par le cache"
-
-    @pytest.mark.asyncio
-    @respx.mock
-    async def test_job_en_echec_leve(self, service, pdf):
-        respx.post(f"{BASE}/upload").mock(
-            return_value=httpx.Response(200, json={"job": {"id": "job-1"}})
-        )
-        respx.get(f"{BASE}/job-1").mock(
-            return_value=httpx.Response(
-                200,
-                json={"job": {"status": "FAILED", "error_message": "document illisible"}},
-            )
-        )
-        service.POLL_INTERVAL = 0
-        with pytest.raises(LlamaParseError, match="document illisible"):
-            await service.extract_text(pdf)
-
-    @pytest.mark.asyncio
-    @respx.mock
-    async def test_cle_refusee_ne_declenche_pas_de_retry(self, service, pdf):
-        """Un 401 est définitif : réessayer 4 fois ne ferait que perdre du temps."""
-        route = respx.post(f"{BASE}/upload").mock(
-            return_value=httpx.Response(401, text="cle invalide")
-        )
-        service.POLL_INTERVAL = 0
-        with pytest.raises(LlamaParseError):
-            await service.extract_text(pdf)
-        assert route.call_count == 1
-
-    @pytest.mark.asyncio
-    async def test_fichier_absent(self, service, tmp_path):
-        with pytest.raises(LlamaParseError, match="introuvable"):
-            await service.extract_text(tmp_path / "absent.pdf")
-
-
-class TestNormalisationDesReponses:
-    """L'API v2 renvoie le markdown sous plusieurs formes selon les options."""
-
-    def test_liste_de_pages(self):
-        pages = LlamaParseService._pages_from(
-            {"markdown": [{"markdown": "a"}, {"markdown": "b"}]}
-        )
-        assert pages == ["a", "b"]
-
-    def test_chaine_unique(self):
-        assert LlamaParseService._pages_from({"markdown": "seule"}) == ["seule"]
-
-    def test_champ_pages(self):
-        assert LlamaParseService._pages_from({"pages": [{"markdown": "x"}]}) == ["x"]
-
-    def test_markdown_full(self):
-        assert LlamaParseService._pages_from({"markdown_full": "tout"}) == ["tout"]
-
-    def test_reponse_vide(self):
-        assert LlamaParseService._pages_from({}) == []
-
-
-class TestDecoupeEnPages:
-    """
-    Restitution de la pagination quand la réponse arrive en un seul bloc.
-
-    Sans cette découpe, `_pages_from` renvoyait une page unique pour tout le
-    document : les 849 articles en base portaient tous `page_number = 1`, alors
-    que les vraies coupures étaient présentes dans le markdown.
-
-    Le seuil « exactement trois tirets » n'est pas arbitraire : mesuré sur 64
-    documents appariés à leur PDF (473 pages de vérité terrain), `-{3,}`
-    donnait 44/64 comptes exacts et 10 sur-découpes, `-{3}` en donne 51/64 et 3.
-    """
-
-    def test_saut_de_page_decoupe(self):
-        pages = _split_markdown_pages("Page un.\n\n---\nPage deux.\n\n---\nPage trois.")
-        assert pages == ["Page un.", "Page deux.", "Page trois."]
-
-    def test_separateur_de_tableau_ne_decoupe_pas(self):
-        # `|---|---|` contient des barres : la ligne entière est exigée.
-        markdown = "Avant\n\n| a | b |\n|---|---|\n| 1 | 2 |\n\nApres"
-        assert len(_split_markdown_pages(markdown)) == 1
-
-    def test_tirets_longs_ne_decoupent_pas(self):
-        # Soulignements et bordures ASCII : du contenu, jamais un saut de page.
-        assert len(_split_markdown_pages("A\n-----\nB")) == 1
-
-    def test_document_sans_saut_reste_une_page(self):
-        assert _split_markdown_pages("Un seul bloc.") == ["Un seul bloc."]
-
-    def test_page_vide_interieure_conservee(self):
-        # `extract_text` numérote AVANT de filtrer : supprimer une page blanche
-        # ici décalerait tous les numéros suivants.
-        assert _split_markdown_pages("A\n---\n\n---\nC") == ["A", "", "C"]
-
-    def test_sauts_aux_bords_ne_creent_pas_de_page_vide(self):
-        assert _split_markdown_pages("\n---\nA\n---\nB\n---\n") == ["A", "B"]
-
-    def test_frontmatter_yaml_ne_decoupe_pas(self):
-        # Le délimiteur fermant du frontmatter est lui aussi `---`.
-        assert len(_split_markdown_pages("---\ntitre: x\n---\nCorps unique.")) == 1
-
-    def test_frontmatter_reste_sur_la_premiere_page(self):
-        pages = _split_markdown_pages("---\nt: x\n---\nPage un\n---\nPage deux")
-        assert pages == ["---\nt: x\n---\nPage un", "Page deux"]
-
-    def test_texte_vide(self):
-        assert _split_markdown_pages("") == [""]
-
-    def test_applique_a_markdown_full(self):
-        assert LlamaParseService._pages_from({"markdown_full": "A\n---\nB"}) == ["A", "B"]
-
-
-class TestIntegriteDuCache:
-    """
-    Le cache ne doit jamais graver un échec, ni rejouer un format périmé.
-
-    Une liste vide comptait comme un HIT : une extraction ratée se gravait sur
-    disque et se rejouait indéfiniment sans jamais rappeler l'API. Le seul moyen
-    d'en sortir était d'effacer le fichier à la main.
-    """
-
-    def test_extraction_vide_non_gravee(self, service, tmp_path):
-        service._write_cache("vide", [])
-        assert service._cache_path("vide").exists() is False
-
-    def test_pages_blanches_non_gravees(self, service):
-        service._write_cache("blanc", ["", "   "])
-        assert service._cache_path("blanc").exists() is False
-
-    def test_entree_vide_sur_disque_est_un_miss(self, service):
-        service._cache_path("poison").write_text(
-            json.dumps({"schema": service.CACHE_SCHEMA, "tier": service.tier, "pages": []})
-        )
-        assert service._read_cache("poison") is None
-
-    def test_aller_retour(self, service):
-        service._write_cache("ok", ["page un", "page deux"])
-        assert service._read_cache("ok") == ["page un", "page deux"]
-
-    def test_cache_v1_redecoupe_sur_place_sans_rappeler_l_api(self, service):
-        # Le contenu d'une entrée v1 est valide et déjà payé : il est redécoupé
-        # localement, jamais re-extrait.
-        service._cache_path("v1").write_text(
-            json.dumps({"tier": service.tier, "pages": ["Page un\n---\nPage deux"]})
-        )
-        assert service._read_cache("v1") == ["Page un", "Page deux"]
-
-        # Et réécrit au format courant, donc redécoupé une seule fois.
-        payload = json.loads(service._cache_path("v1").read_text())
-        assert payload["schema"] == service.CACHE_SCHEMA
-        assert len(payload["pages"]) == 2
-
-    def test_cache_v1_vide_reste_un_miss(self, service):
-        service._cache_path("v1vide").write_text(
-            json.dumps({"tier": service.tier, "pages": []})
-        )
-        assert service._read_cache("v1vide") is None
-
-    def test_ecriture_atomique_sans_residu(self, service, tmp_path):
-        service._write_cache("atomique", ["contenu"])
-        assert list(service.cache_dir.glob("*.tmp")) == []

tests/test_services/test_pdf_extraction.py

@@ -0,0 +1,332 @@
+"""
+Extraction des PDF par Gemini.
+
+Ces tests figent le CONTRAT que le decoupeur impose, et les trois modes d'echec
+observes en production sur le palier gratuit.
+
+Le reseau est simule par monkeypatch sur `client.aio.models.generate_content`,
+et non par respx : respx intercepte httpx, alors que le SDK google-genai a sa
+propre pile — un mock respx ne verrait jamais l'appel.
+
+La fixture PDF est un VRAI PDF construit par pypdf. Le talon de 22 octets qui
+suffisait a l'ancien service ne convient plus : celui-ci ouvre le fichier pour
+compter et decouper ses pages.
+
+Usage:
+    pytest tests/test_services/test_pdf_extraction.py -v
+"""
+
+import json
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from app.services.pdf_extraction_service import (
+    GeminiPdfExtractor,
+    PdfExtractionError,
+    PdfExtractionQuotaError,
+)
+
+SATURATION = (
+    "503 UNAVAILABLE. {'error': {'code': 503, 'message': 'This model is currently "
+    "experiencing high demand.', 'status': 'UNAVAILABLE'}}"
+)
+QUOTA = (
+    "429 RESOURCE_EXHAUSTED. {'error': {'code': 429, 'message': 'You exceeded your "
+    "current quota', 'details': [{'retryDelay': '42s'}]}}"
+)
+
+
+def _pdf(chemin: Path, pages: int = 1) -> Path:
+    from pypdf import PdfWriter
+
+    writer = PdfWriter()
+    for _ in range(pages):
+        writer.add_blank_page(width=200, height=200)
+    with chemin.open("wb") as fh:
+        writer.write(fh)
+    return chemin
+
+
+def _reponse(texte: str, finish: str = "STOP"):
+    """Reponse a la forme de celle du SDK, reflexion comprise si demandee."""
+    part = SimpleNamespace(text=texte, thought=False)
+    contenu = SimpleNamespace(parts=[part] if texte else [])
+    candidat = SimpleNamespace(content=contenu, finish_reason=SimpleNamespace(name=finish))
+    return SimpleNamespace(candidates=[candidat], text=texte)
+
+
+@pytest.fixture
+def extracteur(tmp_path):
+    return GeminiPdfExtractor(
+        api_key="cle-de-test",
+        model_name="modele-test",
+        pages_per_call=2,
+        cache_dir=tmp_path / "cache",
+    )
+
+
+def _brancher(extracteur, reponses):
+    """
+    Remplace l'appel au modele par une liste de reponses ou d'exceptions.
+
+    Rend un compteur d'appels : c'est lui qui prouve qu'un quota epuise n'est
+    PAS retente.
+    """
+    etat = {"appels": 0}
+    file = list(reponses)
+
+    async def _faux(**kwargs):
+        etat["appels"] += 1
+        item = file.pop(0) if file else file_defaut
+        if isinstance(item, Exception):
+            raise item
+        return item
+
+    file_defaut = _reponse("<<PAGE:1>>\nContenu.")
+    extracteur._client = SimpleNamespace(
+        aio=SimpleNamespace(models=SimpleNamespace(generate_content=_faux))
+    )
+    return etat
+
+
+class TestContratDeSortie:
+    """Ce que `text_chunker` exige, au caractere pres."""
+
+    @pytest.mark.asyncio
+    async def test_marqueur_sans_espace_apres_les_deux_points(self, extracteur, tmp_path):
+        """
+        `text_chunker.py` compile `r'<<PAGE:(\\d+)>>'`. Un `<<PAGE: 1>>` serait
+        invisible pour lui tout en etant retire ailleurs : la page existerait
+        dans le texte et pas dans la numerotation.
+        """
+        _brancher(extracteur, [_reponse("<<PAGE:1>>\nLe contenu.")])
+
+        texte = await extracteur.extract_text(_pdf(tmp_path / "d.pdf"))
+
+        assert texte.startswith("<<PAGE:1>>\n")
+        assert "<<PAGE: " not in texte
+
+    @pytest.mark.asyncio
+    async def test_pages_jointes_par_deux_sauts_de_ligne(self, extracteur, tmp_path):
+        _brancher(extracteur, [_reponse("<<PAGE:1>>\nUn\n<<PAGE:2>>\nDeux")])
+
+        texte = await extracteur.extract_text(_pdf(tmp_path / "d.pdf", pages=2))
+
+        assert texte == "<<PAGE:1>>\nUn\n\n<<PAGE:2>>\nDeux"
+
+    @pytest.mark.asyncio
+    async def test_une_page_blanche_ne_decale_pas_les_suivantes(self, extracteur, tmp_path):
+        """
+        La numerotation se fait AVANT le filtrage des pages vides : une page
+        blanche consomme son numero, ce qui garde les suivantes alignees sur la
+        page physique du PDF.
+        """
+        _brancher(extracteur, [_reponse("<<PAGE:1>>\nUn\n<<PAGE:2>>\n\n<<PAGE:3>>\nTrois")])
+
+        texte = await extracteur.extract_text(_pdf(tmp_path / "d.pdf", pages=3))
+
+        assert "<<PAGE:1>>\nUn" in texte
+        assert "<<PAGE:3>>\nTrois" in texte
+        assert "<<PAGE:2>>" not in texte, "la page vide est omise, pas rendue vide"
+
+    @pytest.mark.asyncio
+    async def test_texte_sans_aucun_marqueur_compte_pour_une_page(self, extracteur, tmp_path):
+        """Mieux vaut une page unique qu'une exception : le contenu est bon."""
+        _brancher(extracteur, [_reponse("Du texte sans le moindre marqueur.")])
+
+        texte = await extracteur.extract_text(_pdf(tmp_path / "d.pdf"))
+
+        assert texte == "<<PAGE:1>>\nDu texte sans le moindre marqueur."
+
+
+class TestNumerotationAbsolue:
+    """
+    Le point qui casse en silence : sans consigne explicite, le modele
+    renumerote chaque lot a partir de 1 et chaque lot ecrase le precedent.
+    """
+
+    @pytest.mark.asyncio
+    async def test_la_consigne_annonce_la_premiere_page_du_lot(self, extracteur, tmp_path):
+        consignes = []
+
+        async def _capturer(**kwargs):
+            consignes.append(kwargs["contents"][1])
+            depart = len(consignes)  # lot 1 -> pages 1-2, lot 2 -> pages 3-4
+            base = 1 if depart == 1 else 3
+            return _reponse(f"<<PAGE:{base}>>\nA\n<<PAGE:{base + 1}>>\nB")
+
+        extracteur._client = SimpleNamespace(
+            aio=SimpleNamespace(models=SimpleNamespace(generate_content=_capturer))
+        )
+
+        await extracteur.extract_text(_pdf(tmp_path / "d.pdf", pages=4))
+
+        assert len(consignes) == 2, "4 pages en lots de 2 font deux appels"
+        assert "premiere page" in consignes[0]
+        assert "page 3 du document complet" in consignes[1]
+        assert "ABSOLUE" in consignes[1]
+
+    @pytest.mark.asyncio
+    async def test_les_lots_se_recollent_en_numerotation_continue(self, extracteur, tmp_path):
+        import re
+
+        appels = {"n": 0}
+
+        async def _faux(**kwargs):
+            appels["n"] += 1
+            base = 1 if appels["n"] == 1 else 3
+            return _reponse(f"<<PAGE:{base}>>\nA\n<<PAGE:{base + 1}>>\nB")
+
+        extracteur._client = SimpleNamespace(
+            aio=SimpleNamespace(models=SimpleNamespace(generate_content=_faux))
+        )
+
+        texte = await extracteur.extract_text(_pdf(tmp_path / "d.pdf", pages=4))
+
+        assert [int(n) for n in re.findall(r"<<PAGE:(\d+)>>", texte)] == [1, 2, 3, 4]
+
+
+class TestModesDEchec:
+    @pytest.mark.asyncio
+    async def test_un_quota_epuise_n_est_jamais_retente(self, extracteur, tmp_path):
+        """Retenter un quota epuise brule des appels qu'on n'a deja plus."""
+        etat = _brancher(extracteur, [Exception(QUOTA)])
+
+        with pytest.raises(PdfExtractionQuotaError, match="42 secondes"):
+            await extracteur.extract_text(_pdf(tmp_path / "d.pdf"))
+
+        assert etat["appels"] == 1
+
+    @pytest.mark.asyncio
+    async def test_une_saturation_est_retentee_puis_abandonnee(self, extracteur, tmp_path):
+        etat = _brancher(extracteur, [Exception(SATURATION)] * 3)
+        extracteur.OVERLOAD_BASE_DELAY_S = 0
+
+        with pytest.raises(PdfExtractionError):
+            await extracteur.extract_text(_pdf(tmp_path / "d.pdf"))
+
+        assert etat["appels"] == 3
+
+    @pytest.mark.asyncio
+    async def test_une_saturation_passagere_est_absorbee(self, extracteur, tmp_path):
+        _brancher(extracteur, [Exception(SATURATION), _reponse("<<PAGE:1>>\nOK")])
+        extracteur.OVERLOAD_BASE_DELAY_S = 0
+
+        assert "OK" in await extracteur.extract_text(_pdf(tmp_path / "d.pdf"))
+
+    @pytest.mark.asyncio
+    async def test_un_refus_de_recitation_rejoue_page_par_page(self, extracteur, tmp_path):
+        """
+        Mesure sur un document reel : la page 1 refuse, les pages 2 et 3
+        s'extraient. Refuser tout le document pour une page de garde ferait
+        perdre les soixante-huit autres.
+        """
+        reponses = [
+            _reponse("", finish="RECITATION"),   # le lot entier
+            _reponse("", finish="RECITATION"),   # page 1, refusee a nouveau
+            _reponse("<<PAGE:2>>\nDeux"),        # page 2, elle passe
+        ]
+        _brancher(extracteur, reponses)
+
+        texte = await extracteur.extract_text(_pdf(tmp_path / "d.pdf", pages=2))
+
+        assert "Deux" in texte
+        assert extracteur.pages_refusees == [1]
+
+    @pytest.mark.asyncio
+    async def test_une_reponse_vide_sans_raison_connue_leve(self, extracteur, tmp_path):
+        _brancher(extracteur, [_reponse("", finish="SAFETY")])
+
+        with pytest.raises(PdfExtractionError, match="SAFETY"):
+            await extracteur.extract_text(_pdf(tmp_path / "d.pdf"))
+
+    @pytest.mark.asyncio
+    async def test_sans_cle_le_service_refuse_de_travailler(self, tmp_path):
+        muet = GeminiPdfExtractor(api_key="", cache_dir=tmp_path / "c")
+
+        assert muet.is_available() is False
+        with pytest.raises(PdfExtractionError, match="GEMINI_API_KEY"):
+            await muet.extract_text(_pdf(tmp_path / "d.pdf"))
+
+    @pytest.mark.asyncio
+    async def test_fichier_absent(self, extracteur, tmp_path):
+        with pytest.raises(PdfExtractionError, match="introuvable"):
+            await extracteur.extract_text(tmp_path / "nexiste-pas.pdf")
+
+
+class TestCache:
+    @pytest.mark.asyncio
+    async def test_un_second_appel_ne_touche_pas_le_modele(self, extracteur, tmp_path):
+        etat = _brancher(extracteur, [_reponse("<<PAGE:1>>\nContenu.")])
+        pdf = _pdf(tmp_path / "d.pdf")
+
+        premier = await extracteur.extract_text(pdf)
+        second = await extracteur.extract_text(pdf)
+
+        assert premier == second
+        assert etat["appels"] == 1
+
+    @pytest.mark.asyncio
+    async def test_une_extraction_vide_n_est_jamais_gravee(self, extracteur, tmp_path):
+        """
+        Une entree vide comptait autrefois comme un succes : l'echec se figeait
+        et le seul moyen d'en sortir etait d'effacer le fichier a la main.
+        """
+        extracteur._write_cache("cle-test", ["", "   "])
+
+        assert list((tmp_path / "cache").glob("*.json")) == []
+
+    def test_une_entree_vide_sur_disque_est_un_defaut_de_cache(self, extracteur, tmp_path):
+        chemin = tmp_path / "cache" / "cle-test.json"
+        chemin.write_text(json.dumps(
+            {"schema": extracteur.CACHE_SCHEMA, "engine": "modele-test", "pages": []}
+        ))
+
+        assert extracteur._read_cache("cle-test") is None
+
+    def test_une_entree_d_un_autre_moteur_est_ignoree(self, extracteur, tmp_path):
+        """
+        Les 61 entrees heritees de LlamaParse portent `tier`, pas `engine` :
+        elles ne doivent pas etre resservies comme si elles venaient d'ici — ce
+        sont precisement les extractions non paginees qu'on remplace.
+        """
+        chemin = tmp_path / "cache" / "cle-test.json"
+        chemin.write_text(json.dumps(
+            {"schema": 2, "tier": "cost_effective", "pages": ["ancien contenu"]}
+        ))
+
+        assert extracteur._read_cache("cle-test") is None
+
+    def test_ecriture_atomique_sans_residu(self, extracteur, tmp_path):
+        extracteur._write_cache("cle-test", ["du contenu"])
+
+        assert list((tmp_path / "cache").glob("*.tmp")) == []
+        assert extracteur._read_cache("cle-test") == ["du contenu"]
+
+
+class TestDecoupageEtComptage:
+    def test_un_document_court_ne_fait_qu_un_lot(self, extracteur, tmp_path):
+        pdf = _pdf(tmp_path / "d.pdf", pages=2)
+
+        assert extracteur.compter_appels(pdf) == 1
+        assert len(extracteur._decouper_en_lots(pdf)) == 1
+
+    def test_un_document_long_est_decoupe(self, extracteur, tmp_path):
+        pdf = _pdf(tmp_path / "d.pdf", pages=5)
+
+        lots = extracteur._decouper_en_lots(pdf)
+
+        assert extracteur.compter_appels(pdf) == 3
+        assert [premiere for premiere, _ in lots] == [1, 3, 5]
+
+    def test_le_decoupage_ne_laisse_aucun_fichier_temporaire(self, extracteur, tmp_path):
+        """
+        Decoupage EN MEMOIRE : l'ancien service ecrivait des fichiers a cote de
+        l'original et les effacait dans un `finally`, ce qu'un processus tue ne
+        fait jamais.
+        """
+        extracteur._decouper_en_lots(_pdf(tmp_path / "d.pdf", pages=5))
+
+        assert sorted(p.name for p in tmp_path.glob("*.pdf")) == ["d.pdf"]

tests/test_utils/test_markdown_cleanup.py

@@ -0,0 +1,74 @@
+"""
+Nettoyage du markdown d'extraction.
+
+Ces six tests ont suivi leur code : ils vivaient dans
+`test_llama_parse_service.py` et auraient disparu avec lui. Les regles qu'ils
+figent ne dependent d'aucun fournisseur — elles decrivent ce que le corpus
+prc.cm contient et qu'il ne faut pas indexer — et servent autant a l'extraction
+par Gemini qu'a celle qu'elles protegeaient avant.
+
+Usage:
+    pytest tests/test_utils/test_markdown_cleanup.py -v
+"""
+
+from app.utils.markdown_cleanup import strip_stamp_blocks
+
+
+class TestNettoyageDesCachets:
+    """
+    Retrait du cachet officiel apposé sur chaque page du corpus prc.cm.
+
+    L'extracteur restitue ce cachet en texte ; sans nettoyage il pollue le FTS et
+    les embeddings de tous les documents.
+    """
+
+    def test_bloc_multiligne_retire(self):
+        md = (
+            "Article 1er. Contenu juridique réel.\n\n"
+            "PRESIDENCE DE LA REPUBLIQUE\n"
+            "SERVICE DU FICHIER LEGISLATIF ET REGLEMENTAIRE\n"
+            "COPIE CERTIFIEE CONFORME\n"
+            "CERTIFIED TRUE COPY\n\n"
+            "Article 2. Autre contenu."
+        )
+        out = strip_stamp_blocks(md)
+        assert "COPIE CERTIFIEE CONFORME" not in out
+        assert "Article 1er. Contenu juridique réel." in out
+        assert "Article 2. Autre contenu." in out
+
+    def test_cachet_sur_une_seule_ligne_retire(self):
+        """LlamaParse condense parfois tout le cachet sur une ligne."""
+        md = (
+            "Article 1er. Contenu.\n"
+            "[signature: PRESIDENCE DE LA REPUBLIQUE SECRETARIAT GENERAL "
+            "COPIE CERTIFIEE CONFORME CERTIFIED TRUE COPY]\n"
+        )
+        out = strip_stamp_blocks(md)
+        assert "CERTIFIED TRUE COPY" not in out
+        assert "Article 1er. Contenu." in out
+
+    def test_en_tete_legitime_conserve(self):
+        """
+        Beaucoup de décrets portent « PRESIDENCE DE LA REPUBLIQUE » comme
+        autorité émettrice : c'est du contenu, pas un tampon. Une seule phrase
+        du vocabulaire ne doit donc rien déclencher.
+        """
+        md = "PRESIDENCE DE LA REPUBLIQUE\n\nDECRET N° 2024/191 du 4 juin 2024"
+        out = strip_stamp_blocks(md)
+        assert "PRESIDENCE DE LA REPUBLIQUE" in out
+        assert "DECRET N° 2024/191" in out
+
+    def test_filigrane_retire(self):
+        md = "Article 1er.\nw\nw\nw\n.prc\n.cm\nContenu."
+        out = strip_stamp_blocks(md)
+        assert "Article 1er." in out and "Contenu." in out
+        assert "\nw\n" not in f"\n{out}\n"
+
+    def test_tableaux_conserves(self):
+        """Les <table> portent du sens (annexes budgétaires) — jamais retirés."""
+        md = "<table><tr><td>Recettes</td><td>66 900 000</td></tr></table>"
+        assert "<table>" in strip_stamp_blocks(md)
+
+    def test_balises_inline_retirees(self):
+        """<sup> casse la détection d'article : ARTICLE 1<sup>ER</sup>."""
+        assert strip_stamp_blocks("**ARTICLE 1<sup>ER</sup>**") == "**ARTICLE 1ER**"