veye-lalwa/pipeline/rapprochement.py

"""Rapprochement d'un texte collecté avec les textes déjà en base.

Deux voies, dans cet ordre :

1. **Le numéro officiel.** Deux textes portant le même numéro de loi sont le
   même texte, sans discussion. C'est le cas facile, et le plus fréquent une
   fois la promulgation intervenue.
2. **La similarité de titre**, pour les textes non encore promulgués, qui n'ont
   pas de numéro. Le rapport les désigne par un nom d'usage (« loi Riposte »)
   là où les sources officielles emploient l'intitulé complet (« Loi visant à
   offrir des réponses immédiates aux phénomènes troublant l'ordre public… »).
   La comparaison brute échouerait ; on compare donc des titres normalisés, et
   on tient compte des mots significatifs communs.

Trois issues possibles, et c'est important : rapproché, **à signaler** (au-delà
du seuil bas mais en-deçà du seuil haut — un humain tranche), ou nouveau. Le
pipeline ne fusionne jamais sur une correspondance douteuse.
"""

from __future__ import annotations

import functools
import re
from dataclasses import dataclass
from difflib import SequenceMatcher
from enum import StrEnum

import yaml

from pipeline import chemins
from pipeline.parseurs.dates_fr import sans_accents

# Mots vides et vocabulaire juridique passe-partout : deux lois quelconques les
# partagent, ils ne prouvent donc aucune parenté.
_MOTS_IGNORES = frozenset(
    """loi organique projet proposition relative relatif visant portant diverses
    dispositions de des du la le les l a au aux et en pour par sur dans un une
    ainsi que qui plus mesures texte article n no ndeg""".split()
)

_NON_ALPHANUM = re.compile(r"[^a-z0-9]+")


class Issue(StrEnum):
    RAPPROCHE = "rapproche"
    A_SIGNALER = "a_signaler"
    NOUVEAU = "nouveau"


@dataclass(slots=True)
class Correspondance:
    """Résultat d'un rapprochement."""

    issue: Issue
    texte_id: str | None = None
    score: float = 0.0
    motif: str = ""

    @property
    def est_certaine(self) -> bool:
        return self.issue is Issue.RAPPROCHE


@functools.lru_cache(maxsize=1)
def _seuils() -> tuple[float, float]:
    configuration = yaml.safe_load(chemins.CONFIG.read_text(encoding="utf-8"))
    rapprochement = configuration.get("rapprochement", {})
    return (
        float(rapprochement.get("seuil_similarite", 0.86)),
        float(rapprochement.get("seuil_signalement", 0.72)),
    )


def normaliser(titre: str) -> str:
    """Réduit un titre à ses mots significatifs, sans accent ni ponctuation."""
    plat = sans_accents(titre).lower()
    mots = [m for m in _NON_ALPHANUM.split(plat) if m and m not in _MOTS_IGNORES]
    return " ".join(mots)


def similarite(gauche: str, droite: str) -> float:
    """Similarité de deux titres, entre 0 et 1.

    Combine deux mesures : le recouvrement des mots significatifs, qui résiste
    aux réordonnancements, et la similarité de séquence, qui capte les
    variantes orthographiques. La plus favorable des deux l'emporte — un titre
    court mais entièrement contenu dans un titre long doit être reconnu.
    """
    a, b = normaliser(gauche), normaliser(droite)
    if not a or not b:
        return 0.0
    if a == b:
        return 1.0

    mots_a, mots_b = set(a.split()), set(b.split())
    communs = mots_a & mots_b
    recouvrement = len(communs) / min(len(mots_a), len(mots_b))
    sequence = SequenceMatcher(None, a, b).ratio()

    return max(recouvrement, sequence)


def rapprocher(
    *,
    titre: str,
    numero_officiel: str | None,
    candidats: list[tuple[str, str, str | None]],
) -> Correspondance:
    """Rapproche un texte collecté des textes en base.

    `candidats` est une liste de `(identifiant, titre, numero_officiel)`.
    """
    seuil_haut, seuil_bas = _seuils()

    if numero_officiel:
        for identifiant, _, numero in candidats:
            if numero and numero == numero_officiel:
                return Correspondance(
                    issue=Issue.RAPPROCHE,
                    texte_id=identifiant,
                    score=1.0,
                    motif=f"numéro officiel identique ({numero_officiel})",
                )

    meilleur: tuple[float, str] | None = None
    for identifiant, titre_candidat, _ in candidats:
        score = similarite(titre, titre_candidat)
        if meilleur is None or score > meilleur[0]:
            meilleur = (score, identifiant)

    if meilleur is None:
        return Correspondance(issue=Issue.NOUVEAU, motif="aucun candidat en base")

    score, identifiant = meilleur
    if score >= seuil_haut:
        return Correspondance(
            issue=Issue.RAPPROCHE,
            texte_id=identifiant,
            score=score,
            motif=f"similarité de titre {score:.2f} ≥ {seuil_haut}",
        )
    if score >= seuil_bas:
        return Correspondance(
            issue=Issue.A_SIGNALER,
            texte_id=identifiant,
            score=score,
            motif=(
                f"similarité de titre {score:.2f}, entre le seuil de signalement "
                f"({seuil_bas}) et celui de rapprochement ({seuil_haut}) — "
                "vérification humaine requise"
            ),
        )
    return Correspondance(
        issue=Issue.NOUVEAU, score=score, motif=f"meilleure similarité {score:.2f}, sous les seuils"
    )