"""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, }