batchCooking/services/tech-step-intent-service/intent_service/locale_pipeline.py

"""Pipeline spaCy pour UNE locale — l'équivalent Python de ce que
`node-nlp`'s `NlpManager` faisait pour cette locale dans
`TechStepClassifierService` (`apps/api/src/lib/recipe-matching/tech-step-matcher.ts`) :
NER par entités enum (ici un `PhraseMatcher`) + classification d'intention
(ici un `textcat`), les deux entraînés à partir du même corpus
(`TECH_STEP_TRAINING_DATA`, poussé par `apps/api` via `POST /v1/train`).

Le modèle de base spaCy (tokenizer + vecteurs + le composant
`diacritics_normalizer` défini plus bas) est chargé une seule fois
(`preload()`, appelé au démarrage du process — voir `main.py` — pas
paresseusement au premier `train()`, pour que `GET /health` ne devienne
`200` qu'une fois ce coût payé) puis réutilisé à chaque `train()` : seul le
`textcat` (retiré puis rajouté à neuf) et le `PhraseMatcher` (remplacé) sont
reconstruits à chaque appel, jamais le tokenizer/les vecteurs. Rien n'est
jamais persisté sur disque — même posture que `autoSave`/`autoLoad: false`
sur l'ancien `NlpManager` : `TECH_STEP_TRAINING_DATA` (côté `apps/api`) reste
l'unique source de vérité.
"""

from __future__ import annotations

import random
from dataclasses import dataclass, field

import spacy
from spacy.language import Language
from spacy.matcher import PhraseMatcher
from spacy.tokens import Doc, Span
from spacy.training import Example
from spacy.util import filter_spans, minibatch

from .text_normalization import normalize_text

# Modèle spaCy de base par locale — voir pyproject.toml pour la version
# pinnée exacte. `md` (pas `sm`) : conserve les vecteurs de mots, inutilisés
# par le pipeline v1 (textcat bag-of-words) mais retenus pour l'ambition
# future de similarité sémantique (voir le README de ce service).
SUPPORTED_LOCALES = {
    "fr": "fr_core_news_md",
    "en": "en_core_web_md",
}

# Composants du modèle de base non utilisés par ce pipeline (on ne s'appuie
# ni sur le NER générique de spaCy, ni sur l'analyse syntaxique/morphologique
# — seuls le tokenizer et les vecteurs de mots restent nécessaires) : les
# exclure au chargement évite le coût mémoire/CPU de composants qui ne
# tourneraient jamais.
_EXCLUDED_COMPONENTS = ["parser", "ner", "tagger", "morphologizer", "attribute_ruler", "lemmatizer"]

_TEXTCAT_PIPE_NAME = "textcat"

# Nombre d'itérations d'entraînement du textcat — calibré empiriquement pour
# converger sur un corpus de cette taille (quelques centaines d'utterances
# par locale) sans allonger inutilement le warm-up. À revoir si
# `calibrate-tech-step-threshold.ts` (côté apps/api) montre un F1 anormalement
# bas qui s'améliore avec plus d'itérations.
_TRAINING_ITERATIONS = 30
_TRAINING_BATCH_SIZE = 8
_TRAINING_DROPOUT = 0.2
# Seed fixe — un warm-up reproductible d'un redémarrage à l'autre (même
# corpus en entrée) est préférable à un score qui varie légèrement à chaque
# déploiement pour la même donnée, en particulier pendant la calibration du
# seuil de confiance côté apps/api.
_TRAINING_SEED = 0


@Language.factory("diacritics_normalizer")
def _create_diacritics_normalizer(nlp: Language, name: str) -> "_DiacriticsNormalizer":
    return _DiacriticsNormalizer()


class _DiacriticsNormalizer:
    """Composant de pipeline réécrivant `token.norm_` avec `normalize_text()`
    (le port Python de `normalizeText()` côté `apps/api`) pour chaque token.

    Point clé : ce composant tourne aussi bien sur les `Doc` construits pour
    les *patterns* du `PhraseMatcher` (voir `LocalePipeline.train`) que sur
    le *texte cible* passé à `process()` — les deux passent donc par
    exactement la même normalisation, ce qui garantit qu'un synonyme comme
    "mijoter" matche indifféremment "MIJOTER"/"mijoté"/"Mijotée" dans le
    texte, reproduisant le comportement `ner.threshold: 1` (exact après
    normalisation, sans tolérance floue Levenshtein) de l'ancien `NlpManager`.
    Indépendant des `entries` entraînées — ajouté une seule fois par
    `preload()`, jamais retiré/rajouté par `train()`.
    """

    def __call__(self, doc: Doc) -> Doc:
        for token in doc:
            token.norm_ = normalize_text(token.text)
        return doc


