Spaces:
Runtime error
Runtime error
| """ | |
| 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 | |
| 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 | |
| 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 | |