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

"""Extraction des températures mentionnées dans une étape de recette — 3ᵉ
type de métadonnée de clause, à côté des techniques et des ustensiles (voir
`LocalePipeline.process`).

Contrairement aux deux autres, ce n'est pas un `PhraseMatcher` : une
température est numérique/motif (« 180 °C », « thermostat 6 »,
« feu doux »), pas un mot d'un vocabulaire fini. De simples expressions
régulières suffisent et n'ont besoin ni du modèle spaCy ni de
l'entraînement.

Trois formes, non exclusives (une clause peut n'en avoir aucune) :

- `celsius` — une valeur en degrés Celsius (« 180 °C », « 180 degrés »,
  « 180° »). Une valeur en Fahrenheit (« 350 °F », rare en FR) est
  convertie et arrondie.
- `gas_mark` — un numéro de thermostat / gas mark (« th. 6 »,
  « thermostat 7 », « gas mark 4 »).
- `qualitative` — une intensité de feu qualitative, normalisée en
  `"low" | "medium" | "high"` (« feu doux/moyen/vif/fort »,
  « low/medium/high heat »).

Les offsets `[start, end)` suivent la même convention que `Entity`
(`String.prototype.slice` côté `apps/api`).
"""

import re
from dataclasses import dataclass

# Intensités qualitatives -> échelle normalisée à 3 niveaux.
_QUALITATIVE_FR: dict[str, str] = {
    "doux": "low",
    "moyen": "medium",
    "moyen-vif": "medium",
    "vif": "high",
    "fort": "high",
}
_QUALITATIVE_EN: dict[str, str] = {
    "low": "low",
    "medium": "medium",
    "medium-high": "medium",
    "high": "high",
}

# `re.IGNORECASE | re.UNICODE` sur tous les motifs — le texte d'entrée n'est
# pas normalisé ici (contrairement au `PhraseMatcher`), les accents de
# « degrés » comptent donc, d'où `degr[ée]s?`.
_FLAGS = re.IGNORECASE | re.UNICODE

# Fahrenheit d'abord (motif plus spécifique : le `f` explicite), sinon un
# nombre suivi d'un signe degré / « degré(s) » est lu en Celsius.
_FAHRENHEIT_RE = re.compile(r"(\d{2,3})\s*(?:°\s*f|degr[ée]s?\s*fahrenheit|°f)\b", _FLAGS)
_CELSIUS_RE = re.compile(
    r"(\d{2,3})\s*(?:°\s*c\b|°(?!\s*f)|degr[ée]s?(?:\s*(?:celsius|c))?\b)",
    _FLAGS,
)
_GAS_MARK_RE = re.compile(r"\b(?:th\.?|thermostat|gas\s*mark)\s*(\d{1,2})\b", _FLAGS)
_QUALITATIVE_RE = re.compile(
    r"\b(?:feu\s+(doux|moyen-vif|moyen|vif|fort)"
    r"|(low|medium-high|medium|high)\s+heat)\b",
    _FLAGS,
)


@dataclass
class Temperature:
    """Une température trouvée dans `text`. Au moins un de `celsius` /
    `gas_mark` / `qualitative` est non nul ; `raw` est le fragment exact
    reconnu (`text[start:end]`)."""

    start: int
    end: int
    celsius: float | None
    gas_mark: int | None
    qualitative: str | None
    raw: str


def _fahrenheit_to_celsius(fahrenheit: float) -> float:
    return round((fahrenheit - 32) * 5 / 9)


def extract_temperatures(text: str, locale: str) -> list[Temperature]:
    """Toutes les températures de `text`, ordonnées par position, sans
    chevauchement (le fragment le plus long gagne à position égale).
    `locale` n'influe que sur le dictionnaire qualitatif retenu — les
    motifs numériques sont communs."""
    if not text.strip():
        return []

    qualitative_map = _QUALITATIVE_EN if locale == "en" else _QUALITATIVE_FR
    found: list[Temperature] = []

    for match in _FAHRENHEIT_RE.finditer(text):
        found.append(
            Temperature(
                start=match.start(),
                end=match.end(),
                celsius=_fahrenheit_to_celsius(int(match.group(1))),
                gas_mark=None,
                qualitative=None,
                raw=match.group(0),
            )
        )

    fahrenheit_spans = [(temp.start, temp.end) for temp in found]

    for match in _CELSIUS_RE.finditer(text):
        # Un « 350 °F » a déjà été capté par la passe Fahrenheit — ne pas le
        # relire en Celsius sur le nombre seul.
        if any(start <= match.start() < end for start, end in fahrenheit_spans):
            continue
        found.append(
            Temperature(
                start=match.start(),
                end=match.end(),
                celsius=float(int(match.group(1))),
                gas_mark=None,
                qualitative=None,
                raw=match.group(0),
            )
        )

    for match in _GAS_MARK_RE.finditer(text):
        found.append(
            Temperature(
                start=match.start(),
                end=match.end(),
                celsius=None,
                gas_mark=int(match.group(1)),
                qualitative=None,
                raw=match.group(0),
            )
        )

    for match in _QUALITATIVE_RE.finditer(text):
        raw_level = (match.group(1) or match.group(2) or "").lower()
        level = qualitative_map.get(raw_level) or _QUALITATIVE_FR.get(raw_level) or _QUALITATIVE_EN.get(raw_level)
        if level is None:
            continue
        found.append(
            Temperature(
                start=match.start(),
                end=match.end(),
                celsius=None,
                gas_mark=None,
                qualitative=level,
                raw=match.group(0),
            )
        )

    # Résolution des chevauchements : à position de départ égale (ou
    # imbrication), on garde le fragment le plus long, puis on avance.
    found.sort(key=lambda temp: (temp.start, -(temp.end - temp.start)))
    resolved: list[Temperature] = []
    last_end = -1
    for temp in found:
        if temp.start < last_end:
            continue
        resolved.append(temp)
        last_end = temp.end
    return resolved