@dataclass
class TrainEntry:
    """Une technique à entraîner pour une locale — miroir de
    `TrainEntryPayload` (`schemas.py`)."""

    uid: str
    synonyms: list[str] = field(default_factory=list)
    utterances: list[str] = field(default_factory=list)


@dataclass
class Entity:
    """Une mention candidate trouvée par le `PhraseMatcher` — offsets
    caractère `[start, end)`, miroir de `EntityPayload` (`schemas.py`)."""

    uid: str
    start: int
    end: int


@dataclass
class ProcessResult:
    """Résultat complet d'un `process()` — miroir de `ProcessResponse`
    (`schemas.py`)."""

    entities: list[Entity]
    intent: str | None
    score: float


class UnsupportedLocaleError(ValueError):
    """`locale` ne correspond à aucun modèle spaCy connu (voir
    `SUPPORTED_LOCALES`) — distinct d'une locale simplement "pas encore
    entraînée" (`LocalePipeline.is_trained is False`), qui n'est pas une
    erreur (voir `process()`)."""


class LocalePipeline:
    """Pipeline spaCy (NER par phrases + textcat) pour une locale donnée.
    Un `PipelineRegistry` (voir `pipeline_registry.py`) en détient une
    instance par locale supportée.
    """

    def __init__(self, locale: str) -> None:
        if locale not in SUPPORTED_LOCALES:
            raise UnsupportedLocaleError(f"Unsupported locale: {locale!r}")
        self._locale = locale
        self._model_name = SUPPORTED_LOCALES[locale]
        # `None` tant que `preload()` n'a pas tourné.
        self._base_nlp: Language | None = None
        # `None` tant qu'aucun `train()` n'a réussi — `process()` traite ça
        # comme "rien à trouver" plutôt qu'une erreur, exactement le
        # comportement testé côté `apps/api` pour "une locale jamais
        # entraînée".
        self._matcher: PhraseMatcher | None = None
        self._trained = False

    @property
    def is_trained(self) -> bool:
        return self._trained

    def preload(self) -> None:
        """Charge le modèle spaCy de base (tokenizer + vecteurs) et le
        composant `diacritics_normalizer` — idempotent, sans effet si déjà
        chargé. Appelé au démarrage du process pour les deux locales
        connues (voir `main.py`), pas paresseusement au premier `train()`.
        """
        if self._base_nlp is not None:
            return
        nlp = spacy.load(self._model_name, exclude=_EXCLUDED_COMPONENTS)
        nlp.add_pipe("diacritics_normalizer", first=True)
        self._base_nlp = nlp

    def train(self, entries: list[TrainEntry]) -> tuple[int, int, int]:
        """Reconstruit le `textcat` et le `PhraseMatcher` de ce pipeline à
        partir de `entries` (le tokenizer/les vecteurs restent ceux chargés
        par `preload()`). Retourne `(label_count, utterance_count,
        synonym_count)` pour la réponse `/v1/train`.

        `entries` vide retombe à `is_trained == False` plutôt que de lever —
        un appelant qui n'a rien à entraîner pour cette locale obtient le
        même comportement que "jamais entraîné", pas une erreur 500.
        """
        self.preload()
        assert self._base_nlp is not None  # garanti par preload() ci-dessus

        if _TEXTCAT_PIPE_NAME in self._base_nlp.pipe_names:
            self._base_nlp.remove_pipe(_TEXTCAT_PIPE_NAME)

        if not entries:
            self._matcher = None
            self._trained = False
            return (0, 0, 0)

        nlp = self._base_nlp
        # `nlp.make_doc()` ne fait tourner *que* le tokenizer, pas les
        # composants du pipeline — le `diacritics_normalizer` ajouté par
        # `preload()` ne tournerait donc jamais sur les `Doc` de patterns
        # s'ils n'étaient construits qu'avec `make_doc()`, alors que
        # `process()` appelle `nlp(text)` (le pipeline complet) sur le texte
        # cible. Sans ce correctif, un synonyme accentué comme "préchauffer"
        # n'aurait jamais matché "PRÉCHAUFFER"/"Préchauffer" : trouvé en
        # calibrant contre les cas exacts de `tech-step-matcher.test.ts`
        # (fr, la locale la plus concernée par les accents) — un synonyme
        # sans diacritique comme "faire fondre" masquait le bug en semblant
        # fonctionner par coïncidence. Appliquer explicitement le même
        # composant aux deux côtés garantit qu'ils passent par la même
        # normalisation.
        diacritics_normalizer = nlp.get_pipe("diacritics_normalizer")

        matcher = PhraseMatcher(nlp.vocab, attr="NORM")
        synonym_count = 0
        for entry in entries:
            if not entry.synonyms:
                continue
            patterns = [diacritics_normalizer(nlp.make_doc(synonym)) for synonym in entry.synonyms]
            matcher.add(entry.uid, patterns)
            synonym_count += len(entry.synonyms)

        # `textcat` (exclusive_classes) exige au moins deux labels (voir
        # spaCy's error E867) — jamais un problème avec le vrai corpus
        # (`TECH_STEP_TRAINING_DATA` a ~27 techniques), mais un `entries` à
        # un seul élément resterait structurellement valide pour le NER
        # seul : ne pas planter, juste ne pas construire de textcat du tout
        # (`process()` retombe alors sur `intent: null` via son garde
        # `if not cats`, exactement comme "rien à classifier").
        examples: list[Example] = []
        if len(entries) >= 2:
            textcat = nlp.add_pipe(
                _TEXTCAT_PIPE_NAME,
                config={
                    "model": {
                        "@architectures": "spacy.TextCatBOW.v3",
                        "exclusive_classes": True,
                        "ngram_size": 1,
                        "no_output_layer": False,
                    },
                },
            )
            for entry in entries:
                textcat.add_label(entry.uid)

            for entry in entries:
                for utterance in entry.utterances:
                    doc = nlp.make_doc(utterance)
                    cats = {other.uid: 0.0 for other in entries}
                    cats[entry.uid] = 1.0
                    examples.append(Example.from_dict(doc, {"cats": cats}))

            rng = random.Random(_TRAINING_SEED)
            if examples:
                optimizer = nlp.initialize(lambda: examples)
                for _ in range(_TRAINING_ITERATIONS):
                    rng.shuffle(examples)
                    for batch in minibatch(examples, size=_TRAINING_BATCH_SIZE):
                        nlp.update(batch, sgd=optimizer, drop=_TRAINING_DROPOUT)
            else:
                # Des `entries` avec des `uid` mais aucune `utterance` nulle
                # part (corpus incomplet) : le textcat a des labels mais rien
                # pour apprendre à les distinguer — toujours initialisé pour
                # rester un pipeline valide ; `process()` renverra alors un
                # score ~uniforme entre labels. Ce n'est pas ce module qui doit
                # juger la qualité du corpus reçu (voir `tech-step-eval-runner.ts`
                # côté apps/api pour ce rôle).
                nlp.initialize()

        self._matcher = matcher
        self._trained = True
        return (len(entries), len(examples), synonym_count)

    def process(self, text: str) -> ProcessResult:
        """Reproduit la forme de `NlpManager.process(locale, text)` : les
        entités candidates (NER) et le verdict du classifieur d'intention
        sur `text` tel quel — que ce soit la description complète ou une
        clause déjà découpée côté `apps/api`, ce module ne le sait pas et ne
        s'en soucie pas, exactement comme l'ancien `NlpManager`.
        """
        if not self._trained or self._base_nlp is None or self._matcher is None or not text.strip():
            return ProcessResult(entities=[], intent=None, score=0.0)

        doc = self._base_nlp(text)

        # A technique's own synonym list can legitimately contain one phrase
        # nested inside another (`melt`'s "fondre" is a literal substring of
        # its own "faire fondre") — the `PhraseMatcher` reports *both* as
        # separate matches at overlapping positions, which without
        # resolution would hand `splitIntoClauses` (apps/api) two candidates
        # for what a human reads as one mention, producing the same
        # techStepId twice in the final result. `filter_spans` keeps only
        # the longest match at each position (so "faire fondre" wins over
        # the "fondre" it contains) — found by a real regression in
        # `tech-step-matcher.test.ts`'s "detects several distinct
        # techniques..." case once this service replaced node-nlp (which
        # apparently resolved this internally; nothing here recreates that
        # by choice, `filter_spans` is spaCy's own documented tool for
        # exactly this "one span per position" problem, e.g. as used for
        # NER-style outputs).
        matched_spans = [
            Span(doc, start, end, label=match_id) for match_id, start, end in self._matcher(doc)
        ]
        entities = sorted(
            (
                Entity(uid=self._base_nlp.vocab.strings[span.label], start=span.start_char, end=span.end_char)
                for span in filter_spans(matched_spans)
            ),
            key=lambda entity: entity.start,
        )

        cats = doc.cats
        if not cats:
            return ProcessResult(entities=entities, intent=None, score=0.0)
        intent = max(cats, key=cats.get)
        return ProcessResult(entities=entities, intent=intent, score=cats[intent])