""" flowchart_engine.py - Moteur du logigramme d'aide à la détermination de l'impact carbone. Suit le logigramme JSON pour déterminer quelle valeur d'impact carbone utiliser en fonction de la provenance, du niveau de transformation, et des données disponibles. """ from __future__ import annotations from dataclasses import dataclass, field from typing import List, Optional, Tuple import config import data_loader import llm_service @dataclass class StepLog: """Un pas dans le parcours du logigramme.""" node_id: str question: Optional[str] answer: Optional[str] action: Optional[str] = None result: Optional[str] = None @dataclass class CarbonResult: """Résultat complet de l'évaluation carbone d'une matière première.""" matiere_premiere: str pays_production: Optional[str] pays_transformation: Optional[str] classification: str # "brut" | "transforme" classification_justification: str # Valeur finale impact_kg_co2_eq: Optional[float] = None impact_tonne_co2_eq: Optional[float] = None # conversion en tonnes unite_source: str = "" # "kg CO2 eq / kg" ou "kg CO2 eq / tonne" # Traçabilité source_db: str = "" # "ECOALIM" | "GFLI" intrant_utilise: str = "" # nom exact dans la BDD match_exact: bool = True justification_alternative: Optional[str] = None # Parcours de logique parcours: List[StepLog] = field(default_factory=list) node_resultat: str = "" # node_id du résultat actions_appliquees: List[str] = field(default_factory=list) # Candidats alternatifs (pour affichage comparatif quand match non exact) candidats_alternatifs: List[dict] = field(default_factory=list) erreur: Optional[str] = None def _is_france(pays: Optional[str]) -> bool: """Vérifie si le pays est la France.""" if not pays: return False return pays.lower().strip() in ("france", "fr") def _is_european(pays: Optional[str]) -> bool: """Vérifie si le pays est européen.""" if not pays: return False pays_low = pays.lower().strip() if pays_low in config.EUROPEAN_COUNTRIES_FR: return True pays_iso = config.PAYS_FR_TO_ISO.get(pays_low, "").upper() return pays_iso in config.EUROPEAN_COUNTRIES_ISO def _get_country_iso(pays: Optional[str]) -> Optional[str]: """Convertit un nom de pays FR en code ISO.""" if not pays: return None return config.PAYS_FR_TO_ISO.get(pays.lower().strip()) def _is_name_match(matiere: str, intrant_name: str) -> bool: """ Vérifie si le nom de la matière est une correspondance réelle (mot entier) dans le nom de l'intrant, et non un simple sous-chaîne accidentelle. Délègue à data_loader.is_name_match. """ return data_loader.is_name_match(matiere, intrant_name) # ============================================================================ # Fonctions de résolution par node de résultat # ============================================================================ def _resolve_node_4(matiere: str, result: CarbonResult) -> CarbonResult: """ Node 4 : Provenance inconnue + intrant brut. 1. Valeur la plus défavorable dans GFLI 2. Sinon la plus défavorable dans ECOALIM 3. Sinon valeur GFLI de l'intrant au schéma cultural le plus proche (LLM) """ # Étape 1 : GFLI worst result.actions_appliquees.append("1. Recherche de la valeur la plus défavorable dans GFLI") gfli_worst = data_loader.get_gfli_worst_value(matiere) # Rejeter les faux positifs (ex : "blé" → "blend") if gfli_worst and not _is_name_match(matiere, gfli_worst[1]): result.actions_appliquees.append(f" ⚠ Faux positif rejeté : {gfli_worst[1]}") gfli_worst = None llm_justification = None llm_match_exact = None if not gfli_worst: gfli_smart = llm_service.smart_search_gfli(matiere) if gfli_smart and "valeur_kg_co2_eq_par_tonne" in gfli_smart: llm_match_exact = gfli_smart.get("match_exact", False) llm_justification = gfli_smart.get("justification") base_name = gfli_smart["nom_intrant"].split(",")[0].split("/")[0].strip() gfli_worst = data_loader.get_gfli_worst_value(base_name) if not gfli_worst: # Utiliser directement la valeur du LLM gfli_worst = ( gfli_smart["valeur_kg_co2_eq_par_tonne"], gfli_smart["nom_intrant"], gfli_smart.get("source", "GFLI"), ) if gfli_worst: val, nom, src = gfli_worst result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / tonne de produit" result.source_db = src result.intrant_utilise = nom # Déterminer si le match est exact if llm_match_exact is not None: result.match_exact = llm_match_exact else: result.match_exact = _is_name_match(matiere, nom) if llm_justification: result.justification_alternative = llm_justification result.actions_appliquees.append(f" → Trouvé dans GFLI : {nom} = {val:.2f} kg CO2 eq/t") return result # Étape 2 : ECOALIM worst result.actions_appliquees.append("2. Recherche de la valeur la plus défavorable dans ECOALIM") eco_worst = data_loader.get_ecoalim_worst_value(matiere) # Rejeter les faux positifs if eco_worst and not _is_name_match(matiere, eco_worst[1]): result.actions_appliquees.append(f" ⚠ Faux positif rejeté : {eco_worst[1]}") eco_worst = None llm_justification_eco = None llm_match_exact_eco = None if not eco_worst: eco_smart = llm_service.smart_search_ecoalim(matiere) if eco_smart: llm_match_exact_eco = eco_smart.get("match_exact", False) llm_justification_eco = eco_smart.get("justification") eco_worst = data_loader.get_ecoalim_worst_value( eco_smart["nom_intrant"].split(",")[0].strip() ) if not eco_worst: eco_worst = ( eco_smart["valeur_kg_co2_eq"], eco_smart["nom_intrant"], eco_smart.get("source", "ECOALIM"), ) if eco_worst: val, nom, src = eco_worst result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / kg de produit" result.source_db = src result.intrant_utilise = nom if llm_match_exact_eco is not None: result.match_exact = llm_match_exact_eco else: result.match_exact = _is_name_match(matiere, nom) if llm_justification_eco: result.justification_alternative = llm_justification_eco result.actions_appliquees.append(f" → Trouvé dans ECOALIM : {nom} = {val:.4f} kg CO2 eq/kg") return result # Étape 3 : LLM pour trouver le schéma cultural le plus proche result.actions_appliquees.append("3. Recherche via LLM de l'intrant au schéma cultural le plus proche (GFLI)") gfli_smart = llm_service.smart_search_gfli(matiere) if gfli_smart and "valeur_kg_co2_eq_par_tonne" in gfli_smart: val = gfli_smart["valeur_kg_co2_eq_par_tonne"] result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / tonne de produit" result.source_db = gfli_smart["source"] result.intrant_utilise = gfli_smart["nom_intrant"] result.match_exact = gfli_smart.get("match_exact", False) result.justification_alternative = gfli_smart.get("justification") result.actions_appliquees.append(f" → Via LLM : {gfli_smart['nom_intrant']} = {val:.2f} kg CO2 eq/t") return result result.erreur = f"Aucune valeur trouvée pour '{matiere}' dans GFLI ni ECOALIM." return result def _resolve_node_5(matiere: str, result: CarbonResult) -> CarbonResult: """ Node 5 : Provenance inconnue + intrant transformé. Mêmes étapes que node_4 mais pour un intrant transformé. """ # Étape 1 : GFLI worst result.actions_appliquees.append("1. Recherche de la valeur la plus défavorable pour l'intrant transformé dans GFLI") gfli_worst = data_loader.get_gfli_worst_value(matiere) # Rejeter les faux positifs if gfli_worst and not _is_name_match(matiere, gfli_worst[1]): result.actions_appliquees.append(f" ⚠ Faux positif rejeté : {gfli_worst[1]}") gfli_worst = None llm_justification = None llm_match_exact = None if not gfli_worst: gfli_smart = llm_service.smart_search_gfli(matiere) if gfli_smart and "valeur_kg_co2_eq_par_tonne" in gfli_smart: llm_match_exact = gfli_smart.get("match_exact", False) llm_justification = gfli_smart.get("justification") base_name = gfli_smart["nom_intrant"].split(",")[0].split("/")[0].strip() gfli_worst = data_loader.get_gfli_worst_value(base_name) if not gfli_worst: gfli_worst = ( gfli_smart["valeur_kg_co2_eq_par_tonne"], gfli_smart["nom_intrant"], gfli_smart.get("source", "GFLI"), ) if gfli_worst: val, nom, src = gfli_worst result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / tonne de produit" result.source_db = src result.intrant_utilise = nom if llm_match_exact is not None: result.match_exact = llm_match_exact else: result.match_exact = _is_name_match(matiere, nom) if llm_justification: result.justification_alternative = llm_justification result.actions_appliquees.append(f" → Trouvé dans GFLI : {nom} = {val:.2f} kg CO2 eq/t") return result # Étape 2 : ECOALIM worst result.actions_appliquees.append("2. Recherche de la valeur la plus défavorable dans ECOALIM") eco_worst = data_loader.get_ecoalim_worst_value(matiere) if eco_worst and not _is_name_match(matiere, eco_worst[1]): result.actions_appliquees.append(f" ⚠ Faux positif rejeté : {eco_worst[1]}") eco_worst = None llm_justification_eco = None llm_match_exact_eco = None if not eco_worst: eco_smart = llm_service.smart_search_ecoalim(matiere) if eco_smart: llm_match_exact_eco = eco_smart.get("match_exact", False) llm_justification_eco = eco_smart.get("justification") eco_worst = data_loader.get_ecoalim_worst_value( eco_smart["nom_intrant"].split(",")[0].strip() ) if not eco_worst: eco_worst = ( eco_smart["valeur_kg_co2_eq"], eco_smart["nom_intrant"], eco_smart.get("source", "ECOALIM"), ) if eco_worst: val, nom, src = eco_worst result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / kg de produit" result.source_db = src result.intrant_utilise = nom if llm_match_exact_eco is not None: result.match_exact = llm_match_exact_eco else: result.match_exact = _is_name_match(matiere, nom) if llm_justification_eco: result.justification_alternative = llm_justification_eco result.actions_appliquees.append(f" → Trouvé dans ECOALIM : {nom} = {val:.4f} kg CO2 eq/kg") return result # Étape 3 : LLM result.actions_appliquees.append("3. Recherche via LLM de l'intrant transformé au process le plus proche (GFLI)") gfli_smart = llm_service.smart_search_gfli(matiere) if gfli_smart and "valeur_kg_co2_eq_par_tonne" in gfli_smart: val = gfli_smart["valeur_kg_co2_eq_par_tonne"] result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / tonne de produit" result.source_db = gfli_smart["source"] result.intrant_utilise = gfli_smart["nom_intrant"] result.match_exact = gfli_smart.get("match_exact", False) result.justification_alternative = gfli_smart.get("justification") result.actions_appliquees.append(f" → Via LLM : {gfli_smart['nom_intrant']}") return result result.erreur = f"Aucune valeur trouvée pour '{matiere}' (transformé, provenance inconnue)." return result def _resolve_node_8(matiere: str, result: CarbonResult) -> CarbonResult: """ Node 8 : Provenance connue + brut + cultivé en France. 1. EcoALIM 2. GFLI 3. Intrant à la pratique culturale la plus proche dans EcoALIM (LLM) """ result.actions_appliquees.append("1. Recherche dans ECOALIM pour la France") eco_result = llm_service.smart_search_ecoalim(matiere, pays_production="France") if eco_result: val = eco_result["valeur_kg_co2_eq"] result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / kg de produit" result.source_db = eco_result["source"] result.intrant_utilise = eco_result["nom_intrant"] result.match_exact = eco_result["match_exact"] result.justification_alternative = eco_result.get("justification") result.actions_appliquees.append(f" → Trouvé dans ECOALIM : {eco_result['nom_intrant']} = {val:.4f} kg CO2 eq/kg") return result result.actions_appliquees.append("2. Recherche dans GFLI pour FR") gfli_result = llm_service.smart_search_gfli(matiere, country_iso="FR") if gfli_result: val = gfli_result["valeur_kg_co2_eq_par_tonne"] result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / tonne de produit" result.source_db = gfli_result["source"] result.intrant_utilise = gfli_result["nom_intrant"] result.match_exact = gfli_result["match_exact"] result.justification_alternative = gfli_result.get("justification") result.actions_appliquees.append(f" → Trouvé dans GFLI : {gfli_result['nom_intrant']}") return result result.actions_appliquees.append("3. Recherche via LLM de la pratique culturale la plus proche dans ECOALIM") eco_smart = llm_service.smart_search_ecoalim(matiere) if eco_smart: val = eco_smart["valeur_kg_co2_eq"] result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / kg de produit" result.source_db = eco_smart["source"] result.intrant_utilise = eco_smart["nom_intrant"] result.match_exact = False result.justification_alternative = eco_smart.get("justification") result.actions_appliquees.append(f" → Via LLM : {eco_smart['nom_intrant']}") return result result.erreur = f"Aucune valeur trouvée pour '{matiere}' (brut, France)." return result def _resolve_node_9(matiere: str, pays_production: str, result: CarbonResult) -> CarbonResult: """ Node 9 : Provenance connue + brut + cultivé hors France. 1. GFLI du pays correspondant 2. RER (Europe) ou GLO (autre continent) 3. EcoALIM """ country_iso = _get_country_iso(pays_production) result.actions_appliquees.append(f"1. Recherche dans GFLI pour le pays {pays_production} (ISO: {country_iso})") gfli_result = llm_service.smart_search_gfli(matiere, country_iso=country_iso) if gfli_result: val = gfli_result["valeur_kg_co2_eq_par_tonne"] result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / tonne de produit" result.source_db = gfli_result["source"] result.intrant_utilise = gfli_result["nom_intrant"] result.match_exact = gfli_result["match_exact"] result.justification_alternative = gfli_result.get("justification") result.actions_appliquees.append(f" → Trouvé dans GFLI : {gfli_result['nom_intrant']}") return result # Étape 2 : RER ou GLO is_eu = _is_european(pays_production) if is_eu: result.actions_appliquees.append("2. Pays européen → Recherche Mix Européen (RER) dans GFLI") rer = llm_service.smart_search_gfli(matiere, country_iso="RER") if rer: val = rer["valeur_kg_co2_eq_par_tonne"] result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / tonne de produit" result.source_db = rer["source"] + " (Mix Européen RER)" result.intrant_utilise = rer["nom_intrant"] result.match_exact = rer["match_exact"] result.justification_alternative = rer.get("justification") result.actions_appliquees.append(f" → Trouvé RER : {rer['nom_intrant']}") return result else: result.actions_appliquees.append("2. Pays hors Europe → Recherche Mix Monde (GLO) dans GFLI") glo = llm_service.smart_search_gfli(matiere, country_iso="GLO") if glo: val = glo["valeur_kg_co2_eq_par_tonne"] result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / tonne de produit" result.source_db = glo["source"] + " (Mix Monde GLO)" result.intrant_utilise = glo["nom_intrant"] result.match_exact = glo["match_exact"] result.justification_alternative = glo.get("justification") result.actions_appliquees.append(f" → Trouvé GLO : {glo['nom_intrant']}") return result result.actions_appliquees.append("3. Recherche dans ECOALIM") eco_result = llm_service.smart_search_ecoalim(matiere, pays_production=pays_production) if eco_result: val = eco_result["valeur_kg_co2_eq"] result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / kg de produit" result.source_db = eco_result["source"] result.intrant_utilise = eco_result["nom_intrant"] result.match_exact = eco_result["match_exact"] result.justification_alternative = eco_result.get("justification") result.actions_appliquees.append(f" → Trouvé dans ECOALIM : {eco_result['nom_intrant']}") return result result.erreur = f"Aucune valeur trouvée pour '{matiere}' (brut, {pays_production})." return result def _resolve_node_10(matiere: str, result: CarbonResult) -> CarbonResult: """ Node 10 : Provenance connue + transformé + France/France. 1. EcoALIM pour l'intrant transformé 2. A/ impact process connu : brut EcoALIM + process / B/ sinon GFLI 3. Intrant au process le plus proche dans EcoALIM (LLM) """ result.actions_appliquees.append("1. Recherche dans ECOALIM (transformé France/France)") eco_result = llm_service.smart_search_ecoalim(matiere, pays_production="France", pays_transformation="France") if eco_result: val = eco_result["valeur_kg_co2_eq"] result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / kg de produit" result.source_db = eco_result["source"] result.intrant_utilise = eco_result["nom_intrant"] result.match_exact = eco_result["match_exact"] result.justification_alternative = eco_result.get("justification") result.actions_appliquees.append(f" → Trouvé dans ECOALIM : {eco_result['nom_intrant']} = {val:.4f}") return result # Étape 2 : GFLI France result.actions_appliquees.append("2. Impact process non connu → Recherche dans GFLI (FR)") gfli_result = llm_service.smart_search_gfli(matiere, country_iso="FR") if gfli_result: val = gfli_result["valeur_kg_co2_eq_par_tonne"] result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / tonne de produit" result.source_db = gfli_result["source"] result.intrant_utilise = gfli_result["nom_intrant"] result.match_exact = gfli_result["match_exact"] result.justification_alternative = gfli_result.get("justification") result.actions_appliquees.append(f" → Trouvé dans GFLI : {gfli_result['nom_intrant']}") return result # Étape 3 : LLM process le plus proche result.actions_appliquees.append("3. Recherche via LLM du process le plus proche dans ECOALIM") eco_smart = llm_service.smart_search_ecoalim(matiere) if eco_smart: val = eco_smart["valeur_kg_co2_eq"] result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / kg de produit" result.source_db = eco_smart["source"] result.intrant_utilise = eco_smart["nom_intrant"] result.match_exact = False result.justification_alternative = eco_smart.get("justification") result.actions_appliquees.append(f" → Via LLM : {eco_smart['nom_intrant']}") return result result.erreur = f"Aucune valeur trouvée pour '{matiere}' (transformé, France/France)." return result def _resolve_node_11(matiere: str, result: CarbonResult) -> CarbonResult: """ Node 11 : Transformé en France, MP brute non FR ou inconnue. 1. GFLI France 2. A/ process connu → brut GFLI + process / B/ sinon RER 3. EcoALIM 4. Pratique culturale la plus proche GFLI (LLM) """ result.actions_appliquees.append("1. Recherche dans GFLI (France)") gfli_result = llm_service.smart_search_gfli(matiere, country_iso="FR") if gfli_result: val = gfli_result["valeur_kg_co2_eq_par_tonne"] result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / tonne de produit" result.source_db = gfli_result["source"] result.intrant_utilise = gfli_result["nom_intrant"] result.match_exact = gfli_result["match_exact"] result.justification_alternative = gfli_result.get("justification") result.actions_appliquees.append(f" → Trouvé dans GFLI : {gfli_result['nom_intrant']}") return result # Étape 2 : RER result.actions_appliquees.append("2. Impact process non connu → Recherche Mix Européen (RER) dans GFLI") rer = llm_service.smart_search_gfli(matiere, country_iso="RER") if rer: val = rer["valeur_kg_co2_eq_par_tonne"] result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / tonne de produit" result.source_db = rer["source"] + " (Mix Européen RER)" result.intrant_utilise = rer["nom_intrant"] result.match_exact = rer["match_exact"] result.justification_alternative = rer.get("justification") result.actions_appliquees.append(f" → Trouvé RER : {rer['nom_intrant']}") return result # Étape 3 : EcoALIM result.actions_appliquees.append("3. Recherche dans ECOALIM") eco_result = llm_service.smart_search_ecoalim(matiere) if eco_result: val = eco_result["valeur_kg_co2_eq"] result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / kg de produit" result.source_db = eco_result["source"] result.intrant_utilise = eco_result["nom_intrant"] result.match_exact = eco_result["match_exact"] result.justification_alternative = eco_result.get("justification") result.actions_appliquees.append(f" → Trouvé dans ECOALIM : {eco_result['nom_intrant']}") return result # Étape 4 : LLM result.actions_appliquees.append("4. Recherche via LLM de la pratique culturale la plus proche (GFLI)") gfli_smart = llm_service.smart_search_gfli(matiere) if gfli_smart: val = gfli_smart["valeur_kg_co2_eq_par_tonne"] result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / tonne de produit" result.source_db = gfli_smart["source"] result.intrant_utilise = gfli_smart["nom_intrant"] result.match_exact = False result.justification_alternative = gfli_smart.get("justification") result.actions_appliquees.append(f" → Via LLM : {gfli_smart['nom_intrant']}") return result result.erreur = f"Aucune valeur trouvée pour '{matiere}' (transformé France, MP brute hors FR)." return result def _resolve_node_12(matiere: str, pays_transformation: str, result: CarbonResult) -> CarbonResult: """ Node 12 : Transformé hors France. 1. GFLI du pays correspondant 2. A/ process connu / B/ sinon RER (Europe) ou GLO (autre) 3. EcoALIM 4. Pratique culturale la plus proche GFLI (LLM) """ country_iso = _get_country_iso(pays_transformation) result.actions_appliquees.append(f"1. Recherche dans GFLI pour {pays_transformation} (ISO: {country_iso})") gfli_result = llm_service.smart_search_gfli(matiere, country_iso=country_iso) if gfli_result: val = gfli_result["valeur_kg_co2_eq_par_tonne"] result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / tonne de produit" result.source_db = gfli_result["source"] result.intrant_utilise = gfli_result["nom_intrant"] result.match_exact = gfli_result["match_exact"] result.justification_alternative = gfli_result.get("justification") result.actions_appliquees.append(f" → Trouvé dans GFLI : {gfli_result['nom_intrant']}") return result # Étape 2 : RER ou GLO is_eu = _is_european(pays_transformation) if is_eu: result.actions_appliquees.append("2. Pays européen → Recherche Mix Européen (RER)") fallback = llm_service.smart_search_gfli(matiere, country_iso="RER") else: result.actions_appliquees.append("2. Pays hors Europe → Recherche Mix Monde (GLO)") fallback = llm_service.smart_search_gfli(matiere, country_iso="GLO") if fallback: val = fallback["valeur_kg_co2_eq_par_tonne"] result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 mix_type = "RER" if is_eu else "GLO" result.unite_source = "kg CO2 eq / tonne de produit" result.source_db = fallback["source"] + f" (Mix {mix_type})" result.intrant_utilise = fallback["nom_intrant"] result.match_exact = fallback["match_exact"] result.justification_alternative = fallback.get("justification") result.actions_appliquees.append(f" → Trouvé {mix_type} : {fallback['nom_intrant']}") return result # Étape 3 : EcoALIM result.actions_appliquees.append("3. Recherche dans ECOALIM") eco_result = llm_service.smart_search_ecoalim(matiere) if eco_result: val = eco_result["valeur_kg_co2_eq"] result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / kg de produit" result.source_db = eco_result["source"] result.intrant_utilise = eco_result["nom_intrant"] result.match_exact = eco_result["match_exact"] result.justification_alternative = eco_result.get("justification") result.actions_appliquees.append(f" → Trouvé dans ECOALIM : {eco_result['nom_intrant']}") return result # Étape 4 : LLM result.actions_appliquees.append("4. Recherche via LLM de la pratique culturale la plus proche (GFLI)") gfli_smart = llm_service.smart_search_gfli(matiere) if gfli_smart: val = gfli_smart["valeur_kg_co2_eq_par_tonne"] result.impact_kg_co2_eq = val result.impact_tonne_co2_eq = val / 1000.0 result.unite_source = "kg CO2 eq / tonne de produit" result.source_db = gfli_smart["source"] result.intrant_utilise = gfli_smart["nom_intrant"] result.match_exact = False result.justification_alternative = gfli_smart.get("justification") result.actions_appliquees.append(f" → Via LLM : {gfli_smart['nom_intrant']}") return result result.erreur = f"Aucune valeur trouvée pour '{matiere}' (transformé hors France)." return result # ============================================================================ # Moteur principal # ============================================================================ def evaluate_carbon_impact( matiere_premiere: str, pays_production: Optional[str] = None, pays_transformation: Optional[str] = None, ) -> CarbonResult: """ Point d'entrée principal : évalue l'impact carbone d'une matière première en suivant le logigramme. Args: matiere_premiere: Nom de la matière première (ex: "BLE", "T.TNSL DEC.", "SOJA") pays_production: Pays de production de la MP brute (ex: "France", "Brésil") ou None si inconnu pays_transformation: Pays de transformation (ex: "France") ou None si pas de transformation Returns: CarbonResult avec toutes les informations """ result = CarbonResult( matiere_premiere=matiere_premiere, pays_production=pays_production, pays_transformation=pays_transformation, classification="", classification_justification="", ) # ----------------------------------------------------------------------- # Étape 1 : Classifier brut vs transformé via LLM + PDF CIR # ----------------------------------------------------------------------- classification = llm_service.determine_brut_ou_transforme(matiere_premiere) result.classification = classification.get("classification", "brut") result.classification_justification = classification.get("justification", "") is_transformed = result.classification == "transforme" result.parcours.append(StepLog( node_id="classification", question="La matière est-elle brute ou transformée ?", answer=f"{'Transformée' if is_transformed else 'Brute'} — {result.classification_justification}", )) # ----------------------------------------------------------------------- # Étape 2 : Node 1 - Connaît-on la provenance ? # ----------------------------------------------------------------------- provenance_connue = bool(pays_production) if not provenance_connue: result.parcours.append(StepLog( node_id="node_1", question="Connaissez-vous l'endroit où l'intrant a été cultivé ou produit ?", answer="Non — provenance inconnue", )) # Node 2 : brut ou transformé ? if not is_transformed: result.parcours.append(StepLog( node_id="node_2", question="Quel est le niveau de transformation ?", answer="Intrant brut/non transformé", )) result.node_resultat = "node_4" return _resolve_node_4(matiere_premiere, result) else: result.parcours.append(StepLog( node_id="node_2", question="Quel est le niveau de transformation ?", answer="Coproduit/intrant transformé", )) result.node_resultat = "node_5" return _resolve_node_5(matiere_premiere, result) # Provenance connue result.parcours.append(StepLog( node_id="node_1", question="Connaissez-vous l'endroit où l'intrant a été cultivé ou produit ?", answer=f"Oui — Production: {pays_production}" + (f", Transformation: {pays_transformation}" if pays_transformation else ""), )) if not is_transformed: # Node 3 → Node 6 : où a-t-il été cultivé ? result.parcours.append(StepLog( node_id="node_3", question="Quel est le niveau de transformation ?", answer="Intrant brut/non transformé", )) if _is_france(pays_production): result.parcours.append(StepLog( node_id="node_6", question="Où l'intrant brut a-t-il été cultivé ?", answer="En France", )) result.node_resultat = "node_8" return _resolve_node_8(matiere_premiere, result) else: result.parcours.append(StepLog( node_id="node_6", question="Où l'intrant brut a-t-il été cultivé ?", answer=f"Hors France — {pays_production}", )) result.node_resultat = "node_9" return _resolve_node_9(matiere_premiere, pays_production, result) else: # Node 3 → Node 7 : où transformé + origine MP brute ? result.parcours.append(StepLog( node_id="node_3", question="Quel est le niveau de transformation ?", answer="Coproduit/intrant transformé", )) if _is_france(pays_transformation) and _is_france(pays_production): result.parcours.append(StepLog( node_id="node_7", question="Où l'intrant a-t-il été transformé et d'où provient la MP brute ?", answer="Transformé en France à partir de MP brute française", )) result.node_resultat = "node_10" return _resolve_node_10(matiere_premiere, result) elif _is_france(pays_transformation): result.parcours.append(StepLog( node_id="node_7", question="Où l'intrant a-t-il été transformé et d'où provient la MP brute ?", answer=f"Transformé en France, MP brute de {pays_production or 'origine inconnue'}", )) result.node_resultat = "node_11" return _resolve_node_11(matiere_premiere, result) else: result.parcours.append(StepLog( node_id="node_7", question="Où l'intrant a-t-il été transformé et d'où provient la MP brute ?", answer=f"Transformé hors France — {pays_transformation}", )) result.node_resultat = "node_12" result = _resolve_node_12(matiere_premiere, pays_transformation or pays_production or "", result) # ------------------------------------------------------------------ # Post-processing : collecter les candidats alternatifs # ------------------------------------------------------------------ result = _collect_candidates(result) # Générer une justification LLM si le match n'est pas exact et qu'il n'y en a pas if not result.match_exact and not result.justification_alternative and not result.erreur: if result.intrant_utilise and result.impact_kg_co2_eq is not None: try: result.justification_alternative = llm_service.justify_alternative_value( result.matiere_premiere, result.intrant_utilise, result.impact_kg_co2_eq, result.source_db, ) except Exception: result.justification_alternative = ( f"Valeur de '{result.intrant_utilise}' utilisée comme proxy pour " f"'{result.matiere_premiere}' (matière la plus proche dans {result.source_db})." ) return result def _collect_candidates(result: CarbonResult) -> CarbonResult: """ Après résolution, cherche les autres produits correspondants dans la même base de données pour proposer des alternatives triées par pertinence. """ if result.erreur or result.intrant_utilise is None: return result matiere = result.matiere_premiere source = result.source_db or "" candidates: list[dict] = [] # Déterminer le pays ISO pour GFLI country_iso = None if result.pays_production: country_iso = _get_country_iso(result.pays_production) if result.pays_transformation: country_iso = _get_country_iso(result.pays_transformation) or country_iso # Collecter depuis la source utilisée + l'autre source # D'abord la source principalement utilisée if "ECOALIM" in source.upper(): candidates.extend(data_loader.get_top_ecoalim_candidates( matiere, pays_production=result.pays_production, pays_transformation=result.pays_transformation, top_n=8, )) candidates.extend(data_loader.get_top_gfli_candidates( matiere, country_iso=country_iso, top_n=4, )) else: # Essayer aussi avec le nom traduit si on est sur GFLI # Le nom d'intrant utilisé contient le terme anglais intrant_base = result.intrant_utilise.split(",")[0].split("/")[0].strip() candidates.extend(data_loader.get_top_gfli_candidates( intrant_base, country_iso=country_iso, top_n=8, )) candidates.extend(data_loader.get_top_ecoalim_candidates( matiere, pays_production=result.pays_production, pays_transformation=result.pays_transformation, top_n=4, )) # Dédupliquer, exclure l'intrant sélectionné, et filtrer les faux positifs seen = set() unique_candidates = [] intrant_base = "" if result.intrant_utilise: intrant_base = result.intrant_utilise.split(",")[0].split("/")[0].strip().lower() for c in candidates: key = (c["nom"], c["source"]) if key in seen or c["nom"] == result.intrant_utilise: continue # Pour GFLI, vérifier que le candidat est pertinent if c["source"] == "GFLI" and not _is_name_match(matiere, c["nom"]): # Accepter quand même si ça matche le nom de base de l'intrant validé if intrant_base and _is_name_match(intrant_base, c["nom"]): pass # OK, même famille de produit else: continue # Faux positif seen.add(key) unique_candidates.append(c) result.candidats_alternatifs = unique_candidates return result