app/postprocess.py

"""Post-traitement métier des lectures du CRNN : normalisation numérique,
calcul du montant/litres manquant, vérification de cohérence.

Contrairement au pipeline OCR générique (qui doit deviner quel nombre
correspond à quel champ par proximité de libellé ou par magnitude), le CRNN
étiquette déjà chaque ligne lue ("prix" = montant, "volume" = litres,
"prix_litre" = prix unitaire affiché) grâce au découpage positionnel — il
n'y a donc pas d'ambiguïté champ/valeur à résoudre ici.
"""
from typing import Any, Dict, List, Optional
import re

from .config import RuleConfig, load_config

NUMBER_RE = re.compile(r"[-+]?[0-9]+[\.,]?[0-9]*")


def _norm_number(s: str) -> Optional[float]:
    """Normalise un nombre lu par le CRNN (virgule/point, espaces) en float."""
    if not s:
        return None
    s = s.strip().replace(" ", "").replace("\xa0", "")
    m = NUMBER_RE.search(s)
    if not m:
        return None
    s = m.group(0)
    if "," in s and "." in s:
        s = s.replace(",", "")
    elif s.count(",") == 1 and s.count(".") == 0:
        s = s.replace(",", ".")
    try:
        return float(s)
    except ValueError:
        return None


def evaluate_consistency(liters: Optional[float], amount: Optional[float],
                         price: Optional[float], tol: Optional[float] = None,
                         cfg: Optional[RuleConfig] = None) -> Dict[str, Any]:
    """Vérifie montant == litres x prix (§5.5/§10 du cahier des charges)."""
    if cfg is None:
        cfg = load_config()
    if tol is None:
        tol = cfg.consistency_tolerance

    out = {"is_consistent": None, "calculated_amount": None, "calculated_liters": None}
    if liters is not None and price is not None:
        calc_amt = liters * price
        out["calculated_amount"] = round(calc_amt, 2)
        if amount is not None:
            out["is_consistent"] = abs(amount - calc_amt) / max(1.0, calc_amt) <= tol
    if amount is not None and price is not None and liters is None:
        calc_l = amount / price if price != 0 else None
        out["calculated_liters"] = round(calc_l, 2) if calc_l is not None else None
    return out


def process(recognized_fields: List[Dict[str, Any]],
           fuel_price: Optional[float] = None,
           cfg: Optional[RuleConfig] = None) -> Dict[str, Any]:
    """Convertit les lignes lues par le CRNN en résultat métier structuré.

    Args:
        recognized_fields: sortie de `recognizer.recognize_screen` :
            liste de {'field': 'prix'|'volume'|'prix_litre', 'text', 'confidence'}
        fuel_price: prix du litre configuré côté YELY (fait toujours autorité
            sur un prix lu à l'écran — règle métier n°1 du cahier des charges).
    """
    if cfg is None:
        cfg = load_config()

    by_field = {f["field"]: f for f in recognized_fields if f.get("field")}

    amount_val = _norm_number(by_field.get("prix", {}).get("text", ""))
    liters_val = _norm_number(by_field.get("volume", {}).get("text", ""))
    screen_price_val = _norm_number(by_field.get("prix_litre", {}).get("text", ""))

    # Le prix configuré côté YELY fait autorité ; le prix lu à l'écran n'est
    # utilisé que si l'appelant n'en a fourni aucun.
    price_val = fuel_price if fuel_price is not None else screen_price_val

    evalr = evaluate_consistency(liters_val, amount_val, price_val, cfg=cfg)

    confidences = [f["confidence"] for f in recognized_fields if f.get("confidence") is not None]
    ocr_confidence = (sum(confidences) / len(confidences)) if confidences else None

    raw_numbers = [(f["text"], v) for f, v in (
        (by_field.get("prix", {}), amount_val),
        (by_field.get("volume", {}), liters_val),
        (by_field.get("prix_litre", {}), screen_price_val),
    ) if v is not None]

    return {
        "detected_liters": liters_val,
        "detected_amount": amount_val,
        "fuel_price": price_val,
        "calculated_amount": evalr["calculated_amount"],
        "calculated_liters": evalr["calculated_liters"],
        "is_consistent": evalr["is_consistent"],
        "ocr_confidence": round(ocr_confidence, 2) if ocr_confidence is not None else None,
        "field_confidences": {
            "liters": by_field.get("volume", {}).get("confidence"),
            "amount": by_field.get("prix", {}).get("confidence"),
            "price": by_field.get("prix_litre", {}).get("confidence"),
        },
        "raw_numbers": raw_numbers,
    }