""" Raffinage des chunks pour le RAG juridique camerounais. Couche de post-traitement appliquee APRES text_chunker.extract_articles(). Le chunker existant decoupe correctement par article ; ce module regle ce qui se passe autour, et qui determine la qualite des reponses du chatbot. Constat qui motive ce module — composition mesuree du corpus prc.cm (1883 docs) : nominatif (listes de noms, avancements) 52,6 % ratification / emprunt (1-3 articles) 19,4 % NORMATIF (contenu juridique reel) 28,0 % Un chunking naif produit donc l'essentiel de ses chunks a partir des 52,6 % qui ne repondront jamais a une question de droit — et comme toutes les listes de noms s'embeddent au meme endroit de l'espace vectoriel, elles saturent le top-k de chaque recherche semantique. Regles appliquees : R2 contextualisation en-tete document prepose a chaque chunk (embed_text) R3 visas hors index LEGAL_BASIS sorti de l'index + graphe de citations R4 listes nominatives liste gardee au contenu, resumee au vecteur, entrees aussi en table separee R5 tableaux entiers jamais coupes au milieu d'une ligne R6 tailles decoupe aux alineas, boilerplate et signature hors index R7 deduplication chunks identiques fusionnes Principe directeur : RIEN N'EST SUPPRIME. Les chunks ecartes du vectoriel gardent `embed=False` et restent cherchables en FTS exact. Sur une base juridique, ne jamais perdre de contenu — seulement le hierarchiser. Author: JuriX Team """ import hashlib import html import logging import re from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Tuple logger = logging.getLogger(__name__) # ==================== SEUILS ==================== # ~800 tokens. Au-dela on coupe aux alineas (1), (2), (3)... TARGET_MAX_CHARS = 3000 # Debut du chapeau d'un article coupe, repete devant chacun de ses morceaux. LONGUEUR_CHAPEAU_REPETE = 200 # En-deca, un chunk isole n'est pas repondable seul MIN_CHARS = 120 # Seuil d'un VRAI article : « Article 4.- Le decret n° 2001/041 est abroge. » # est court mais normatif, et l'ecarter du vectoriel le rendait introuvable par # le sens. En deca de ce seuil, l'article n'a plus de texte (OCR vide). MIN_CHARS_ARTICLE = 30 # Proportion de lignes "nominatives" a partir de laquelle un tableau est un roster ROSTER_ROW_RATIO = 0.6 ROSTER_MIN_ROWS = 5 # ==================== MOTIFS ==================== # Matricule administratif camerounais : "765 609-Y", "699 536-N", et le format # actuel sans tiret "0570984M", "583397-B". _MATRICULE = re.compile(r"\b(?:\d{3}\s?\d{3}\s?-\s?[A-Z]|\d{6,7}\s?-?\s?[A-Z])\b") # Montant : « 12 512 163 », « 1.200.000 ». Une ligne de tableau qui en porte # deux est une ligne budgetaire, pas une personne. _MONTANT = re.compile(r"\b\d{1,3}(?:[ .\u202f]\d{3})+\b") # Civilite : signal d'une personne dans une cellule _CIVILITE = re.compile(r"\b(?:M\.|MM\.|Mme|Mlle|Mr|Mrs|Dr|Monsieur|Madame)\s", re.IGNORECASE) # En-tete de tableau nominatif : « N° | Nom | Matricule » _ENTETE_TABLEAU = re.compile( r"\b(?:n[°ºo]|nom|noms|pr[ée]noms?|matricules?|grades?|indices?|corps|fonctions?|rang)\b", re.IGNORECASE, ) # Nom en capitales : au moins deux mots, apostrophes et tirets admis _CAPS_NAME = re.compile(r"^[A-ZÀ-Þ][A-ZÀ-Þ'’\-\.]*(?:\s+[A-ZÀ-Þ][A-ZÀ-Þ'’\-\.]*){1,6}$") # Tableaux HTML produits par LlamaParse _TABLE_BLOCK = re.compile(r"
| n'apporte rien a un vecteur et dilue le signal semantique. On garde le HTML dans `content` (affichage, citation) et on vectorise cette version texte. """ lines = [] for tr in _TR_BLOCK.findall(table_html): cells = [_cell_text(td) for td in _TD_CELL.findall(tr)] if any(cells): lines.append(" | ".join(cells)) return "\n".join(lines) def _md_table_rows(table_md: str, skip_header: bool = True) -> List[List[str]]: """ Lignes d'un tableau markdown, sous forme de listes de cellules. La ligne de separation `|---|` est ecartee ; avec skip_header, les lignes qui la precedent aussi — l'en-tete « N° | Nom | Matricule » ne doit pas compter pour une personne. """ lignes = [ln for ln in table_md.splitlines() if ln.strip()] separateur = next((i for i, ln in enumerate(lignes) if _MD_SEPARATOR_ROW.match(ln)), None) if separateur is not None: entete = lignes[:separateur] # Docling met en « en-tete » la premiere ligne de DONNEES quand le # tableau n'en a pas : « | 1. NDOUNGA EVINA JEAN | 607 027-V | ». La # sauter perdait une personne par page de liste. Elle n'est sautee que # si elle ressemble a un en-tete : ni matricule, ni rang, et des mots # comme Nom ou Matricule. vrai_entete = all( not _MATRICULE.search(ln) and not re.match(r"\s*\|?\s*\d{1,4}\s*[.)-]", ln) and _ENTETE_TABLEAU.search(ln) for ln in entete ) if skip_header and vrai_entete: lignes = lignes[separateur + 1:] else: lignes = entete + lignes[separateur + 1:] rows = [] for ligne in lignes: cells = [c.strip() for c in ligne.strip().strip("|").split("|")] if any(cells): rows.append(cells) return rows def _md_table_to_text(table_md: str) -> str: """Tableau markdown aplati pour l'embedding, sans la ligne de separation.""" return "\n".join( " | ".join(cells) for cells in _md_table_rows(table_md, skip_header=False) ) def _contains_table(content: str) -> bool: return bool(_TABLE_BLOCK.search(content) or _MD_TABLE_BLOCK.search(content)) def _looks_nominative(rows: List[List[str]], contexte: str = "") -> bool: """ Le tableau est-il une liste de personnes ? Critere : au moins ROSTER_ROW_RATIO des lignes portent soit un matricule, soit un nom entierement en capitales de 2 mots ou plus — et le tableau porte un SIGNAL DE PERSONNE : des matricules, des civilites, ou un texte de nomination autour. Sans ce signal, les tableaux budgetaires passaient : « DEVELOPPEMENT DU PRESCOLAIRE » est un nom en capitales, et une loi de finances perdait ses montants dans une « liste de 56 personnes ». Une ligne qui porte deux montants n'est jamais une personne. """ if len(rows) < ROSTER_MIN_ROWS: return False hits = matricules = civilites = 0 for cells in rows: joined = " ".join(cells) if len(_MONTANT.findall(joined)) >= 2: continue if _MATRICULE.search(joined): hits += 1 matricules += 1 continue if _CIVILITE.search(joined): civilites += 1 if any(_CAPS_NAME.match(c) or _nom_colle(c) for c in cells if len(c) > 4): hits += 1 signal = matricules >= 0.3 * len(rows) or civilites or _NOMINATION_CUE.search(contexte) return bool(signal) and hits / len(rows) >= ROSTER_ROW_RATIO def _nom_colle(cellule: str) -> bool: """« MBINCHOBLAISEAMBE » : un nom en capitales que l'OCR a colle en un mot.""" c = re.sub(r"^\s*\d{1,4}\s*[.)-]\s*", "", cellule.strip()) return bool(re.fullmatch(r"[A-ZÀ-Þ'’\-]{8,60}", c)) def _parse_roster_rows( rows: List[List[str]], article_number: str, start: int = 0 ) -> List[RosterEntry]: """Convertit les lignes d'un tableau nominatif en RosterEntry.""" entries: List[RosterEntry] = [] for cells in rows: joined = " ".join(cells) mat = _MATRICULE.search(joined) identifier = mat.group(0).strip() if mat else None name = None rank_in_cell = None for c in cells: # « 2. ELOUNDOU ESSOMBA ALBERT » : rang et nom dans la meme cellule m = re.match(r"\s*(\d{1,4})\s*[.)-]\s+(.+)$", c) candidat = m.group(2).strip() if m else c if len(candidat) > 4 and (_CAPS_NAME.match(candidat) or _nom_colle(candidat)): name = candidat rank_in_cell = m.group(1) if m else None break if not name: # repli : la cellule la plus longue qui n'est ni le rang ni le matricule candidates = [ c for c in cells if len(c) > 4 and not _MATRICULE.fullmatch(c.strip()) and not re.fullmatch(r"\d{1,4}\s*[.)-]?", c.strip()) ] name = max(candidates, key=len) if candidates else None if not name: continue rank = rank_in_cell or next( (c.strip(" .)-") for c in cells if re.fullmatch(r"\d{1,4}\s*[.)-]?", c.strip())), None, ) entries.append( RosterEntry( article_number=article_number, position=start + len(entries), name=name.strip(), identifier=identifier, rank=rank, ) ) return entries def _parse_plain_roster(text: str, article_number: str) -> Tuple[List[RosterEntry], str]: """ Variante texte brut : une personne par ligne. Deux passes. D'abord les lignes qui portent un matricule : signature sure, 5 suffisent. Sinon, les lignes « rang + nom en capitales », plus ambigues : elles ne valent liste que dans un texte de nomination, et si elles font au moins ROSTER_ROW_RATIO des lignes du chunk — une enumeration de ministeres dans un article normatif ne doit pas disparaitre du contenu. Returns: (entrees, texte sans les lignes nominatives) """ lignes = text.splitlines() def _collecte(motif, avec_matricule: bool) -> Tuple[List[RosterEntry], List[str]]: entries: List[RosterEntry] = [] kept: List[str] = [] for line in lignes: m = motif.match(line) if m: entries.append( RosterEntry( article_number=article_number, position=len(entries), name=m.group(2).strip(), identifier=m.group(3).strip() if avec_matricule else None, rank=m.group(1), ) ) else: kept.append(line) return entries, kept entries, kept = _collecte(_PLAIN_ROSTER_LINE, avec_matricule=True) if len(entries) >= ROSTER_MIN_ROWS: return entries, "\n".join(kept) if not _NOMINATION_CUE.search(text): return [], text non_vides = sum(1 for ln in lignes if ln.strip()) entries, kept = _collecte(_NUMBERED_NAME_LINE, avec_matricule=False) if len(entries) >= ROSTER_MIN_ROWS and len(entries) >= ROSTER_ROW_RATIO * non_vides: return entries, "\n".join(kept) # Puces sans rang : « - TEMGOUA NOUMBO HONORINE » puces: List[RosterEntry] = [] reste: List[str] = [] for line in lignes: m = _BULLET_NAME_LINE.match(line) if m: puces.append(RosterEntry(article_number=article_number, position=len(puces), name=m.group(1).strip())) else: reste.append(line) if len(puces) >= ROSTER_MIN_ROWS and len(puces) >= ROSTER_ROW_RATIO * non_vides: return puces, "\n".join(reste) return [], text # ==================== REGLES ==================== def _apply_roster_rule( chunk: Dict[str, Any] ) -> Tuple[Dict[str, Any], List[RosterEntry]]: """ R4 — Effondre une liste nominative en un seul chunk normatif. Un arrete nommant 904 inspecteurs ne doit pas produire 904 vecteurs : ils sont semantiquement indiscernables et satureraient toutes les recherches. On garde l'enveloppe juridique (qui, promu a quoi, a quel indice, a compter de quand) et on renvoie les personnes en table a part, cherchables en FTS exact. """ content = chunk["content"] number = str(chunk.get("number", "")) entries: List[RosterEntry] = [] resume = content # Variante tableau : HTML (LlamaParse) ou markdown a barres (Docling) tables = [(t, _table_rows(t)) for t in _TABLE_BLOCK.findall(content)] tables += [(m.group(0), _md_table_rows(m.group(0))) for m in _MD_TABLE_BLOCK.finditer(content)] # Le texte de nomination se cherche HORS des tableaux : dans une loi de # reglement, « PROMOTION DE LA FEMME », intitule de programme, se faisait # passer pour lui, et un tableau budgetaire devenait une liste de 35 # « personnes ». hors_tableaux = _MD_TABLE_BLOCK.sub(" ", _TABLE_BLOCK.sub(" ", content)) nominatives = [_looks_nominative(rows, hors_tableaux) for _, rows in tables] # Une liste nominative s'etale sur un tableau par page, et la detection en # manque quelques-uns (matricule « J-057 035 », rang absent). Restes dans # le texte vectorise, ils en faisaient 155 000 caracteres. Des que les # tableaux reconnus portent la majorite des lignes, l'article est une # liste : TOUS ses tableaux en font partie. lignes_nominatives = sum(len(rows) for (_, rows), nom in zip(tables, nominatives) if nom) if lignes_nominatives and lignes_nominatives >= 0.5 * sum(len(rows) for _, rows in tables): for table, rows in tables: lignes = _parse_roster_rows(rows, number, start=len(entries)) entries.extend(lignes) # Un decompte PAR tableau : le total global repete deux fois trompait. resume = resume.replace( table.rstrip("\n"), f"[{len(lignes)} personnes — liste nominative]" ) # Variante texte brut if not entries: entries, candidate = _parse_plain_roster(content, number) if entries: resume = candidate + f"\n\n[{len(entries)} personnes — liste nominative]" if not entries: return chunk, [] # Le CONTENU garde la liste : c'est lui qu'on affiche, qu'on cite et que # le plein texte indexe. La liste n'etait enregistree nulle part ailleurs — # les noms disparaissaient de l'application, remplaces par une « annexe » # qui n'existait pas. Seul le texte VECTORISE prend la forme resumee : 904 # inspecteurs ne doivent pas faire 904 vecteurs, ni saturer la recherche. chunk = dict(chunk) resume = resume.strip() if len(resume) > TARGET_MAX_CHARS: # Garde-fou : le vecteur d'une liste ne lit que son chapeau. Au-dela, # l'embedding tronquerait de toute facon. coupe = resume.rfind("\n", 0, TARGET_MAX_CHARS) resume = resume[:coupe if coupe > TARGET_MAX_CHARS // 2 else TARGET_MAX_CHARS].rstrip() chunk["embed_body"] = resume chunk["kind"] = "roster" chunk["roster_count"] = len(entries) return chunk, entries def _split_roster( chunk: Dict[str, Any], target_max_chars: int = TARGET_MAX_CHARS ) -> List[Dict[str, Any]]: """ R4/R5 — Decoupe une longue liste nominative entre ses lignes. La liste reste dans le contenu, pour l'affichage, la citation et le plein texte. Mais entiere, une liste de 904 personnes fait un chunk de 40 000 caracteres dont le contexte du chat ne lit que les 6 000 premiers : un nom de la fin, retrouve par le plein texte, n'arrivait jamais au modele. Chaque morceau repete le chapeau de l'article, tronque, pour dire de quelle liste il s'agit. Seul le premier est vectorise, avec le resume de la liste (`embed_body`) : 904 noms ne font toujours qu'un vecteur. Une ligne n'est jamais coupee ; un tableau HTML ne se coupe pas du tout. """ content = chunk["content"] if len(content) <= target_max_chars or _TABLE_BLOCK.search(content): return [chunk] lignes = content.splitlines() debut = next((i for i, ln in enumerate(lignes) if _LIGNE_DE_LISTE.match(ln)), len(lignes)) chapeau = "\n".join(lignes[:debut]).strip() prefixe = chapeau[:200].strip() morceaux: List[List[str]] = [] taille = len(chapeau) for ligne in lignes[debut:]: if morceaux and morceaux[-1] and taille + len(ligne) + 1 > target_max_chars: morceaux.append([]) taille = len(prefixe) if not morceaux: morceaux.append([]) morceaux[-1].append(ligne) taille += len(ligne) + 1 if len(morceaux) < 2: return [chunk] out: List[Dict[str, Any]] = [] for i, morceau in enumerate(morceaux): piece = dict(chunk) tete = chapeau if i == 0 else prefixe piece["content"] = (tete + "\n" + "\n".join(morceau)).strip() piece["number"] = f"{chunk['number']}.{i + 1}" piece["parent_id"] = str(chunk["number"]) piece["char_count"] = len(piece["content"]) piece["word_count"] = len(piece["content"].split()) if i > 0: piece["embed"] = False piece.pop("embed_body", None) out.append(piece) return out def _ligne_d_en_tete(ligne: str) -> bool: """ La premiere ligne d'un tableau est-elle un en-tete, a repeter ? Docling met en « en-tete » la premiere ligne de DONNEES quand le tableau n'en a pas. Un montant ou un rang la trahit : la repeter dans chaque morceau dupliquerait une ligne budgetaire. """ return not _MONTANT.search(ligne) and not re.match(r"\s*\|?\s*\d{1,4}\s*[.)|-]", ligne) def _split_tableaux( chunk: Dict[str, Any], target_max_chars: int = TARGET_MAX_CHARS ) -> List[Dict[str, Any]]: """ R5 — Decoupe un article qui porte un grand tableau, entre ses lignes. Les tableaux n'etaient jamais coupes : ceux des lois de finances faisaient des chunks de 79 000 caracteres, dont l'embedding ne lisait que les 10 000 premiers — 110 chunks du corpus, dont la fin n'etait jamais cherchable par le sens. Ils se coupent desormais entre deux lignes, jamais au milieu d'une : « 31 | 180 | EDUCATION PRESCOLAIRE | 31 915 303 » reste entier. Chaque morceau repete l'en-tete du tableau, pour que ses colonnes aient un nom, et le debut de l'article, pour dire de quel tableau il s'agit. Le texte autour des tableaux se coupe aux paragraphes. Rien n'est perdu ni duplique, hors en-tetes repetes. """ content = chunk["content"] blocs: List[Tuple[str, str]] = [] debut = 0 for m in _MD_TABLE_BLOCK.finditer(content): if content[debut:m.start()].strip(): blocs.append(("texte", content[debut:m.start()])) blocs.append(("tableau", m.group(0))) debut = m.end() if content[debut:].strip(): blocs.append(("texte", content[debut:])) chapeau = blocs[0][1].strip() if blocs[0][0] == "texte" else "" if len(chapeau) > 600: chapeau = "" prefixe = chapeau[:200].strip() place = max(target_max_chars - len(prefixe) - 1, 500) unites: List[Tuple[str, bool]] = [] for i, (nature, texte) in enumerate(blocs): if nature == "texte": if i == 0 and chapeau: continue unites += [(p.strip(), False) for p in re.split(r"\n\s*\n", texte) if p.strip()] continue lignes = [ln for ln in texte.splitlines() if ln.strip()] sep = next((k for k, ln in enumerate(lignes) if _MD_SEPARATOR_ROW.match(ln)), None) entete = lignes[:sep + 1] if sep is not None and sep <= 2 and all( _ligne_d_en_tete(ln) for ln in lignes[:sep] ) else [] corps = [ln for ln in lignes[len(entete):] if not _MD_SEPARATOR_ROW.match(ln)] budget = max(place - sum(len(ln) + 1 for ln in entete), 300) groupe: List[str] = [] taille = 0 for ligne in corps: if groupe and taille + len(ligne) + 1 > budget: unites.append(("\n".join(entete + groupe), True)) groupe, taille = [], 0 groupe.append(ligne) taille += len(ligne) + 1 if groupe: unites.append(("\n".join(entete + groupe), True)) morceaux: List[Tuple[str, bool]] = [] for texte, tableau in unites: if morceaux and len(morceaux[-1][0]) + len(texte) + 2 <= place: morceaux[-1] = (morceaux[-1][0] + "\n\n" + texte, morceaux[-1][1] or tableau) else: morceaux.append((texte, tableau)) if len(morceaux) < 2: chunk = dict(chunk) if chunk.get("kind") in (None, "article"): chunk["kind"] = "table" chunk["oversized"] = len(content) > target_max_chars return [chunk] out: List[Dict[str, Any]] = [] for i, (texte, tableau) in enumerate(morceaux): piece = dict(chunk) tete = chapeau if i == 0 else prefixe piece["content"] = (tete + "\n" + texte).strip() if tete else texte piece["number"] = f"{chunk['number']}.{i + 1}" piece["parent_id"] = str(chunk["number"]) if tableau and chunk.get("kind") in (None, "article"): piece["kind"] = "table" piece["char_count"] = len(piece["content"]) piece["word_count"] = len(piece["content"].split()) out.append(piece) return out def _split_long_article( chunk: Dict[str, Any], target_max_chars: int = TARGET_MAX_CHARS ) -> List[Dict[str, Any]]: """ R5/R6 — Decoupe un article trop long, sans jamais casser un tableau. Coupe prioritairement aux alineas (1), (2), (3), en repetant l'en-tete de l'article dans chaque morceau pour qu'il reste citable seul. """ content = chunk["content"] if len(content) <= target_max_chars: return [chunk] # Une liste nominative se coupe entre ses lignes, tableau ou non : chaque # ligne y est une personne, complete en elle-meme. if chunk.get("kind") == "roster": return _split_roster(chunk, target_max_chars) # R5 : un tableau markdown se coupe ENTRE ses lignes, son en-tete repete # (voir _split_tableaux). Un tableau HTML, forme de l'ancien extracteur, # reste entier. # # La nature « table » n'etait jamais posee : refine() fixe "article" par # defaut AVANT cet appel, et `get("kind") or "table"` gardait donc # "article". Un roster garde sa nature. if _TABLE_BLOCK.search(content): chunk = dict(chunk) if chunk.get("kind") in (None, "article"): chunk["kind"] = "table" chunk["oversized"] = True return [chunk] if _MD_TABLE_BLOCK.search(content): return _split_tableaux(chunk, target_max_chars) parts = [p.strip() for p in _ALINEA.split(content) if p and p.strip()] if len(parts) < 2: # Sans alineas numerotes, aux paragraphes, puis aux items de liste. # Un article de definitions de 18 000 caracteres restait d'un bloc : # tronque a 10 000 caracteres puis a 2048 jetons, la fin n'etait # jamais vectorisee. parts = [p.strip() for p in re.split(r"\n\s*\n", content) if p.strip()] if len(parts) < 2: parts = [p.strip() for p in re.split(r"\n(?=\s*[-*•]\s)", content) if p.strip()] if len(parts) < 2: return [chunk] head, body = "", parts else: head = parts[0] if not _DEBUT_ALINEA.match(parts[0]) else "" body = parts[1:] if head else parts # Le chapeau est repete devant chaque morceau, TRONQUE a 200 caracteres. Un # chapeau plus long n'existait donc nulle part en entier : tout ce qui # depassait etait perdu. Loi 2010/012 : les articles 69 a 74 al. 2, colles # sur une ligne par l'OCR avant le premier alinea « (3) », reduits a # « …celui qui accede san ». Un chapeau long ouvre desormais le premier # morceau EN ENTIER (aux paragraphes s'il depasse lui-meme la taille # cible), et seul son debut est repete devant les morceaux suivants. chapeau_long = len(head) > LONGUEUR_CHAPEAU_REPETE if chapeau_long: tete = [p.strip() for p in re.split(r"\n\s*\n", head) if p.strip()] if len(head) <= target_max_chars or len(tete) < 2: tete = [head] body = tete + body # Regroupement glouton jusqu'a la taille cible : pas de morceau minuscule # a chaque alinea court, pas de morceau geant non plus. morceaux: List[str] = [] for part in body: if morceaux and len(morceaux[-1]) + len(part) + 1 <= target_max_chars: morceaux[-1] += "\n" + part else: morceaux.append(part) # Le chapeau est repete en tete de chaque morceau pour qu'il reste citable # seul, mais tronque : colle entier au premier, un long chapeau faisait un # premier morceau a peine plus petit que l'article. prefix = (head[:LONGUEUR_CHAPEAU_REPETE].strip() + "\n") if head else "" out: List[Dict[str, Any]] = [] for i, morceau in enumerate(morceaux): piece = dict(chunk) # Le premier morceau commence par le chapeau long lui-meme : le # repeter devant lui le doublerait. tete_du_morceau = "" if (chapeau_long and i == 0) else prefix piece["content"] = (tete_du_morceau + morceau).strip() piece["number"] = f"{chunk['number']}.{i + 1}" piece["parent_id"] = str(chunk["number"]) piece["char_count"] = len(piece["content"]) piece["word_count"] = len(piece["content"].split()) out.append(piece) return out or [chunk] def _detacher_signature(chunk: Dict[str, Any]) -> List[Dict[str, Any]]: """ R6 — Detache du dernier article la signature et ce qui la suit. Le decoupeur colle au dernier article tout ce qui le suit : lieu et date, « LE PRESIDENT DE LA REPUBLIQUE », le nom du signataire — et, sur les decrets d'une page, le titre de l'acte, que Docling lit APRES la signature. La formule d'execution depassait alors le seuil de 600 caracteres, n'etait plus reconnue, et partait dans l'index vectoriel avec la signature. Retourne l'article, puis un chunk SIGNATURE hors vectoriel, puis, si une annexe suit, un chunk ANNEXE traite comme un article. Rien n'est supprime. """ content = chunk["content"] m = _SIGNATURE.search(content) if not m or not content[:m.start()].strip(): return [chunk] reste = content[m.start():] annexe = _DEBUT_ANNEXE.search(reste, m.end() - m.start()) morceaux = [ (content[:m.start()], {}), (reste[:annexe.start()] if annexe else reste, {"number": "SIGNATURE", "kind": "signature", "embed": False}), ] if annexe: morceaux.append((reste[annexe.start():], {"number": "ANNEXE"})) out: List[Dict[str, Any]] = [] for texte, champs in morceaux: if not texte.strip(): continue piece = dict(chunk) piece.update(champs) if champs: piece["parent_id"] = str(chunk.get("number", "")) piece["content"] = texte.strip() piece["char_count"] = len(piece["content"]) piece["word_count"] = len(piece["content"].split()) out.append(piece) return out def _est_formule_d_execution(contenu: str) -> bool: """ R6 — Le chunk n'est-il QUE la formule d'execution ? Il porte une expression de la formule, il est court, et hors des paragraphes qui la portent, il ne reste presque aucun mot. Le seul indice de la formule ne suffisait pas : quand l'OCR rate le marqueur de l'article 2, la formule se colle a l'article 1. « La journee du lundi 16 aout 2021 est declaree feriee » partait alors hors de l'index vectoriel avec elle — et le decret entier n'avait plus aucun vecteur. """ if len(contenu) >= 600 or not _BOILERPLATE.search(contenu): return False hors_formule = " ".join( p for p in re.split(r"\n\s*\n", contenu) if p.strip() and not _BOILERPLATE.search(p) ) return len(_MOT_EN_MINUSCULES.findall(hors_formule)) < _MOTS_HORS_FORMULE def _separer_dispositif(chunk: Dict[str, Any]) -> List[Dict[str, Any]]: """ R3 — Detache des visas le texte qui suit « DECRETE : ». Le marqueur « Article 1er » est souvent illisible : le cachet ou le sceau le recouvre dans la marge. text_chunker range alors l'article 1 dans la base legale, avec les visas, hors de l'index vectoriel. Mesure sur le corpus : la moitie des 53 documents qui n'avaient AUCUN vecteur. Ce texte devient un chunk DISPOSITIF, traite comme un article. """ contenu = chunk["content"] fins = [m.end() for m in _DISPOSITIF.finditer(contenu)] mots_min = 3 if not fins: # Sans « DECRETE : » lisible, apres le dernier visa. Plus exigeant : il # faut une vraie phrase, qui ne soit pas la suite d'un visa coupe. visas = list(_VISA_LINE.finditer(contenu)) if not visas: return [chunk] fins, mots_min = [visas[-1].end()], 15 if not re.match(r"\s*[A-ZÀ-Þ0-9«\"|]", contenu[fins[-1]:]): return [chunk] tete, reste = contenu[:fins[-1]].strip(), contenu[fins[-1]:].strip() # Docling lit parfois un dernier visa, ou la date en marge, apres # « DECRETE : » : ce n'est pas le dispositif. if len(_MOT_EN_MINUSCULES.findall(reste)) < mots_min or re.match( r"\s*(?:[-*•]\s*)?Vu\b", reste, re.IGNORECASE ): return [chunk] dispositif = dict(chunk) dispositif.update(number="DISPOSITIF", parent_id="LEGAL_BASIS", content=reste) preambule = dict(chunk) preambule["content"] = tete for c in (preambule, dispositif): c["char_count"] = len(c["content"]) c["word_count"] = len(c["content"].split()) return [preambule, dispositif] def _extract_citations(legal_basis: str) -> List[str]: """ R3 — Parse les visas en graphe de citations. "Vu le decret n° 2011/412 du 09 decembre 2011 portant reorganisation..." devient une arete du graphe. C'est une fonctionnalite produit ("quels textes s'appuient sur cette loi ?") et ca ne coute rien puisqu'on a deja le texte. """ refs: List[str] = [] visas = [v for bloc in _VISA_LINE.findall(legal_basis) for v in re.split(r";\s*Vu\b", bloc)] for visa in visas: for kind, num in _CITED_REF.findall(visa): ref = f"{kind.strip().lower()} n° {num.strip()}" if ref not in refs: refs.append(ref) return refs def _dedupe(chunks: List[Dict[str, Any]]) -> Tuple[List[Dict[str, Any]], int]: """R7 — Fusionne les chunks au contenu strictement identique.""" seen: Dict[str, int] = {} out: List[Dict[str, Any]] = [] removed = 0 for c in chunks: key = hashlib.sha256( re.sub(r"\s+", " ", c["content"]).strip().lower().encode("utf-8") ).hexdigest() if key in seen: removed += 1 continue seen[key] = 1 out.append(c) return out, removed # ==================== POINT D'ENTREE ==================== def refine( chunks: List[Dict[str, Any]], context: DocumentContext, target_max_chars: int = TARGET_MAX_CHARS, ) -> RefinedChunks: """ Applique les regles R2-R7 aux chunks bruts de text_chunker. Args: chunks: Sortie de extract_articles() context: Metadonnees du document (reference, titre, date, categorie) target_max_chars: Taille cible avant decoupe aux alineas Returns: RefinedChunks — `.embeddable` donne les seuls chunks a vectoriser """ assert isinstance(chunks, list), "chunks doit etre une liste" assert context is not None, "context requis" result = RefinedChunks() working: List[Dict[str, Any]] = [] entrees: List[Dict[str, Any]] = [] for raw in chunks: if str(raw.get("number", "")) == "LEGAL_BASIS": entrees.extend(_separer_dispositif(dict(raw))) else: entrees.append(raw) for raw in entrees: chunk = dict(raw) number = str(chunk.get("number", "")) # Le motif d'article consomme "Article 5." mais laisse le tiret de # "Article 5.-" en tete du contenu. Cosmetique, mais ce tiret orphelin # se retrouverait dans chaque citation affichee a l'utilisateur. chunk["content"] = re.sub(r"^\s*[-–—.:]\s*", "", chunk["content"]).strip() chunk["char_count"] = len(chunk["content"]) chunk["word_count"] = len(chunk["content"].split()) # R3 — visas hors index vectoriel, convertis en citations. # # Attention : text_chunker etiquette LEGAL_BASIS TOUT le texte precedant # le premier article. Sur une page qui ne commence pas par un article # — cas courant au milieu d'un document — c'est la suite de l'article # precedent, donc du contenu juridique reel. L'exclure de l'index ferait # disparaitre silencieusement le debut de chaque page de la recherche. # On ne classe en base legale que si des visas "Vu ..." sont presents. if number == "LEGAL_BASIS": if _VISA_LINE.search(chunk["content"]): result.legal_basis = chunk["content"] result.citations = _extract_citations(chunk["content"]) chunk["kind"] = "legal_basis" chunk["embed"] = False elif _BOILERPLATE.search(chunk["content"]) and len(chunk["content"]) < 600: # Titre de la loi et formule de promulgation, sans visas chunk["kind"] = "boilerplate" chunk["embed"] = False elif not _MOT_EN_MINUSCULES.search(chunk["content"]): chunk["kind"] = "fragment" chunk["embed"] = False else: # Suite d'une page : du contenu reel, coupe comme un article chunk["kind"] = "continuation" chunk["embed"] = True working.extend(_split_long_article(chunk, target_max_chars)) continue working.append(chunk) continue if number == "PREAMBULE": chunk["kind"] = "preamble" chunk["embed"] = True working.append(chunk) continue for piece in _detacher_signature(chunk): if piece.get("kind") == "signature": working.append(piece) continue # R4 — listes nominatives piece, entries = _apply_roster_rule(piece) if entries: result.roster.extend(entries) # R6 — articles d'execution : conserves, mais hors vectoriel if _est_formule_d_execution(piece["content"]): piece["kind"] = "boilerplate" piece["embed"] = False working.append(piece) continue piece.setdefault("kind", "annexe" if piece.get("number") == "ANNEXE" else "article") piece["embed"] = True # Chapeau de section ou repli par paragraphes fait de seuls titres # en capitales : conserve, hors vectoriel. Un tableau en capitales # (annexe budgetaire) reste vectorise. if ( str(piece.get("number", "")).upper().startswith(("SECTION_", "PARA_")) and not _MOT_EN_MINUSCULES.search(piece["content"]) and not _contains_table(piece["content"]) ): piece["kind"] = "fragment" piece["embed"] = False working.append(piece) continue # R5/R6 — decoupe des articles trop longs working.extend(_split_long_article(piece, target_max_chars)) # R6 — chunks trop courts : conserves, mais hors vectoriel. Un vrai article # a son propre seuil, bien plus bas : court ne veut pas dire vide de sens. for chunk in working: numero = str(chunk.get("number") or "") seuil = ( MIN_CHARS_ARTICLE if _REAL_ARTICLE_NUMBER.match(numero) or numero.startswith("DISPOSITIF") else MIN_CHARS ) if chunk.get("embed") and len(chunk["content"]) < seuil: chunk["embed"] = False chunk["kind"] = "fragment" # R7 — deduplication working, removed = _dedupe(working) # R2 — contextualisation : embed_text = en-tete + contenu for position, chunk in enumerate(working): chunk["position"] = position if not chunk.get("embed"): chunk["embed_text"] = None continue header = context.header( page=chunk.get("page_number"), section=chunk.get("section") ) # Ne labelliser "Article X" que si X est un vrai numero d'article. # PARA_n est le repli par paragraphes de text_chunker : l'annoncer # comme un article ferait citer au chatbot des references inexistantes. number = str(chunk.get("number") or "") label = f"Article {number}" if _REAL_ARTICLE_NUMBER.match(number) else "" # Les tableaux sont vectorises en texte delimite, pas en HTML brut ni # avec la ligne de separation markdown. Une liste nominative l'est # sous sa forme resumee (R4). body = chunk.get("embed_body") or chunk["content"] body = _TABLE_BLOCK.sub(lambda m: _table_to_text(m.group(0)), body) body = _MD_TABLE_BLOCK.sub(lambda m: _md_table_to_text(m.group(0)) + "\n", body) chunk["embed_text"] = "\n".join( part for part in (header, "———", label, body) if part ).strip() result.chunks = working result.stats = { "chunks_in": len(chunks), "chunks_out": len(working), "embeddable": len(result.embeddable), "roster_entries": len(result.roster), "citations": len(result.citations), "duplicates_removed": removed, "kinds": { k: sum(1 for c in working if c.get("kind") == k) for k in sorted({c.get("kind", "article") for c in working}) }, } logger.info( f"🔧 Raffinage {context.reference}: {len(chunks)} → {len(working)} chunks, " f"{len(result.embeddable)} a vectoriser " f"({len(result.roster)} personnes, {len(result.citations)} citations)" ) return result |