""" 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) @dataclass class StepLog: """Un pas dans le parcours du logigramme.""" node_id: str question: Optional[str] answer: Optional[str] action: Optional[str] = None result: Optional[str] = None @dataclass class CarbonResult: """Résultat complet de l'évaluation carbone d'une matière première.""" matiere_premiere: str pays_production: Optional[str] pays_transformation: Optional[str] classification: str # "brut" | "transforme" classification_justification: str # Valeur finale impact_kg_co2_eq: Optional[float] = None impact_tonne_co2_eq: Optional[float] = None # conversion en tonnes unite_source: str = "" # "kg CO2 eq / kg" ou "kg CO2 eq / tonne" # Traçabilité source_db: str = "" # "ECOALIM" | "GFLI" intrant_utilise: str = "" # nom exact dans la BDD match_exact: bool = True justification_alternative: Optional[str] = None # Parcours de logique parcours: List[StepLog] = field(default_factory=list) node_resultat: str = "" # node_id du résultat actions_appliquees: List[str] = field(default_factory=list) # Candidats alternatifs (pour affichage comparatif quand match non exact) candidats_alternatifs: List[dict] = field(default_factory=list) 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