Spaces:
Sleeping
Sleeping
| """ | |
| 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 | |
| import logging | |
| from dataclasses import dataclass, field | |
| from typing import List, Optional, Tuple | |
| import config | |
| import data_loader | |
| import llm_service | |
| # Configure logging for flowchart engine | |
| logger = logging.getLogger(__name__) | |
| logger.setLevel(logging.DEBUG) | |
| 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) | |
| candidat_recommande: Optional[str] = None | |
| candidats_reflexion: Optional[str] = None | |
| # 4 propositions d'alternatives (itinerary, locality, form, combined) | |
| alternatives_itinerary: Optional[dict] = None | |
| alternatives_locality: Optional[dict] = None | |
| alternatives_form: Optional[dict] = None | |
| alternatives_combined: Optional[dict] = None | |
| erreur: Optional[str] = None | |
| def _normalize_country_name(pays: Optional[str]) -> str: | |
| """Normalise un nom de pays : minuscule, strip, espaces → tirets. | |
| Ex: 'Pays Bas' → 'pays-bas', 'Etats Unis' → 'etats-unis' | |
| """ | |
| if not pays: | |
| return "" | |
| return pays.lower().strip().replace(" ", "-") | |
| def _is_france(pays: Optional[str]) -> bool: | |
| """Vérifie si le pays est la France.""" | |
| if not pays: | |
| return False | |
| n = _normalize_country_name(pays) | |
| if n in ("france", "fr"): | |
| return True | |
| # Fallback: check if input is ISO code "FR" | |
| return pays.strip().upper() == "FR" | |
| def _is_european(pays: Optional[str]) -> bool: | |
| """Vérifie si le pays est européen.""" | |
| if not pays: | |
| return False | |
| n = _normalize_country_name(pays) | |
| if n in config.EUROPEAN_COUNTRIES_FR: | |
| return True | |
| # Try to get ISO from French mapping, or use uppercase input as fallback | |
| pays_iso = config.PAYS_FR_TO_ISO.get(n, pays.strip().upper()) | |
| is_eu = pays_iso in config.EUROPEAN_COUNTRIES_ISO | |
| logger.debug(f"_is_european({pays}) → {is_eu} (ISO: {pays_iso})") | |
| return is_eu | |
| def _get_country_iso(pays: Optional[str]) -> Optional[str]: | |
| """Convertit un nom de pays FR en code ISO.""" | |
| if not pays: | |
| return None | |
| n = _normalize_country_name(pays) | |
| # Try to get ISO from French mapping, or use uppercase input as fallback if it's already an ISO code | |
| iso = config.PAYS_FR_TO_ISO.get(n) | |
| if iso: | |
| return iso | |
| # Check if input is already a valid ISO code | |
| pays_upper = pays.strip().upper() | |
| if pays_upper in config.EUROPEAN_COUNTRIES_ISO: | |
| return pays_upper | |
| # Check if it's a valid ISO code in our mapping values | |
| if pays_upper in config.PAYS_FR_TO_ISO.values(): | |
| return pays_upper | |
| return None | |
| 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 | |
| # Note: impact_tonne_co2_eq will be set by post-processing based on unite_source | |
| 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 | |
| # Étape 4 : Fallback - Proposer des matières alternatives | |
| result.actions_appliquees.append("4. Fallback - Recherche via LLM de 4 alternatives") | |
| alternatives = llm_service.find_alternative_materials(matiere, db_name="GFLI") | |
| if alternatives: | |
| # Stocker les 4 alternatives dans CarbonResult | |
| if alternatives.get("itinerary"): | |
| alt = alternatives["itinerary"] | |
| result.alternatives_itinerary = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| if alternatives.get("locality"): | |
| alt = alternatives["locality"] | |
| result.alternatives_locality = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| if alternatives.get("form"): | |
| alt = alternatives["form"] | |
| result.alternatives_form = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| if alternatives.get("combined"): | |
| alt = alternatives["combined"] | |
| result.alternatives_combined = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| # Utiliser la combined comme valeur principale | |
| val = alt["impact"] | |
| 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 = alt["source"] | |
| result.intrant_utilise = alt["name"] | |
| result.match_exact = False | |
| result.justification_alternative = alt["reasoning"] | |
| result.actions_appliquees.append(f" → Matière proposée (combo) : {alt['name']} = {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_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 | |
| # Note: impact_tonne_co2_eq will be set by post-processing based on unite_source | |
| 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 | |
| # Note: impact_tonne_co2_eq will be set by post-processing based on unite_source | |
| 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 | |
| # Étape 4 : Fallback - Proposer des matières alternatives | |
| result.actions_appliquees.append("4. Fallback - Recherche via LLM de 4 alternatives (France)") | |
| alternatives = llm_service.find_alternative_materials(matiere, db_name="GFLI", country_hint="France") | |
| if alternatives: | |
| # Stocker les 4 alternatives dans CarbonResult | |
| if alternatives.get("itinerary"): | |
| alt = alternatives["itinerary"] | |
| result.alternatives_itinerary = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| if alternatives.get("locality"): | |
| alt = alternatives["locality"] | |
| result.alternatives_locality = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| if alternatives.get("form"): | |
| alt = alternatives["form"] | |
| result.alternatives_form = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| if alternatives.get("combined"): | |
| alt = alternatives["combined"] | |
| result.alternatives_combined = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| # Utiliser la combined comme valeur principale | |
| val = alt["impact"] | |
| 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 = alt["source"] | |
| result.intrant_utilise = alt["name"] | |
| result.match_exact = False | |
| result.justification_alternative = alt["reasoning"] | |
| result.actions_appliquees.append(f" → Matière proposée (combo) : {alt['name']} = {val:.2f} kg CO2 eq/t") | |
| 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) | |
| logger.info(f"Node 9: Étape 1 - Recherche '{matiere}' dans GFLI pour {pays_production} (ISO: {country_iso})") | |
| 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']}") | |
| logger.info(f"✓ Trouvé pays spécifique: {gfli_result['nom_intrant']} = {val:.2f} kg CO2/t") | |
| return result | |
| else: | |
| logger.warning(f"✗ Pays spécifique ({country_iso}) non trouvé pour '{matiere}'") | |
| # Étape 2 : RER ou GLO | |
| is_eu = _is_european(pays_production) | |
| if is_eu: | |
| logger.info(f"Node 9: Pays européen ({pays_production}) → Recherche RER") | |
| 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']}") | |
| logger.info(f"✓ Trouvé RER: {rer['nom_intrant']} = {val:.2f} kg CO2/t") | |
| return result | |
| else: | |
| logger.warning(f"✗ RER non trouvé pour '{matiere}' → Fallback vers ECOALIM") | |
| result.actions_appliquees.append(f" → RER non trouvé pour '{matiere}'") | |
| else: | |
| logger.info(f"Node 9: Pays NON européen ({pays_production}) → Recherche GLO") | |
| 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") | |
| logger.info(f"✓ Trouvé GLO: {glo['nom_intrant']} = {val:.2f} kg CO2/t") | |
| result.actions_appliquees.append(f" → Trouvé GLO : {glo['nom_intrant']}") | |
| return result | |
| else: | |
| logger.warning(f"✗ GLO non trouvé pour '{matiere}' → Fallback vers ECOALIM") | |
| result.actions_appliquees.append(f" → GLO non trouvé pour '{matiere}'") | |
| logger.info(f"Node 9: Étape 3 - Recherche ECOALIM comme fallback") | |
| 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 | |
| # Note: impact_tonne_co2_eq will be set by post-processing based on unite_source | |
| 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']}") | |
| logger.info(f"✓ Trouvé ECOALIM: {eco_result['nom_intrant']} = {val:.2f} kg CO2/kg") | |
| return result | |
| else: | |
| logger.warning(f"✗ ECOALIM non trouvé pour '{matiere}' → Fallback vers alternatives LLM") | |
| result.actions_appliquees.append(f" → ECOALIM non trouvé pour '{matiere}'") | |
| # Étape 4 : Fallback - Proposer des matières alternatives | |
| logger.info(f"Node 9: Étape 4 - Recherche d'alternatives via LLM pour '{matiere}'") | |
| result.actions_appliquees.append(f"4. Fallback - Recherche via LLM de 4 alternatives ({pays_production})") | |
| alternatives = llm_service.find_alternative_materials(matiere, db_name="GFLI", country_hint=pays_production) | |
| if alternatives: | |
| # Stocker les 4 alternatives dans CarbonResult | |
| if alternatives.get("itinerary"): | |
| alt = alternatives["itinerary"] | |
| result.alternatives_itinerary = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| if alternatives.get("locality"): | |
| alt = alternatives["locality"] | |
| result.alternatives_locality = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| if alternatives.get("form"): | |
| alt = alternatives["form"] | |
| result.alternatives_form = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| if alternatives.get("combined"): | |
| alt = alternatives["combined"] | |
| result.alternatives_combined = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| # Utiliser la combined comme valeur principale | |
| val = alt["impact"] | |
| 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 = alt["source"] | |
| result.intrant_utilise = alt["name"] | |
| result.match_exact = False | |
| result.justification_alternative = alt["reasoning"] | |
| result.actions_appliquees.append(f" → Matière proposée (combo) : {alt['name']} = {val:.2f} kg CO2 eq/t") | |
| 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 | |
| # Note: impact_tonne_co2_eq will be set by post-processing based on unite_source | |
| 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']}") | |
| # Update node to reflect that GFLI was used instead of ECOALIM | |
| result.node_resultat = "node_base_gfli_fr" | |
| 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 | |
| # Note: impact_tonne_co2_eq will be set by post-processing based on unite_source | |
| 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 | |
| # Étape 4 : Fallback - Proposer des matières alternatives | |
| result.actions_appliquees.append("4. Fallback - Recherche via LLM de 4 alternatives (France)") | |
| alternatives = llm_service.find_alternative_materials(matiere, db_name="GFLI", country_hint="France") | |
| if alternatives: | |
| # Stocker les 4 alternatives dans CarbonResult | |
| if alternatives.get("itinerary"): | |
| alt = alternatives["itinerary"] | |
| result.alternatives_itinerary = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| if alternatives.get("locality"): | |
| alt = alternatives["locality"] | |
| result.alternatives_locality = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| if alternatives.get("form"): | |
| alt = alternatives["form"] | |
| result.alternatives_form = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| if alternatives.get("combined"): | |
| alt = alternatives["combined"] | |
| result.alternatives_combined = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| # Utiliser la combined comme valeur principale | |
| val = alt["impact"] | |
| result.impact_kg_co2_eq = val | |
| # Note: Alternatives from GFLI are in kg CO2 eq / tonne | |
| # impact_tonne_co2_eq will be set by post-processing | |
| result.unite_source = "kg CO2 eq / tonne de produit" | |
| result.source_db = alt["source"] | |
| result.intrant_utilise = alt["name"] | |
| result.match_exact = False | |
| result.justification_alternative = alt["reasoning"] | |
| result.actions_appliquees.append(f" → Matière proposée (combo) : {alt['name']} = {val:.2f} kg CO2 eq/t") | |
| # Update node to reflect that GFLI was used (via LLM alternatives) | |
| result.node_resultat = "node_base_gfli_alternatives" | |
| 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 | |
| # Note: impact_tonne_co2_eq will be set by post-processing based on unite_source | |
| 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 | |
| # Étape 5 : Fallback - Proposer des matières alternatives | |
| result.actions_appliquees.append("5. Fallback - Recherche via LLM de 4 alternatives (France)") | |
| alternatives = llm_service.find_alternative_materials(matiere, db_name="GFLI", country_hint="France") | |
| if alternatives: | |
| # Stocker les 4 alternatives dans CarbonResult | |
| if alternatives.get("itinerary"): | |
| alt = alternatives["itinerary"] | |
| result.alternatives_itinerary = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| if alternatives.get("locality"): | |
| alt = alternatives["locality"] | |
| result.alternatives_locality = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| if alternatives.get("form"): | |
| alt = alternatives["form"] | |
| result.alternatives_form = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| if alternatives.get("combined"): | |
| alt = alternatives["combined"] | |
| result.alternatives_combined = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| # Utiliser la combined comme valeur principale | |
| val = alt["impact"] | |
| 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 = alt["source"] | |
| result.intrant_utilise = alt["name"] | |
| result.match_exact = False | |
| result.justification_alternative = alt["reasoning"] | |
| result.actions_appliquees.append(f" → Matière proposée (combo) : {alt['name']} = {val:.2f} kg CO2 eq/t") | |
| 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 | |
| # Note: impact_tonne_co2_eq will be set by post-processing based on unite_source | |
| 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 | |
| # Étape 5 : Fallback - Proposer des matières alternatives | |
| result.actions_appliquees.append(f"5. Fallback - Recherche via LLM de 4 alternatives ({pays_transformation})") | |
| alternatives = llm_service.find_alternative_materials(matiere, db_name="GFLI", country_hint=pays_transformation) | |
| if alternatives: | |
| # Stocker les 4 alternatives dans CarbonResult | |
| if alternatives.get("itinerary"): | |
| alt = alternatives["itinerary"] | |
| result.alternatives_itinerary = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| if alternatives.get("locality"): | |
| alt = alternatives["locality"] | |
| result.alternatives_locality = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| if alternatives.get("form"): | |
| alt = alternatives["form"] | |
| result.alternatives_form = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| if alternatives.get("combined"): | |
| alt = alternatives["combined"] | |
| result.alternatives_combined = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| # Utiliser la combined comme valeur principale | |
| val = alt["impact"] | |
| 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 = alt["source"] | |
| result.intrant_utilise = alt["name"] | |
| result.match_exact = False | |
| result.justification_alternative = alt["reasoning"] | |
| result.actions_appliquees.append(f" → Matière proposée (combo) : {alt['name']} = {val:.2f} kg CO2 eq/t") | |
| return result | |
| result.erreur = f"Aucune valeur trouvée pour '{matiere}' (transformé hors France)." | |
| return result | |
| # ============================================================================ | |
| # Helpers communs | |
| # ============================================================================ | |
| def _store_alternatives(alternatives: dict, result: CarbonResult) -> CarbonResult: | |
| """Stocke les 4 alternatives dans CarbonResult et utilise 'combined' comme valeur principale.""" | |
| for key in ("itinerary", "locality", "form", "combined"): | |
| alt = alternatives.get(key) | |
| if alt: | |
| entry = { | |
| "name": alt["name"], | |
| "impact": alt["impact"], | |
| "source": alt["source"], | |
| "reasoning": alt["reasoning"], | |
| } | |
| setattr(result, f"alternatives_{key}", entry) | |
| combined = alternatives.get("combined") | |
| if combined: | |
| val = combined["impact"] | |
| 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 = combined["source"] | |
| result.intrant_utilise = combined["name"] | |
| result.match_exact = False | |
| result.justification_alternative = combined["reasoning"] | |
| result.actions_appliquees.append( | |
| f" → Matière proposée (combo) : {combined['name']} = {val:.2f} kg CO2 eq/t" | |
| ) | |
| return result | |
| def _apply_gfli_result(gfli_smart: dict, result: CarbonResult) -> CarbonResult: | |
| """Applique un résultat smart_search_gfli au CarbonResult.""" | |
| 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.get("source", "GFLI") | |
| result.intrant_utilise = gfli_smart["nom_intrant"] | |
| result.match_exact = gfli_smart.get("match_exact", False) | |
| result.justification_alternative = gfli_smart.get("justification") | |
| return result | |
| def _apply_ecoalim_result(eco_smart: dict, result: CarbonResult) -> CarbonResult: | |
| """Applique un résultat smart_search_ecoalim au CarbonResult.""" | |
| val = eco_smart["valeur_kg_co2_eq"] | |
| result.impact_kg_co2_eq = val | |
| result.impact_tonne_co2_eq = val | |
| result.unite_source = "kg CO2 eq / kg de produit" | |
| result.source_db = eco_smart.get("source", "ECOALIM") | |
| result.intrant_utilise = eco_smart["nom_intrant"] | |
| result.match_exact = eco_smart.get("match_exact", False) | |
| result.justification_alternative = eco_smart.get("justification") | |
| return result | |
| # ============================================================================ | |
| # Résolveurs SOJA (logigramme_soja.json, nodes 4-9) | |
| # ============================================================================ | |
| def _resolve_soja_node_4(matiere: str, pays_production: str, result: CarbonResult) -> CarbonResult: | |
| """ | |
| Soja Node 4 : Pays d'origine connu + graines crues (non transformé). | |
| 1. ECOALIM "graines" du pays correspondant | |
| 2. GFLI "graines" du pays correspondant | |
| 3. GFLI RER (Europe) ou GLO (autre continent) pour graines crues | |
| """ | |
| # Termes de recherche spécifiques au soja graines | |
| search_terms = ["soja", "graine de soja", "soybean", "soybeans"] | |
| # Étape 1 : ECOALIM avec pays | |
| result.actions_appliquees.append(f"1. Recherche 'graines de soja' dans ECOALIM pour {pays_production}") | |
| for term in search_terms: | |
| eco = llm_service.smart_search_ecoalim(term, pays_production=pays_production) | |
| if eco: | |
| result = _apply_ecoalim_result(eco, result) | |
| result.actions_appliquees.append(f" → Trouvé dans ECOALIM : {eco['nom_intrant']} = {eco['valeur_kg_co2_eq']:.4f} kg CO2 eq/kg") | |
| return result | |
| # Étape 2 : GFLI avec pays | |
| result.actions_appliquees.append(f"2. Recherche 'soybeans' dans GFLI pour {pays_production}") | |
| country_iso = _get_country_iso(pays_production) | |
| for term in ["soybeans", "soybean", "soja"]: | |
| gfli = llm_service.smart_search_gfli(term, country_iso=country_iso) | |
| if gfli: | |
| result = _apply_gfli_result(gfli, result) | |
| result.actions_appliquees.append(f" → Trouvé dans GFLI : {gfli['nom_intrant']} = {gfli['valeur_kg_co2_eq_par_tonne']:.2f} kg CO2 eq/t") | |
| return result | |
| # Étape 3 : GFLI RER ou GLO | |
| is_eu = _is_european(pays_production) | |
| mix_label = "RER" if is_eu else "GLO" | |
| result.actions_appliquees.append(f"3. Recherche 'soybeans' dans GFLI mix {mix_label}") | |
| for term in ["soybeans", "soybean"]: | |
| gfli = llm_service.smart_search_gfli(term, country_iso=mix_label) | |
| if gfli: | |
| result = _apply_gfli_result(gfli, result) | |
| result.actions_appliquees.append(f" → Trouvé dans GFLI ({mix_label}) : {gfli['nom_intrant']}") | |
| return result | |
| # Fallback | |
| result.actions_appliquees.append("4. Fallback - Recherche via LLM de 4 alternatives (soja graines)") | |
| alts = llm_service.find_alternative_materials(matiere, db_name="GFLI", country_hint=pays_production) | |
| if alts: | |
| result = _store_alternatives(alts, result) | |
| return result | |
| result.erreur = f"Aucune valeur trouvée pour '{matiere}' (soja graines, {pays_production})." | |
| return result | |
| def _resolve_soja_node_5(matiere: str, pays_production: str, pays_transformation: Optional[str], | |
| result: CarbonResult) -> CarbonResult: | |
| """ | |
| Soja Node 5 : Pays d'origine connu + produit dérivé du soja. | |
| 1. ECOALIM couple intrant/origine graine (lieu transfo si dispo) | |
| 2. ECOALIM graines + process OU GFLI dérivé pays transfo OU plus défavorable | |
| 3. GFLI RER/GLO | |
| 4. Dernier recours : process le plus proche | |
| """ | |
| # Étape 1 : ECOALIM couple intrant/origine | |
| result.actions_appliquees.append(f"1. Recherche dérivé soja dans ECOALIM (origine graine : {pays_production})") | |
| eco = llm_service.smart_search_ecoalim( | |
| matiere, pays_production=pays_production, pays_transformation=pays_transformation | |
| ) | |
| if eco: | |
| result = _apply_ecoalim_result(eco, result) | |
| result.actions_appliquees.append(f" → Trouvé dans ECOALIM : {eco['nom_intrant']}") | |
| return result | |
| # Étape 2 : GFLI dérivé avec pays transfo ou origine | |
| pays_ref = pays_transformation or pays_production | |
| country_iso = _get_country_iso(pays_ref) | |
| result.actions_appliquees.append(f"2. Recherche dérivé soja dans GFLI ({pays_ref})") | |
| gfli = llm_service.smart_search_gfli(matiere, country_iso=country_iso) | |
| if gfli: | |
| result = _apply_gfli_result(gfli, result) | |
| result.actions_appliquees.append(f" → Trouvé dans GFLI : {gfli['nom_intrant']}") | |
| return result | |
| # Étape 3 : GFLI RER/GLO | |
| is_eu = _is_european(pays_ref) | |
| mix_label = "RER" if is_eu else "GLO" | |
| result.actions_appliquees.append(f"3. Recherche dérivé soja dans GFLI mix {mix_label}") | |
| gfli = llm_service.smart_search_gfli(matiere, country_iso=mix_label) | |
| if gfli: | |
| result = _apply_gfli_result(gfli, result) | |
| result.actions_appliquees.append(f" → Trouvé dans GFLI ({mix_label}) : {gfli['nom_intrant']}") | |
| return result | |
| # Étape 4 : Process le plus proche | |
| result.actions_appliquees.append("4. Dernier recours — process le plus proche dans ECOALIM/GFLI") | |
| alts = llm_service.find_alternative_materials(matiere, db_name="GFLI", country_hint=pays_production) | |
| if alts: | |
| result = _store_alternatives(alts, result) | |
| return result | |
| result.erreur = f"Aucune valeur trouvée pour '{matiere}' (dérivé soja, origine {pays_production})." | |
| return result | |
| def _resolve_soja_node_6(matiere: str, result: CarbonResult) -> CarbonResult: | |
| """ | |
| Soja Node 6 : Pays d'origine inconnu + graines crues. | |
| 1. Valeur "graines" la plus défavorable entre GFLI et ECOALIM. | |
| """ | |
| result.actions_appliquees.append("1. Recherche valeur 'graines soja' la plus défavorable (GFLI + ECOALIM)") | |
| best_val = None | |
| best_nom = None | |
| best_src = None | |
| best_unite = None | |
| # GFLI worst pour soybeans | |
| for term in ["soybeans", "soybean"]: | |
| gfli_worst = data_loader.get_gfli_worst_value(term) | |
| if gfli_worst: | |
| val, nom, src = gfli_worst | |
| if best_val is None or val > best_val: | |
| best_val = val | |
| best_nom = nom | |
| best_src = src | |
| best_unite = "kg CO2 eq / tonne de produit" | |
| break | |
| # ECOALIM worst pour soja | |
| for term in ["soja", "graine de soja"]: | |
| eco_worst = data_loader.get_ecoalim_worst_value(term) | |
| if eco_worst: | |
| val_eco, nom_eco, src_eco = eco_worst | |
| # Convertir EcoALIM (kg/kg) en kg/t pour comparaison | |
| val_eco_t = val_eco * 1000.0 | |
| if best_val is None or val_eco_t > best_val: | |
| best_val = val_eco | |
| best_nom = nom_eco | |
| best_src = src_eco | |
| best_unite = "kg CO2 eq / kg de produit" | |
| break | |
| if best_val is not None: | |
| result.impact_kg_co2_eq = best_val | |
| if "tonne" in best_unite: | |
| result.impact_tonne_co2_eq = best_val / 1000.0 | |
| else: | |
| result.impact_tonne_co2_eq = best_val | |
| result.unite_source = best_unite | |
| result.source_db = best_src | |
| result.intrant_utilise = best_nom | |
| result.match_exact = _is_name_match(matiere, best_nom) | |
| result.actions_appliquees.append(f" → Valeur la plus défavorable : {best_nom} dans {best_src}") | |
| return result | |
| # Fallback LLM | |
| result.actions_appliquees.append("2. Fallback - Recherche via LLM de 4 alternatives (soja graines, provenance inconnue)") | |
| alts = llm_service.find_alternative_materials(matiere, db_name="GFLI") | |
| if alts: | |
| result = _store_alternatives(alts, result) | |
| return result | |
| result.erreur = f"Aucune valeur trouvée pour '{matiere}' (soja graines, provenance inconnue)." | |
| return result | |
| def _resolve_soja_node_8(matiere: str, pays_transformation: str, result: CarbonResult) -> CarbonResult: | |
| """ | |
| Soja Node 8 : Pays origine inconnu + dérivé + lieu transfo connu. | |
| 1. ECOALIM par pays transfo (plus défavorable si plusieurs) | |
| 2. GFLI par pays transfo | |
| 3. GFLI RER/GLO | |
| 4. Process le plus proche | |
| """ | |
| # Étape 1 : ECOALIM pays transfo | |
| result.actions_appliquees.append(f"1. Recherche dérivé soja dans ECOALIM pour pays transfo = {pays_transformation}") | |
| eco = llm_service.smart_search_ecoalim(matiere, pays_transformation=pays_transformation) | |
| if eco: | |
| result = _apply_ecoalim_result(eco, result) | |
| result.actions_appliquees.append(f" → Trouvé dans ECOALIM : {eco['nom_intrant']}") | |
| return result | |
| # Étape 2 : GFLI pays transfo | |
| country_iso = _get_country_iso(pays_transformation) | |
| result.actions_appliquees.append(f"2. Recherche dérivé soja dans GFLI ({pays_transformation})") | |
| gfli = llm_service.smart_search_gfli(matiere, country_iso=country_iso) | |
| if gfli: | |
| result = _apply_gfli_result(gfli, result) | |
| result.actions_appliquees.append(f" → Trouvé dans GFLI : {gfli['nom_intrant']}") | |
| return result | |
| # Étape 3 : GFLI RER/GLO | |
| is_eu = _is_european(pays_transformation) | |
| mix_label = "RER" if is_eu else "GLO" | |
| result.actions_appliquees.append(f"3. Recherche dérivé soja dans GFLI mix {mix_label}") | |
| gfli = llm_service.smart_search_gfli(matiere, country_iso=mix_label) | |
| if gfli: | |
| result = _apply_gfli_result(gfli, result) | |
| result.actions_appliquees.append(f" → Trouvé dans GFLI ({mix_label}) : {gfli['nom_intrant']}") | |
| return result | |
| # Étape 4 : Process le plus proche | |
| result.actions_appliquees.append("4. Dernier recours — process le plus proche") | |
| alts = llm_service.find_alternative_materials(matiere, db_name="GFLI", country_hint=pays_transformation) | |
| if alts: | |
| result = _store_alternatives(alts, result) | |
| return result | |
| result.erreur = f"Aucune valeur trouvée pour '{matiere}' (dérivé soja, transfo {pays_transformation})." | |
| return result | |
| def _resolve_soja_node_9(matiere: str, result: CarbonResult) -> CarbonResult: | |
| """ | |
| Soja Node 9 : Pays origine inconnu + dérivé + lieu transfo inconnu. | |
| 1. Valeur la plus défavorable GFLI/ECOALIM | |
| 2. Process le plus proche | |
| """ | |
| result.actions_appliquees.append("1. Recherche valeur la plus défavorable pour dérivé soja (GFLI + ECOALIM)") | |
| best_val = None | |
| best_nom = None | |
| best_src = None | |
| best_unite = None | |
| # GFLI worst | |
| gfli_worst = data_loader.get_gfli_worst_value(matiere) | |
| if not gfli_worst: | |
| for term in ["soy", "soybean meal", "soybean"]: | |
| gfli_worst = data_loader.get_gfli_worst_value(term) | |
| if gfli_worst and _is_name_match(matiere, gfli_worst[1]): | |
| break | |
| gfli_worst = None | |
| if gfli_worst: | |
| val, nom, src = gfli_worst | |
| best_val, best_nom, best_src = val, nom, src | |
| best_unite = "kg CO2 eq / tonne de produit" | |
| # ECOALIM worst | |
| eco_worst = data_loader.get_ecoalim_worst_value(matiere) | |
| if not eco_worst: | |
| for term in ["soja", "tourteau de soja"]: | |
| eco_worst = data_loader.get_ecoalim_worst_value(term) | |
| if eco_worst: | |
| break | |
| if eco_worst: | |
| val_eco, nom_eco, src_eco = eco_worst | |
| val_eco_t = val_eco * 1000.0 | |
| if best_val is None or val_eco_t > (best_val if best_unite and "tonne" in best_unite else best_val * 1000.0): | |
| best_val = val_eco | |
| best_nom = nom_eco | |
| best_src = src_eco | |
| best_unite = "kg CO2 eq / kg de produit" | |
| if best_val is not None: | |
| result.impact_kg_co2_eq = best_val | |
| if "tonne" in best_unite: | |
| result.impact_tonne_co2_eq = best_val / 1000.0 | |
| else: | |
| result.impact_tonne_co2_eq = best_val | |
| result.unite_source = best_unite | |
| result.source_db = best_src | |
| result.intrant_utilise = best_nom | |
| result.match_exact = _is_name_match(matiere, best_nom) | |
| result.actions_appliquees.append(f" → Valeur la plus défavorable : {best_nom} dans {best_src}") | |
| return result | |
| # Étape 2 : Process le plus proche | |
| result.actions_appliquees.append("2. Dernier recours — process le plus proche") | |
| alts = llm_service.find_alternative_materials(matiere, db_name="GFLI") | |
| if alts: | |
| result = _store_alternatives(alts, result) | |
| return result | |
| result.erreur = f"Aucune valeur trouvée pour '{matiere}' (dérivé soja, provenance inconnue)." | |
| return result | |
| # ============================================================================ | |
| # Résolveurs MINERAL (logigramme_mineral.json, nodes 3-6) | |
| # ============================================================================ | |
| def _resolve_mineral_node_3(matiere: str, result: CarbonResult) -> CarbonResult: | |
| """ | |
| Minéral Node 3 : Provenance inconnue. | |
| 1. GFLI valeur la plus défavorable | |
| 2. ECOALIM valeur la plus défavorable | |
| 3. "Total Minerals, Additives, Vitamins" dans GFLI | |
| """ | |
| # Étape 1 : GFLI worst | |
| result.actions_appliquees.append("1. Recherche valeur la plus défavorable dans GFLI") | |
| gfli_worst = data_loader.get_gfli_worst_value(matiere) | |
| 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 | |
| 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: | |
| 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 | |
| result.match_exact = _is_name_match(matiere, nom) | |
| 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 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 | |
| if not eco_worst: | |
| eco_smart = llm_service.smart_search_ecoalim(matiere) | |
| if eco_smart: | |
| 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 | |
| result.unite_source = "kg CO2 eq / kg de produit" | |
| result.source_db = src | |
| result.intrant_utilise = nom | |
| result.match_exact = _is_name_match(matiere, nom) | |
| result.actions_appliquees.append(f" → Trouvé dans ECOALIM : {nom} = {val:.4f} kg CO2 eq/kg") | |
| return result | |
| # Étape 3 : Total Minerals, Additives, Vitamins dans GFLI | |
| result.actions_appliquees.append("3. Dernier recours — 'Total Minerals, Additives, Vitamins' dans GFLI") | |
| gfli_total = data_loader.get_gfli_worst_value("Total Minerals, Additives, Vitamins") | |
| if not gfli_total: | |
| gfli_total = data_loader.get_gfli_climate_value("Total Minerals") | |
| if gfli_total: | |
| val, nom, src = gfli_total | |
| 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 | |
| result.match_exact = False | |
| result.justification_alternative = ( | |
| f"Aucune donnée spécifique pour '{matiere}'. " | |
| f"Valeur par défaut '{nom}' utilisée conformément au logigramme minéral." | |
| ) | |
| result.actions_appliquees.append(f" → Valeur par défaut : {nom} = {val:.2f} kg CO2 eq/t") | |
| return result | |
| result.erreur = f"Aucune valeur trouvée pour '{matiere}' (minéral, provenance inconnue)." | |
| return result | |
| def _resolve_mineral_node_4(matiere: str, pays_production: str, result: CarbonResult) -> CarbonResult: | |
| """ | |
| Minéral Node 4 : Présent dans GFLI + provenance connue. | |
| 1. GFLI couple intrant/pays | |
| 2. ECOALIM couple intrant/pays | |
| 3. GFLI continent (RER pour Europe) | |
| 4. GFLI pays même continent (plus défavorable) | |
| 5. GFLI valeur monde (GLO) | |
| 6. GFLI autre pays autre continent (plus défavorable) | |
| """ | |
| country_iso = _get_country_iso(pays_production) | |
| is_eu = _is_european(pays_production) | |
| # Étape 1 : GFLI couple intrant/pays | |
| result.actions_appliquees.append(f"1. Recherche dans GFLI pour {pays_production}") | |
| gfli = llm_service.smart_search_gfli(matiere, country_iso=country_iso) | |
| if gfli: | |
| result = _apply_gfli_result(gfli, result) | |
| result.actions_appliquees.append(f" → Trouvé dans GFLI : {gfli['nom_intrant']}") | |
| return result | |
| # Étape 2 : ECOALIM couple intrant/pays | |
| result.actions_appliquees.append(f"2. Recherche dans ECOALIM pour {pays_production}") | |
| eco = llm_service.smart_search_ecoalim(matiere, pays_production=pays_production) | |
| if eco: | |
| result = _apply_ecoalim_result(eco, result) | |
| result.actions_appliquees.append(f" → Trouvé dans ECOALIM : {eco['nom_intrant']}") | |
| return result | |
| # Étape 3 : GFLI continent (RER pour Europe) | |
| mix_label = "RER" if is_eu else "GLO" | |
| result.actions_appliquees.append(f"3. Recherche dans GFLI mix {mix_label}") | |
| gfli = llm_service.smart_search_gfli(matiere, country_iso=mix_label) | |
| if gfli: | |
| result = _apply_gfli_result(gfli, result) | |
| result.actions_appliquees.append(f" → Trouvé dans GFLI ({mix_label}) : {gfli['nom_intrant']}") | |
| return result | |
| # Étape 4 : GFLI pays même continent — prendre le plus défavorable (on prend worst sans filtre pays) | |
| result.actions_appliquees.append("4. Recherche GFLI valeur la plus défavorable (même continent)") | |
| gfli_worst = data_loader.get_gfli_worst_value(matiere) | |
| if not gfli_worst: | |
| gfli_smart = llm_service.smart_search_gfli(matiere) | |
| if gfli_smart: | |
| base_name = gfli_smart["nom_intrant"].split(",")[0].split("/")[0].strip() | |
| gfli_worst = data_loader.get_gfli_worst_value(base_name) | |
| 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 | |
| result.match_exact = _is_name_match(matiere, nom) | |
| result.actions_appliquees.append(f" → GFLI (plus défavorable) : {nom} = {val:.2f} kg CO2 eq/t") | |
| return result | |
| # Étape 5 : GFLI GLO | |
| if mix_label != "GLO": | |
| result.actions_appliquees.append("5. Recherche dans GFLI mix GLO") | |
| gfli = llm_service.smart_search_gfli(matiere, country_iso="GLO") | |
| if gfli: | |
| result = _apply_gfli_result(gfli, result) | |
| result.actions_appliquees.append(f" → Trouvé dans GFLI (GLO) : {gfli['nom_intrant']}") | |
| return result | |
| # Étape 6 : GFLI autre pays (worst) | |
| result.actions_appliquees.append("6. Recherche GFLI autre pays (valeur la plus défavorable)") | |
| alts = llm_service.find_alternative_materials(matiere, db_name="GFLI", country_hint=pays_production) | |
| if alts: | |
| result = _store_alternatives(alts, result) | |
| return result | |
| result.erreur = f"Aucune valeur trouvée pour '{matiere}' (minéral, {pays_production}, GFLI)." | |
| return result | |
| def _resolve_mineral_node_5(matiere: str, pays_production: str, result: CarbonResult) -> CarbonResult: | |
| """ | |
| Minéral Node 5 : Absent GFLI, présent ECOALIM + provenance connue. | |
| 1. ECOALIM couple intrant/pays | |
| 2. ECOALIM même continent | |
| 3. ECOALIM pays même continent (plus défavorable) | |
| 4. ECOALIM valeur monde | |
| 5. ECOALIM autre pays autre continent (plus défavorable) | |
| """ | |
| # Étape 1 : ECOALIM couple intrant/pays | |
| result.actions_appliquees.append(f"1. Recherche dans ECOALIM pour {pays_production}") | |
| eco = llm_service.smart_search_ecoalim(matiere, pays_production=pays_production) | |
| if eco: | |
| result = _apply_ecoalim_result(eco, result) | |
| result.actions_appliquees.append(f" → Trouvé dans ECOALIM : {eco['nom_intrant']}") | |
| return result | |
| # Étape 2-3 : ECOALIM autre provenance / même continent — simplifié en worst value | |
| result.actions_appliquees.append("2-3. Recherche ECOALIM meilleure correspondance (continent / même continent)") | |
| eco = llm_service.smart_search_ecoalim(matiere) | |
| if eco: | |
| result = _apply_ecoalim_result(eco, result) | |
| result.actions_appliquees.append(f" → Trouvé dans ECOALIM : {eco['nom_intrant']}") | |
| return result | |
| # Étape 4 : ECOALIM valeur monde (worst) | |
| result.actions_appliquees.append("4. Recherche ECOALIM valeur la plus défavorable (monde)") | |
| eco_worst = data_loader.get_ecoalim_worst_value(matiere) | |
| if eco_worst: | |
| val, nom, src = eco_worst | |
| result.impact_kg_co2_eq = val | |
| result.impact_tonne_co2_eq = val | |
| result.unite_source = "kg CO2 eq / kg de produit" | |
| result.source_db = src | |
| result.intrant_utilise = nom | |
| result.match_exact = _is_name_match(matiere, nom) | |
| result.actions_appliquees.append(f" → ECOALIM (plus défavorable) : {nom}") | |
| return result | |
| # Étape 5 : Fallback | |
| result.actions_appliquees.append("5. Dernier recours — alternatives LLM") | |
| alts = llm_service.find_alternative_materials(matiere, db_name="ECOALIM", country_hint=pays_production) | |
| if alts: | |
| result = _store_alternatives(alts, result) | |
| return result | |
| result.erreur = f"Aucune valeur trouvée pour '{matiere}' (minéral, {pays_production}, ECOALIM)." | |
| return result | |
| def _resolve_mineral_node_6(matiere: str, result: CarbonResult) -> CarbonResult: | |
| """ | |
| Minéral Node 6 : Absent GFLI et ECOALIM. | |
| 1. "Total Minerals, Additives, Vitamins" dans GFLI. | |
| """ | |
| result.actions_appliquees.append("1. 'Total Minerals, Additives, Vitamins' dans GFLI (valeur par défaut)") | |
| gfli_total = data_loader.get_gfli_worst_value("Total Minerals, Additives, Vitamins") | |
| if not gfli_total: | |
| gfli_total = data_loader.get_gfli_climate_value("Total Minerals") | |
| if gfli_total: | |
| val, nom, src = gfli_total | |
| 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 | |
| result.match_exact = False | |
| result.justification_alternative = ( | |
| f"'{matiere}' absent de GFLI et ECOALIM. " | |
| f"Valeur par défaut '{nom}' utilisée conformément au logigramme minéral." | |
| ) | |
| result.actions_appliquees.append(f" → Valeur par défaut : {nom} = {val:.2f} kg CO2 eq/t") | |
| return result | |
| result.erreur = ( | |
| f"Aucune valeur trouvée pour '{matiere}' (minéral, absent de toutes les bases). " | |
| "Même le proxy 'Total Minerals, Additives, Vitamins' est introuvable." | |
| ) | |
| return result | |
| # ============================================================================ | |
| # Évaluation SOJA (logigramme complet) | |
| # ============================================================================ | |
| def _evaluate_soja( | |
| matiere_premiere: str, | |
| pays_production: Optional[str], | |
| pays_transformation: Optional[str], | |
| result: CarbonResult, | |
| ) -> CarbonResult: | |
| """Suit le logigramme soja (logigramme_soja.json).""" | |
| # Classification : le LLM a déjà classifié brut/transformé | |
| is_transformed = result.classification == "transforme" | |
| # Node 1 : Connaît-on le pays d'origine de la graine ? | |
| pays_origine_connu = bool(pays_production) | |
| if pays_origine_connu: | |
| # Node 2 : Niveau de transformation ? | |
| result.parcours.append(StepLog( | |
| node_id="soja_node_1", | |
| question="Connaissez-vous le pays d'origine de la graine à l'origine de l'intrant ?", | |
| answer=f"Oui — {pays_production}", | |
| )) | |
| if not is_transformed: | |
| result.parcours.append(StepLog( | |
| node_id="soja_node_2", | |
| question="Quel est le niveau de transformation ?", | |
| answer="Soja non transformé (graines crues)", | |
| )) | |
| result.node_resultat = "soja_node_4" | |
| result = _resolve_soja_node_4(matiere_premiere, pays_production, result) | |
| else: | |
| result.parcours.append(StepLog( | |
| node_id="soja_node_2", | |
| question="Quel est le niveau de transformation ?", | |
| answer="Produit dérivé du soja", | |
| )) | |
| result.node_resultat = "soja_node_5" | |
| result = _resolve_soja_node_5(matiere_premiere, pays_production, pays_transformation, result) | |
| else: | |
| # Node 3 : Niveau de transformation ? | |
| result.parcours.append(StepLog( | |
| node_id="soja_node_1", | |
| question="Connaissez-vous le pays d'origine de la graine à l'origine de l'intrant ?", | |
| answer="Non — provenance inconnue", | |
| )) | |
| if not is_transformed: | |
| result.parcours.append(StepLog( | |
| node_id="soja_node_3", | |
| question="Quel est le niveau de transformation ?", | |
| answer="Soja non transformé (graines crues)", | |
| )) | |
| result.node_resultat = "soja_node_6" | |
| result = _resolve_soja_node_6(matiere_premiere, result) | |
| else: | |
| result.parcours.append(StepLog( | |
| node_id="soja_node_3", | |
| question="Quel est le niveau de transformation ?", | |
| answer="Produit dérivé du soja", | |
| )) | |
| # Node 7 : Connaissez-vous le lieu de transformation ? | |
| lieu_transfo_connu = bool(pays_transformation) | |
| if lieu_transfo_connu: | |
| result.parcours.append(StepLog( | |
| node_id="soja_node_7", | |
| question="Connaissez-vous le lieu de transformation ?", | |
| answer=f"Oui — {pays_transformation}", | |
| )) | |
| result.node_resultat = "soja_node_8" | |
| result = _resolve_soja_node_8(matiere_premiere, pays_transformation, result) | |
| else: | |
| result.parcours.append(StepLog( | |
| node_id="soja_node_7", | |
| question="Connaissez-vous le lieu de transformation ?", | |
| answer="Non — lieu de transformation inconnu", | |
| )) | |
| result.node_resultat = "soja_node_9" | |
| result = _resolve_soja_node_9(matiere_premiere, result) | |
| return result | |
| # ============================================================================ | |
| # Évaluation MINERAL (logigramme complet) | |
| # ============================================================================ | |
| def _evaluate_mineral( | |
| matiere_premiere: str, | |
| pays_production: Optional[str], | |
| result: CarbonResult, | |
| ) -> CarbonResult: | |
| """Suit le logigramme minéral (logigramme_mineral.json).""" | |
| # Node 1 : Connaît-on l'origine ? | |
| provenance_connue = bool(pays_production) | |
| if not provenance_connue: | |
| result.parcours.append(StepLog( | |
| node_id="mineral_node_1", | |
| question="Connaissez-vous l'origine de l'intrant (pays de production) ?", | |
| answer="Non — provenance inconnue", | |
| )) | |
| result.node_resultat = "mineral_node_3" | |
| result = _resolve_mineral_node_3(matiere_premiere, result) | |
| else: | |
| result.parcours.append(StepLog( | |
| node_id="mineral_node_1", | |
| question="Connaissez-vous l'origine de l'intrant (pays de production) ?", | |
| answer=f"Oui — {pays_production}", | |
| )) | |
| # Node 2 : Dans quelle BDD l'intrant est-il présent ? | |
| # Check GFLI first | |
| gfli_results = data_loader.search_gfli(matiere_premiere) | |
| # Also try english translation | |
| if gfli_results.empty: | |
| matiere_en = llm_service.translate_matiere_to_english(matiere_premiere) | |
| if matiere_en.lower() != matiere_premiere.lower(): | |
| gfli_results = data_loader.search_gfli(matiere_en) | |
| eco_results = data_loader.search_ecoalim(matiere_premiere) | |
| if eco_results.empty: | |
| matiere_fr = llm_service.translate_matiere_to_french(matiere_premiere) | |
| if matiere_fr.lower() != matiere_premiere.lower(): | |
| eco_results = data_loader.search_ecoalim(matiere_fr) | |
| in_gfli = not gfli_results.empty | |
| in_ecoalim = not eco_results.empty | |
| if in_gfli: | |
| result.parcours.append(StepLog( | |
| node_id="mineral_node_2", | |
| question="Dans quelle(s) base(s) de données l'intrant est-il présent ?", | |
| answer="Présent dans GFLI", | |
| )) | |
| result.node_resultat = "mineral_node_4" | |
| result = _resolve_mineral_node_4(matiere_premiere, pays_production, result) | |
| elif in_ecoalim: | |
| result.parcours.append(StepLog( | |
| node_id="mineral_node_2", | |
| question="Dans quelle(s) base(s) de données l'intrant est-il présent ?", | |
| answer="Absent GFLI, présent dans ECOALIM", | |
| )) | |
| result.node_resultat = "mineral_node_5" | |
| result = _resolve_mineral_node_5(matiere_premiere, pays_production, result) | |
| else: | |
| result.parcours.append(StepLog( | |
| node_id="mineral_node_2", | |
| question="Dans quelle(s) base(s) de données l'intrant est-il présent ?", | |
| answer="Absent GFLI et ECOALIM", | |
| )) | |
| result.node_resultat = "mineral_node_6" | |
| result = _resolve_mineral_node_6(matiere_premiere, result) | |
| return result | |
| # ============================================================================ | |
| # Moteur principal | |
| # ============================================================================ | |
| def evaluate_carbon_impact( | |
| matiere_premiere: str, | |
| pays_production: Optional[str] = None, | |
| pays_transformation: Optional[str] = None, | |
| type_mp: str = "vegetal_animal", | |
| ) -> CarbonResult: | |
| """ | |
| Point d'entrée principal : évalue l'impact carbone d'une matière première | |
| en suivant le logigramme adapté au type de MP. | |
| 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 | |
| type_mp: "vegetal_animal" | "soja" | "mineral" | |
| Returns: | |
| CarbonResult avec toutes les informations | |
| """ | |
| logger.info(f"=== Début évaluation: {matiere_premiere} ===") | |
| logger.info(f"Type MP: {type_mp}, Pays production: {pays_production or 'inconnu'}, Pays transformation: {pays_transformation or 'N/A'}") | |
| 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 | |
| # ----------------------------------------------------------------------- | |
| if type_mp == "mineral": | |
| # Les minéraux ne suivent pas la classification brut/transformé | |
| result.classification = "mineral" | |
| result.classification_justification = "MP minérale / micro-ingrédient / additif" | |
| logger.info(f"Classification: Minérale (pas de classification brut/transformé)") | |
| result.parcours.append(StepLog( | |
| node_id="classification", | |
| question="Type de matière première", | |
| answer="Minérale / Micro-ingrédient / Additif", | |
| )) | |
| else: | |
| 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" | |
| logger.info(f"Classification: {'Transformée' if is_transformed else 'Brute'} - {result.classification_justification}") | |
| 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}", | |
| )) | |
| # ----------------------------------------------------------------------- | |
| # Dispatch selon le type de MP | |
| # ----------------------------------------------------------------------- | |
| logger.info(f"Dispatch vers logigramme: {type_mp}") | |
| if type_mp == "soja": | |
| result = _evaluate_soja(matiere_premiere, pays_production, pays_transformation, result) | |
| elif type_mp == "mineral": | |
| result = _evaluate_mineral(matiere_premiere, pays_production, result) | |
| else: | |
| # ---- Logigramme végétal/animal optimisé (logigramme.json) ---- | |
| is_transformed = result.classification == "transforme" | |
| provenance_connue = bool(pays_production) | |
| if not provenance_connue: | |
| # node_1 → Non → node_provenance_inconnue | |
| logger.info("Node 1: Provenance INCONNUE → Recherche valeur la plus défavorable") | |
| result.parcours.append(StepLog( | |
| node_id="node_1", | |
| question="Connaissez-vous l'endroit ou l'intrant a ete cultive ou produit ?", | |
| answer="Non — provenance inconnue", | |
| )) | |
| # node_provenance_inconnue : niveau de transformation | |
| if not is_transformed: | |
| result.parcours.append(StepLog( | |
| node_id="node_provenance_inconnue", | |
| question="Quel est le niveau de transformation de l'intrant ?", | |
| answer="Intrant brut", | |
| )) | |
| else: | |
| result.parcours.append(StepLog( | |
| node_id="node_provenance_inconnue", | |
| question="Quel est le niveau de transformation de l'intrant ?", | |
| answer="Intrant transforme ou coproduit", | |
| )) | |
| result.node_resultat = "node_base_gfli_defaut" | |
| logger.info("→ Résolution via Node 4 (valeur la plus défavorable)") | |
| result = _resolve_node_4(matiere_premiere, result) | |
| else: | |
| # node_1 → Oui → node_transformation | |
| logger.info(f"Node 1: Provenance CONNUE - Production: {pays_production}, Transformation: {pays_transformation or 'N/A'}") | |
| result.parcours.append(StepLog( | |
| node_id="node_1", | |
| question="Connaissez-vous l'endroit ou l'intrant a ete cultive ou produit ?", | |
| answer=f"Oui — Production: {pays_production}" + (f", Transformation: {pays_transformation}" if pays_transformation else ""), | |
| )) | |
| if not is_transformed: | |
| # node_transformation → Intrant brut → node_localisation_brut | |
| result.parcours.append(StepLog( | |
| node_id="node_transformation", | |
| question="Quel est le niveau de transformation de l'intrant ?", | |
| answer="Intrant brut", | |
| )) | |
| if _is_france(pays_production): | |
| # node_localisation_brut → Oui (France) → node_base_ecoalim | |
| logger.info("Intrant BRUT cultivé en FRANCE → Node 8 (ECOALIM prioritaire)") | |
| result.parcours.append(StepLog( | |
| node_id="node_localisation_brut", | |
| question="L'intrant brut est-il cultive en France ?", | |
| answer="Oui", | |
| )) | |
| result.node_resultat = "node_base_ecoalim" | |
| result = _resolve_node_8(matiere_premiere, result) | |
| else: | |
| # node_localisation_brut → Non → node_base_gfli_pays | |
| logger.info(f"Intrant BRUT cultivé HORS FRANCE ({pays_production}) → Node 9 (GFLI pays/RER/GLO)") | |
| result.parcours.append(StepLog( | |
| node_id="node_localisation_brut", | |
| question="L'intrant brut est-il cultive en France ?", | |
| answer=f"Non — {pays_production}", | |
| )) | |
| result.node_resultat = "node_base_gfli_pays" | |
| result = _resolve_node_9(matiere_premiere, pays_production, result) | |
| else: | |
| # node_transformation → Intrant transformé → node_localisation_transforme | |
| result.parcours.append(StepLog( | |
| node_id="node_transformation", | |
| question="Quel est le niveau de transformation de l'intrant ?", | |
| answer="Intrant transforme ou coproduit", | |
| )) | |
| if _is_france(pays_transformation) and _is_france(pays_production): | |
| # Transformé en France avec MP française → node_base_ecoalim | |
| result.parcours.append(StepLog( | |
| node_id="node_localisation_transforme", | |
| question="Ou l'intrant est-il transforme ?", | |
| answer="Transforme en France avec MP francaise", | |
| )) | |
| result.node_resultat = "node_base_ecoalim" | |
| result = _resolve_node_10(matiere_premiere, result) | |
| elif _is_france(pays_transformation): | |
| # Transformé en France avec MP non française ou inconnue → node_base_gfli_fr | |
| result.parcours.append(StepLog( | |
| node_id="node_localisation_transforme", | |
| question="Ou l'intrant est-il transforme ?", | |
| answer=f"Transforme en France avec MP non francaise ou inconnue ({pays_production or 'origine inconnue'})", | |
| )) | |
| result.node_resultat = "node_base_gfli_fr" | |
| result = _resolve_node_11(matiere_premiere, result) | |
| else: | |
| # Transformé hors France → node_base_gfli_pays | |
| result.parcours.append(StepLog( | |
| node_id="node_localisation_transforme", | |
| question="Ou l'intrant est-il transforme ?", | |
| answer=f"Transforme hors France — {pays_transformation}", | |
| )) | |
| result.node_resultat = "node_base_gfli_pays" | |
| result = _resolve_node_12(matiere_premiere, pays_transformation or pays_production or "", result) | |
| # ------------------------------------------------------------------ | |
| # Post-processing : normaliser les unités (t CO2 eq / t produit) | |
| # ------------------------------------------------------------------ | |
| logger.info("Post-processing: Normalisation des unités") | |
| if result.impact_kg_co2_eq is not None and result.unite_source: | |
| if "tonne" in result.unite_source: | |
| # GFLI : kg CO2 eq / tonne -> t CO2 eq / t | |
| result.impact_tonne_co2_eq = result.impact_kg_co2_eq / 1000.0 | |
| logger.debug(f"Conversion GFLI: {result.impact_kg_co2_eq} kg CO2/t → {result.impact_tonne_co2_eq} t CO2/t") | |
| else: | |
| # EcoALIM : kg CO2 eq / kg -> t CO2 eq / t (même valeur numérique) | |
| result.impact_tonne_co2_eq = result.impact_kg_co2_eq | |
| logger.debug(f"Conversion EcoALIM: {result.impact_kg_co2_eq} kg CO2/kg → {result.impact_tonne_co2_eq} t CO2/t (no numerical change)") | |
| # ------------------------------------------------------------------ | |
| # Post-processing : collecter les candidats alternatifs | |
| # ------------------------------------------------------------------ | |
| result = _collect_candidates(result) | |
| # Demander au LLM quel candidat est le plus pertinent en cas de doute | |
| if not result.match_exact and result.candidats_alternatifs: | |
| try: | |
| names = [c.get("nom", "") for c in result.candidats_alternatifs if c.get("nom")] | |
| rank = llm_service.rank_candidates(result.matiere_premiere, names) | |
| result.candidat_recommande = rank.get("best_name") | |
| result.candidats_reflexion = rank.get("reasoning") | |
| except Exception: | |
| result.candidat_recommande = None | |
| result.candidats_reflexion = None | |
| # 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})." | |
| ) | |
| # Log final result | |
| if result.erreur: | |
| logger.warning(f"=== Évaluation terminée avec ERREUR: {result.erreur} ===") | |
| else: | |
| logger.info(f"=== Évaluation terminée avec SUCCÈS ===") | |
| logger.info(f"Résultat: {result.intrant_utilise} = {result.impact_kg_co2_eq:.2f} {result.unite_source}") | |
| logger.info(f"Source: {result.source_db}, Match exact: {result.match_exact}, Node: {result.node_resultat}") | |
| 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 | |
| unbounded = not result.match_exact | |
| matiere_fr = llm_service.translate_matiere_to_french(matiere) | |
| matiere_en = llm_service.translate_matiere_to_english(matiere) | |
| 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=None if unbounded else 8, | |
| )) | |
| if matiere_fr.lower() != matiere.lower(): | |
| candidates.extend(data_loader.get_top_ecoalim_candidates( | |
| matiere_fr, | |
| pays_production=result.pays_production, | |
| pays_transformation=result.pays_transformation, | |
| top_n=None if unbounded else 8, | |
| )) | |
| candidates.extend(data_loader.get_top_gfli_candidates( | |
| matiere, country_iso=country_iso, top_n=None if unbounded else 4, | |
| )) | |
| if matiere_en.lower() != matiere.lower(): | |
| candidates.extend(data_loader.get_top_gfli_candidates( | |
| matiere_en, country_iso=country_iso, top_n=None if unbounded else 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=None if unbounded else 8, | |
| )) | |
| if matiere_en.lower() != matiere.lower(): | |
| candidates.extend(data_loader.get_top_gfli_candidates( | |
| matiere_en, country_iso=country_iso, top_n=None if unbounded else 8, | |
| )) | |
| candidates.extend(data_loader.get_top_ecoalim_candidates( | |
| matiere, | |
| pays_production=result.pays_production, | |
| pays_transformation=result.pays_transformation, | |
| top_n=None if unbounded else 4, | |
| )) | |
| if matiere_fr.lower() != matiere.lower(): | |
| candidates.extend(data_loader.get_top_ecoalim_candidates( | |
| matiere_fr, | |
| pays_production=result.pays_production, | |
| pays_transformation=result.pays_transformation, | |
| top_n=None if unbounded else 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 | |
| elif matiere_en and _is_name_match(matiere_en, c["nom"]): | |
| pass # OK, match en anglais | |
| elif matiere_fr and _is_name_match(matiere_fr, c["nom"]): | |
| pass # OK, match en français | |
| else: | |
| continue # Faux positif | |
| seen.add(key) | |
| unique_candidates.append(c) | |
| result.candidats_alternatifs = unique_candidates | |
| return result | |