hotfix: restore synthesis module

src/synthesis.py

@@ -0,0 +1,400 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+"""
+synthesis.py — Mode SYNTHÈSE extractive (déterministe, sans reformulation)
+
+But :
+- Produire une synthèse lisible ET juridiquement sûre (zéro hallucination)
+- On ne "rédige" rien : on sélectionne des extraits exacts, non tronqués
+- On affiche les extraits dans l'ordre du texte
+
+Stratégie :
+1) Découper l'article en "unités juridiques" (items I/II/1°/a)/-, sinon clauses ;, sinon phrases)
+2) Scorer les unités sur des marqueurs normatifs (obligation, interdiction, conditions, exceptions, structure)
+3) Sélectionner 3-5 unités max avec diversité + couverture
+4) Budget global : on supprime des unités si nécessaire (jamais de troncature)
+
+API publique :
+- extractive_summary(article_id, article_text, cfg=None) -> str
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import List, Tuple, Dict, Set
+import re
+
+
+# ==================== CONFIG ====================
+
+@dataclass(frozen=True)
+class SynthesisConfig:
+    # Option A : 5 puces max, aucun tronquage
+    max_bullets: int = 5
+
+    # Budget global (caractères) : on retire des unités si on dépasse
+    max_total_chars: int = 1500
+
+    # Filtres
+    min_unit_len: int = 35
+    max_unit_len_soft: int = 900  # si une unité est trop longue, on essaie de la re-split (sans tronquer)
+
+    # Diversité : seuil de similarité (0 = jamais similaire, 1 = identique)
+    max_jaccard_similarity: float = 0.55
+
+    # Couverture : essayer de prendre au moins 1 unité de chaque catégorie si possible
+    enforce_coverage: bool = True
+
+
+# ==================== TEXT NORMALIZATION ====================
+
+_SPACE_RE = re.compile(r"\s+")
+
+def _norm(s: str) -> str:
+    return _SPACE_RE.sub(" ", s.strip())
+
+
+# ==================== DETECT STRUCTURE / ITEMS ====================
+
+_ITEM_RE = re.compile(
+    r"""^(
+        (?:[IVX]{1,6}\.) |           # I. II. III. IV. ...
+        (?:\d+°) |                   # 1° 2° 3° ...
+        (?:[a-z]\)) |                # a) b) c) ...
+        (?:-\s)                      # - ...
+    )""",
+    re.VERBOSE
+)
+
+def _is_structured_item(line: str) -> bool:
+    return bool(_ITEM_RE.match(line))
+
+
+# ==================== SPLITTING INTO LEGAL UNITS ====================
+
+_SENTENCE_SPLIT_RE = re.compile(r"(?<=[.!?])\s+")
+# Split on semicolons, keeping semantic clauses. We'll keep ";" in the left clause for readability.
+_SEMI_SPLIT_RE = re.compile(r"\s*;\s*")
+
+def _merge_wrapped_lines(lines: List[str]) -> List[str]:
+    """
+    Heuristique simple :
+    - si une ligne n'est pas un item structuré et semble être la continuation de la précédente,
+      on la concatène.
+    """
+    merged: List[str] = []
+    for ln in lines:
+        ln = _norm(ln)
+        if not ln:
+            continue
+        if not merged:
+            merged.append(ln)
+            continue
+
+        prev = merged[-1]
+
+        # Nouveau bloc si item structuré ou si ça ressemble à un titre
+        if _is_structured_item(ln):
+            merged.append(ln)
+            continue
+
+        # Si la ligne précédente se termine par ":" ou "—" ou si la nouvelle commence par une minuscule,
+        # on considère souvent que c'est une continuation.
+        if prev.endswith((":","—","-")) or (ln and ln[0].islower()):
+            merged[-1] = _norm(prev + " " + ln)
+        else:
+            # Sinon, on garde séparé
+            merged.append(ln)
+
+    return merged
+
+
+def _split_long_unit(unit: str, cfg: SynthesisConfig) -> List[str]:
+    """
+    Si l'unité est très longue, on essaie de la découper SANS tronquer :
+    - priorité au ';' (souvent structure juridique)
+    - sinon, phrases complètes
+    """
+    unit = _norm(unit)
+    if len(unit) <= cfg.max_unit_len_soft:
+        return [unit]
+
+    # 1) Essai sur ';'
+    parts = [p.strip() for p in _SEMI_SPLIT_RE.split(unit) if p.strip()]
+    if len(parts) >= 2:
+        out: List[str] = []
+        for i, p in enumerate(parts):
+            # On remet le ';' entre clauses sauf la dernière
+            if i < len(parts) - 1 and not p.endswith(";"):
+                p = p + ";"
+            out.append(_norm(p))
+        # Filtre : on ne veut pas de mini-clause vide
+        out2 = [p for p in out if len(p) >= cfg.min_unit_len]
+        if out2:
+            return out2
+
+    # 2) Fallback phrases
+    sents = [s.strip() for s in _SENTENCE_SPLIT_RE.split(unit) if s.strip()]
+    if len(sents) >= 2:
+        sents2 = [ _norm(s) for s in sents if len(s) >= cfg.min_unit_len ]
+        if sents2:
+            return sents2
+
+    # 3) Dernier recours : on garde l'unité entière (même longue). On préfèrera en éliminer au budget.
+    return [unit]
+
+
+def _split_into_units(article_text: str, cfg: SynthesisConfig) -> List[str]:
+    if not article_text:
+        return []
+
+    raw_lines = [ln for ln in article_text.splitlines()]
+    # enlève les lignes vides
+    raw_lines = [ln for ln in raw_lines if _norm(ln)]
+
+    lines = _merge_wrapped_lines(raw_lines)
+
+    units: List[str] = []
+
+    for ln in lines:
+        ln = _norm(ln)
+        if not ln:
+            continue
+
+        # Si c'est un item structuré, on le garde en tant qu'unité,
+        # mais on peut le re-split s'il est gigantesque.
+        if _is_structured_item(ln):
+            units.extend(_split_long_unit(ln, cfg))
+            continue
+
+        # Sinon, on essaye d'abord de couper par ';' si c'est long
+        if len(ln) > cfg.max_unit_len_soft:
+            units.extend(_split_long_unit(ln, cfg))
+        else:
+            units.append(ln)
+
+    # Filtrage final
+    cleaned = []
+    for u in units:
+        u = _norm(u)
+        if len(u) < cfg.min_unit_len:
+            continue
+        cleaned.append(u)
+
+    return cleaned
+
+
+# ==================== SCORING (NORMATIVE / LEGAL) ====================
+
+# Patterns "durs" (évite le bruit : on NE met PAS "est/sont")
+_PATTERNS: Dict[str, List[re.Pattern]] = {
+    "obligation": [
+        re.compile(r"\bdoit\b", re.I),
+        re.compile(r"\bdoivent\b", re.I),
+        re.compile(r"\best\s+tenu\b", re.I),
+        re.compile(r"\bsont\s+tenus\b", re.I),
+        re.compile(r"\best\s+tenu\s+de\b", re.I),
+        re.compile(r"\bobligatoire\b", re.I),
+        re.compile(r"\bobligation\b", re.I),
+    ],
+    "prohibition": [
+        re.compile(r"\binterdit\b", re.I),
+        re.compile(r"\best\s+interdit\b", re.I),
+        re.compile(r"\bil\s+est\s+interdit\b", re.I),
+        re.compile(r"\bne\s+peut\b", re.I),
+        re.compile(r"\bne\s+peuvent\b", re.I),
+    ],
+    "condition": [
+        re.compile(r"\bsi\b", re.I),
+        re.compile(r"\blorsque\b", re.I),
+        re.compile(r"\bà\s+condition\b", re.I),
+        re.compile(r"\ba\s+condition\b", re.I),
+        re.compile(r"\ben\s+cas\b", re.I),
+        re.compile(r"\bdans\s+le\s+cas\b", re.I),
+    ],
+    "exception": [
+        re.compile(r"\bsauf\b", re.I),
+        re.compile(r"\btoutefois\b", re.I),
+        re.compile(r"\bsous\s+réserve\b", re.I),
+        re.compile(r"\bnonobstant\b", re.I),
+    ],
+    "permission": [
+        re.compile(r"\bpeut\b", re.I),
+        re.compile(r"\bpeuvent\b", re.I),
+    ],
+    "structure": [
+        re.compile(r"^(?:[IVX]{1,6}\.|(?:\d+°)|(?:[a-z]\))|(?:-\s))", re.I),
+    ]
+}
+
+_REF_RE = re.compile(r"\b(décret|arrêté|loi|code)\b", re.I)
+
+def _score_unit(u: str) -> Tuple[int, Set[str]]:
+    """
+    Retourne (score, tags)
+    tags = catégories rencontrées (obligation, prohibition, condition, exception, permission, structure)
+    """
+    score = 0
+    tags: Set[str] = set()
+
+    for tag, pats in _PATTERNS.items():
+        for p in pats:
+            if p.search(u):
+                tags.add(tag)
+                # pondération simple et justifiable
+                if tag in ("obligation", "prohibition"):
+                    score += 4
+                elif tag in ("condition", "exception"):
+                    score += 3
+                elif tag == "structure":
+                    score += 2
+                elif tag == "permission":
+                    score += 1
+                break
+
+    if _REF_RE.search(u):
+        score += 1
+
+    # pénalité légère si c'est extrêmement long (sans tronquer, juste pour favoriser concision)
+    if len(u) > 700:
+        score -= 1
+
+    return score, tags
+
+
+# ==================== DIVERSITY (ANTI-REDUNDANCY) ====================
+
+_WORD_RE = re.compile(r"[a-zA-ZÀ-ÖØ-öø-ÿ0-9]+")
+
+_STOPWORDS = {
+    # très minimaliste : juste assez pour éviter le bruit
+    "le","la","les","un","une","des","du","de","d","à","au","aux","et","ou",
+    "en","dans","sur","pour","par","avec","sans","ce","cet","cette","ces",
+    "qui","que","quoi","dont","où","il","elle","ils","elles","on",
+    "est","sont","été","être","a","ont","avait","avaient","sera","seront",
+    "ne","pas","plus","moins","se","sa","son","ses","leur","leurs",
+}
+
+def _keywords(u: str) -> Set[str]:
+    toks = [t.lower() for t in _WORD_RE.findall(u)]
+    return {t for t in toks if len(t) >= 3 and t not in _STOPWORDS}
+
+def _jaccard(a: Set[str], b: Set[str]) -> float:
+    if not a and not b:
+        return 0.0
+    inter = len(a & b)
+    union = len(a | b)
+    return inter / union if union else 0.0
+
+
+# ==================== SELECTION ====================
+
+def _pick_with_coverage(
+    scored_units: List[Tuple[int, int, str, Set[str], Set[str]]],
+    cfg: SynthesisConfig
+) -> List[Tuple[int, str]]:
+    """
+    scored_units elements: (score, idx, unit_text, tags, keywords)
+    returns list of (idx, unit_text) selected
+    """
+
+    # candidates sorted by score desc
+    candidates = sorted(scored_units, key=lambda x: x[0], reverse=True)
+
+    selected: List[Tuple[int, str, Set[str]]] = []  # (idx, text, keywords)
+
+    def can_add(kw: Set[str]) -> bool:
+        for (_, _, kw2) in selected:
+            if _jaccard(kw, kw2) >= cfg.max_jaccard_similarity:
+                return False
+        return True
+
+    def add_candidate(cand: Tuple[int, int, str, Set[str], Set[str]]) -> bool:
+        score, idx, text, tags, kw = cand
+        if score <= 0:
+            return False
+        if not can_add(kw):
+            return False
+        selected.append((idx, text, kw))
+        return True
+
+    # 1) Coverage pass (si activé)
+    if cfg.enforce_coverage:
+        # catégories prioritaires
+        targets = [
+            ("obligation",),
+            ("prohibition",),
+            ("exception", "condition"),
+            ("structure",),
+        ]
+        for group in targets:
+            if len(selected) >= cfg.max_bullets:
+                break
+            for cand in candidates:
+                score, idx, text, tags, kw = cand
+                if score <= 0:
+                    continue
+                if any(t in tags for t in group):
+                    if add_candidate(cand):
+                        break
+
+    # 2) Fill remaining by best score
+    for cand in candidates:
+        if len(selected) >= cfg.max_bullets:
+            break
+        add_candidate(cand)
+
+    # sort by original order (idx)
+    selected_sorted = sorted(((idx, text) for (idx, text, _) in selected), key=lambda x: x[0])
+    return selected_sorted
+
+
+def _apply_budget(selected: List[Tuple[int, str]], cfg: SynthesisConfig) -> List[str]:
+    """
+    Applique le budget max_total_chars en supprimant des unités (jamais tronquer).
+    On supprime en priorité les dernières (car on affiche dans l'ordre).
+    """
+    texts = [t for (_, t) in selected]
+    # +2 per bullet for "- " and newline approx (rough, OK)
+    def total_len(ts: List[str]) -> int:
+        return sum(len(s) for s in ts) + 3 * len(ts)
+
+    while texts and total_len(texts) > cfg.max_total_chars:
+        texts.pop()  # remove last
+    return texts
+
+
+# ==================== PUBLIC API ====================
+
+def extractive_summary(article_id: str, article_text: str, cfg: SynthesisConfig | None = None) -> str:
+    cfg = cfg or SynthesisConfig()
+
+    units = _split_into_units(article_text, cfg)
+
+    if not units:
+        return f"Synthèse impossible : texte vide ou non exploitable.\n\nArticles cités : {article_id}"
+
+    scored_units: List[Tuple[int, int, str, Set[str], Set[str]]] = []
+    for idx, u in enumerate(units):
+        score, tags = _score_unit(u)
+        kw = _keywords(u)
+        scored_units.append((score, idx, u, tags, kw))
+
+    selected = _pick_with_coverage(scored_units, cfg)
+
+    # si rien (ex: aucun score > 0), fallback : premières unités (mais toujours non tronquées + ordre)
+    if not selected:
+        selected = [(i, units[i]) for i in range(min(cfg.max_bullets, len(units)))]
+
+    texts = _apply_budget(selected, cfg)
+    if not texts:
+        # Si le budget est trop strict et qu'une seule unité est énorme, on affiche quand même la 1ère
+        # (option produit : mieux vaut 1 extrait entier que rien)
+        texts = [selected[0][1]]
+
+    out_lines = ["Synthèse (extraits du texte, sans reformulation) :"]
+    for t in texts:
+        out_lines.append(f"- {t}")
+
+    return "\n".join(out_lines) + f"\n\nArticles cités : {article_id}"