592 lines
28 KiB
Python
592 lines
28 KiB
Python
import os
|
|
import sys
|
|
from networkx.drawing.nx_agraph import read_dot
|
|
|
|
from utils.config import (
|
|
REFERENCE_GRAPH_PATH,
|
|
determine_threshold_color, get_weight_for_color
|
|
)
|
|
|
|
def parse_graphs(graphe_path):
|
|
"""
|
|
Charge et analyse les graphes DOT (analyse et référence).
|
|
"""
|
|
print(graphe_path)
|
|
# Charger le graphe à analyser
|
|
if not os.path.exists(graphe_path):
|
|
print(f"Fichier de graphe à analyser introuvable: {graphe_path}")
|
|
sys.exit(1)
|
|
|
|
# Charger le graphe de référence
|
|
reference_path = REFERENCE_GRAPH_PATH
|
|
if not os.path.exists(reference_path):
|
|
print(f"Fichier de graphe de référence introuvable: {reference_path}")
|
|
sys.exit(1)
|
|
|
|
try:
|
|
# Charger les graphes avec NetworkX
|
|
graph = read_dot(graphe_path)
|
|
ref_graph = read_dot(reference_path)
|
|
|
|
# Convertir les attributs en types appropriés pour les deux graphes
|
|
for g in [graph, ref_graph]:
|
|
for node, attrs in g.nodes(data=True):
|
|
for key, value in list(attrs.items()):
|
|
# Convertir les valeurs numériques
|
|
if key in ['niveau', 'ihh_acteurs', 'ihh_pays', 'isg', 'ivc']:
|
|
try:
|
|
if key in ['isg', 'ivc', 'ihh_acteurs', 'ihh_pays', 'niveau']:
|
|
attrs[key] = int(value.strip('"'))
|
|
else:
|
|
attrs[key] = float(value.strip('"'))
|
|
except (ValueError, TypeError):
|
|
# Garder la valeur originale si la conversion échoue
|
|
pass
|
|
elif key == 'label':
|
|
# Nettoyer les guillemets des étiquettes
|
|
attrs[key] = value.strip('"')
|
|
|
|
# Convertir les attributs des arêtes
|
|
for u, v, attrs in g.edges(data=True):
|
|
for key, value in list(attrs.items()):
|
|
if key in ['ics', 'cout', 'delai', 'technique']:
|
|
try:
|
|
attrs[key] = float(value.strip('"'))
|
|
except (ValueError, TypeError):
|
|
pass
|
|
elif key == 'label' and '%' in value:
|
|
# Extraire le pourcentage
|
|
try:
|
|
percentage = value.strip('"').replace('%', '')
|
|
attrs['percentage'] = float(percentage)
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
return graph, ref_graph
|
|
|
|
except Exception as e:
|
|
print(f"Erreur lors de l'analyse des graphes: {str(e)}")
|
|
sys.exit(1)
|
|
|
|
def extract_data_from_graph(graph, ref_graph):
|
|
"""
|
|
Extrait toutes les données pertinentes des graphes DOT.
|
|
"""
|
|
data = {
|
|
"products": {}, # Produits finaux (N0)
|
|
"components": {}, # Composants (N1)
|
|
"minerals": {}, # Minerais (N2)
|
|
"operations": {}, # Opérations (N10)
|
|
"countries": {}, # Pays (N11)
|
|
"geo_countries": {}, # Pays géographiques (N99)
|
|
"actors": {} # Acteurs (N12)
|
|
}
|
|
|
|
# Extraire tous les pays géographiques du graphe de référence
|
|
for node, attrs in ref_graph.nodes(data=True):
|
|
if attrs.get('niveau') == 99:
|
|
country_name = attrs.get('label', node)
|
|
isg_value = attrs.get('isg', 0)
|
|
|
|
data["geo_countries"][country_name] = {
|
|
"id": node,
|
|
"isg": isg_value
|
|
}
|
|
|
|
# Extraire les nœuds du graphe à analyser
|
|
for node, attrs in graph.nodes(data=True):
|
|
level = attrs.get('niveau', -1)
|
|
label = attrs.get('label', node)
|
|
|
|
if level == 0 or level == 1000: # Produit final
|
|
data["products"][node] = {
|
|
"label": label,
|
|
"components": [],
|
|
"assembly": None,
|
|
"level": level
|
|
}
|
|
elif level == 1 or level == 1001: # Composant
|
|
data["components"][node] = {
|
|
"label": label,
|
|
"minerals": [],
|
|
"manufacturing": None
|
|
}
|
|
elif level == 2: # Minerai
|
|
data["minerals"][node] = {
|
|
"label": label,
|
|
"ivc": attrs.get('ivc', 0),
|
|
"extraction": None,
|
|
"treatment": None,
|
|
"ics_values": {}
|
|
}
|
|
elif level == 10 or level == 1010: # Opération
|
|
op_type = label.lower()
|
|
data["operations"][node] = {
|
|
"label": label,
|
|
"type": op_type,
|
|
"ihh_acteurs": attrs.get('ihh_acteurs', 0),
|
|
"ihh_pays": attrs.get('ihh_pays', 0),
|
|
"countries": {}
|
|
}
|
|
elif level == 11 or level == 1011: # Pays
|
|
data["countries"][node] = {
|
|
"label": label,
|
|
"actors": {},
|
|
"geo_country": None,
|
|
"market_share": 0
|
|
}
|
|
elif level == 12 or level == 1012: # Acteur
|
|
data["actors"][node] = {
|
|
"label": label,
|
|
"country": None,
|
|
"market_share": 0
|
|
}
|
|
|
|
# Extraire les relations et attributs des arêtes
|
|
for source, target, edge_attrs in graph.edges(data=True):
|
|
if source not in graph.nodes or target not in graph.nodes:
|
|
continue
|
|
|
|
source_level = graph.nodes[source].get('niveau', -1)
|
|
target_level = graph.nodes[target].get('niveau', -1)
|
|
|
|
# Extraire part de marché
|
|
market_share = 0
|
|
if 'percentage' in edge_attrs:
|
|
market_share = edge_attrs['percentage']
|
|
elif 'label' in edge_attrs and '%' in edge_attrs['label']:
|
|
try:
|
|
market_share = float(edge_attrs['label'].strip('"').replace('%', ''))
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
# Relations produit → composant
|
|
if (source_level == 0 and target_level == 1) or (source_level == 1000 and target_level == 1001):
|
|
if target not in data["products"][source]["components"]:
|
|
data["products"][source]["components"].append(target)
|
|
|
|
# Relations produit → opération (assemblage)
|
|
elif (source_level == 0 and target_level == 10) or (source_level == 1000 and target_level == 1010):
|
|
if graph.nodes[target].get('label', '').lower() == 'assemblage':
|
|
data["products"][source]["assembly"] = target
|
|
|
|
# Relations composant → minerai avec ICS
|
|
elif (source_level == 1 or source_level == 1001) and target_level == 2:
|
|
if target not in data["components"][source]["minerals"]:
|
|
data["components"][source]["minerals"].append(target)
|
|
|
|
# Stocker l'ICS s'il est présent
|
|
if 'ics' in edge_attrs:
|
|
ics_value = edge_attrs['ics']
|
|
data["minerals"][target]["ics_values"][source] = ics_value
|
|
|
|
# Relations composant → opération (fabrication)
|
|
elif (source_level == 1 or source_level == 1001) and target_level == 10:
|
|
if graph.nodes[target].get('label', '').lower() == 'fabrication':
|
|
data["components"][source]["manufacturing"] = target
|
|
|
|
# Relations minerai → opération (extraction/traitement)
|
|
elif source_level == 2 and target_level == 10:
|
|
op_label = graph.nodes[target].get('label', '').lower()
|
|
if 'extraction' in op_label:
|
|
data["minerals"][source]["extraction"] = target
|
|
elif 'traitement' in op_label:
|
|
data["minerals"][source]["treatment"] = target
|
|
|
|
# Relations opération → pays avec part de marché
|
|
elif (source_level == 10 and target_level == 11) or (source_level == 1010 and target_level == 1011):
|
|
data["operations"][source]["countries"][target] = market_share
|
|
data["countries"][target]["market_share"] = market_share
|
|
|
|
# Relations pays → acteur avec part de marché
|
|
elif (source_level == 11 and target_level == 12) or (source_level == 1011 and target_level == 1012):
|
|
data["countries"][source]["actors"][target] = market_share
|
|
data["actors"][target]["market_share"] = market_share
|
|
data["actors"][target]["country"] = source
|
|
|
|
# Relations pays → pays géographique
|
|
elif (source_level == 11 or source_level == 1011) and target_level == 99:
|
|
country_name = graph.nodes[target].get('label', '')
|
|
data["countries"][source]["geo_country"] = country_name
|
|
|
|
# Compléter les opérations manquantes pour les produits et composants
|
|
# en les récupérant du graphe de référence si elles existent
|
|
|
|
# Pour les produits finaux (N0)
|
|
for product_id, product_data in data["products"].items():
|
|
if product_data["assembly"] is None:
|
|
# Chercher l'opération d'assemblage dans le graphe de référence
|
|
for source, target, edge_attrs in ref_graph.edges(data=True):
|
|
if (source == product_id and
|
|
((ref_graph.nodes[source].get('niveau') == 0 and
|
|
ref_graph.nodes[target].get('niveau') == 10) or
|
|
(ref_graph.nodes[source].get('niveau') == 1000 and
|
|
ref_graph.nodes[target].get('niveau') == 1010)) and
|
|
ref_graph.nodes[target].get('label', '').lower() == 'assemblage'):
|
|
|
|
# L'opération existe dans le graphe de référence
|
|
assembly_id = target
|
|
product_data["assembly"] = assembly_id
|
|
|
|
# Ajouter l'opération si elle n'existe pas déjà
|
|
if assembly_id not in data["operations"]:
|
|
data["operations"][assembly_id] = {
|
|
"label": ref_graph.nodes[assembly_id].get('label', assembly_id),
|
|
"type": "assemblage",
|
|
"ihh_acteurs": ref_graph.nodes[assembly_id].get('ihh_acteurs', 0),
|
|
"ihh_pays": ref_graph.nodes[assembly_id].get('ihh_pays', 0),
|
|
"countries": {}
|
|
}
|
|
|
|
# Extraire les relations de l'opération vers les pays
|
|
for op_source, op_target, op_edge_attrs in ref_graph.edges(data=True):
|
|
if (op_source == assembly_id and
|
|
(ref_graph.nodes[op_target].get('niveau') == 11 or ref_graph.nodes[op_target].get('niveau') == 1011)):
|
|
|
|
country_id = op_target
|
|
|
|
# Extraire part de marché
|
|
market_share = 0
|
|
if 'percentage' in op_edge_attrs:
|
|
market_share = op_edge_attrs['percentage']
|
|
elif 'label' in op_edge_attrs and '%' in op_edge_attrs['label']:
|
|
try:
|
|
market_share = float(op_edge_attrs['label'].strip('"').replace('%', ''))
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
# Ajouter le pays à l'opération
|
|
data["operations"][assembly_id]["countries"][country_id] = market_share
|
|
|
|
# Ajouter le pays s'il n'existe pas déjà
|
|
if country_id not in data["countries"]:
|
|
data["countries"][country_id] = {
|
|
"label": ref_graph.nodes[country_id].get('label', country_id),
|
|
"actors": {},
|
|
"geo_country": None,
|
|
"market_share": market_share
|
|
}
|
|
else:
|
|
data["countries"][country_id]["market_share"] = market_share
|
|
|
|
# Extraire les relations du pays vers les acteurs
|
|
for country_source, country_target, country_edge_attrs in ref_graph.edges(data=True):
|
|
if (country_source == country_id and
|
|
(ref_graph.nodes[country_target].get('niveau') == 12 or ref_graph.nodes[country_target].get('niveau') == 1012)):
|
|
|
|
actor_id = country_target
|
|
|
|
# Extraire part de marché
|
|
actor_market_share = 0
|
|
if 'percentage' in country_edge_attrs:
|
|
actor_market_share = country_edge_attrs['percentage']
|
|
elif 'label' in country_edge_attrs and '%' in country_edge_attrs['label']:
|
|
try:
|
|
actor_market_share = float(country_edge_attrs['label'].strip('"').replace('%', ''))
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
# Ajouter l'acteur au pays
|
|
data["countries"][country_id]["actors"][actor_id] = actor_market_share
|
|
|
|
# Ajouter l'acteur s'il n'existe pas déjà
|
|
if actor_id not in data["actors"]:
|
|
data["actors"][actor_id] = {
|
|
"label": ref_graph.nodes[actor_id].get('label', actor_id),
|
|
"country": country_id,
|
|
"market_share": actor_market_share
|
|
}
|
|
else:
|
|
data["actors"][actor_id]["market_share"] = actor_market_share
|
|
data["actors"][actor_id]["country"] = country_id
|
|
|
|
# Extraire la relation du pays vers le pays géographique
|
|
for geo_source, geo_target, geo_edge_attrs in ref_graph.edges(data=True):
|
|
if (geo_source == country_id and
|
|
ref_graph.nodes[geo_target].get('niveau') == 99):
|
|
|
|
geo_country_name = ref_graph.nodes[geo_target].get('label', '')
|
|
data["countries"][country_id]["geo_country"] = geo_country_name
|
|
|
|
break # Une seule opération d'assemblage par produit
|
|
|
|
# Pour les composants (N1)
|
|
for component_id, component_data in data["components"].items():
|
|
if component_data["manufacturing"] is None:
|
|
# Chercher l'opération de fabrication dans le graphe de référence
|
|
for source, target, edge_attrs in ref_graph.edges(data=True):
|
|
if (source == component_id and
|
|
((ref_graph.nodes[source].get('niveau') == 1 and
|
|
ref_graph.nodes[target].get('niveau') == 10) or
|
|
(ref_graph.nodes[source].get('niveau') == 1001 and
|
|
ref_graph.nodes[target].get('niveau') == 1010)) and
|
|
ref_graph.nodes[target].get('label', '').lower() == 'fabrication'):
|
|
|
|
# L'opération existe dans le graphe de référence
|
|
manufacturing_id = target
|
|
component_data["manufacturing"] = manufacturing_id
|
|
|
|
# Ajouter l'opération si elle n'existe pas déjà
|
|
if manufacturing_id not in data["operations"]:
|
|
data["operations"][manufacturing_id] = {
|
|
"label": ref_graph.nodes[manufacturing_id].get('label', manufacturing_id),
|
|
"type": "fabrication",
|
|
"ihh_acteurs": ref_graph.nodes[manufacturing_id].get('ihh_acteurs', 0),
|
|
"ihh_pays": ref_graph.nodes[manufacturing_id].get('ihh_pays', 0),
|
|
"countries": {}
|
|
}
|
|
|
|
# Extraire les relations de l'opération vers les pays
|
|
for op_source, op_target, op_edge_attrs in ref_graph.edges(data=True):
|
|
if (op_source == manufacturing_id and
|
|
(ref_graph.nodes[op_target].get('niveau') == 11 or ref_graph.nodes[op_target].get('niveau') == 1011)):
|
|
|
|
country_id = op_target
|
|
|
|
# Extraire part de marché
|
|
market_share = 0
|
|
if 'percentage' in op_edge_attrs:
|
|
market_share = op_edge_attrs['percentage']
|
|
elif 'label' in op_edge_attrs and '%' in op_edge_attrs['label']:
|
|
try:
|
|
market_share = float(op_edge_attrs['label'].strip('"').replace('%', ''))
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
# Ajouter le pays à l'opération
|
|
data["operations"][manufacturing_id]["countries"][country_id] = market_share
|
|
|
|
# Ajouter le pays s'il n'existe pas déjà
|
|
if country_id not in data["countries"]:
|
|
data["countries"][country_id] = {
|
|
"label": ref_graph.nodes[country_id].get('label', country_id),
|
|
"actors": {},
|
|
"geo_country": None,
|
|
"market_share": market_share
|
|
}
|
|
else:
|
|
data["countries"][country_id]["market_share"] = market_share
|
|
|
|
# Extraire les relations du pays vers les acteurs
|
|
for country_source, country_target, country_edge_attrs in ref_graph.edges(data=True):
|
|
if (country_source == country_id and
|
|
(ref_graph.nodes[country_target].get('niveau') == 12 or ref_graph.nodes[country_target].get('niveau') == 1012)):
|
|
|
|
actor_id = country_target
|
|
|
|
# Extraire part de marché
|
|
actor_market_share = 0
|
|
if 'percentage' in country_edge_attrs:
|
|
actor_market_share = country_edge_attrs['percentage']
|
|
elif 'label' in country_edge_attrs and '%' in country_edge_attrs['label']:
|
|
try:
|
|
actor_market_share = float(country_edge_attrs['label'].strip('"').replace('%', ''))
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
# Ajouter l'acteur au pays
|
|
data["countries"][country_id]["actors"][actor_id] = actor_market_share
|
|
|
|
# Ajouter l'acteur s'il n'existe pas déjà
|
|
if actor_id not in data["actors"]:
|
|
data["actors"][actor_id] = {
|
|
"label": ref_graph.nodes[actor_id].get('label', actor_id),
|
|
"country": country_id,
|
|
"market_share": actor_market_share
|
|
}
|
|
else:
|
|
data["actors"][actor_id]["market_share"] = actor_market_share
|
|
data["actors"][actor_id]["country"] = country_id
|
|
|
|
# Extraire la relation du pays vers le pays géographique
|
|
for geo_source, geo_target, geo_edge_attrs in ref_graph.edges(data=True):
|
|
if (geo_source == country_id and
|
|
ref_graph.nodes[geo_target].get('niveau') == 99):
|
|
|
|
geo_country_name = ref_graph.nodes[geo_target].get('label', '')
|
|
data["countries"][country_id]["geo_country"] = geo_country_name
|
|
|
|
break # Une seule opération de fabrication par composant
|
|
|
|
return data
|
|
|
|
def calculate_vulnerabilities(data, config):
|
|
"""
|
|
Calcule les vulnérabilités combinées pour toutes les opérations et minerais.
|
|
"""
|
|
thresholds = config.get('thresholds', {})
|
|
results = {
|
|
"ihh_isg_combined": {}, # Pour chaque opération
|
|
"ics_ivc_combined": {}, # Pour chaque minerai
|
|
"chains": [] # Pour stocker tous les chemins possibles
|
|
}
|
|
|
|
# 1. Calculer ISG_combiné pour chaque opération
|
|
for op_id, operation in data["operations"].items():
|
|
isg_weighted_sum = 0
|
|
total_share = 0
|
|
|
|
# Parcourir chaque pays impliqué dans l'opération
|
|
for country_id, share in operation["countries"].items():
|
|
country = data["countries"][country_id]
|
|
geo_country = country.get("geo_country")
|
|
|
|
if geo_country and geo_country in data["geo_countries"]:
|
|
isg_value = data["geo_countries"][geo_country]["isg"]
|
|
isg_weighted_sum += isg_value * share
|
|
total_share += share
|
|
|
|
# Calculer la moyenne pondérée
|
|
isg_combined = 0
|
|
if total_share > 0:
|
|
isg_combined = isg_weighted_sum / total_share
|
|
|
|
# Déterminer couleurs et poids
|
|
ihh_value = operation["ihh_pays"]
|
|
ihh_color, ihh_suffix = determine_threshold_color(ihh_value, "IHH", thresholds)
|
|
isg_color, isg_suffix = determine_threshold_color(isg_combined, "ISG", thresholds)
|
|
|
|
# Calculer poids combiné
|
|
ihh_weight = get_weight_for_color(ihh_color)
|
|
isg_weight = get_weight_for_color(isg_color)
|
|
combined_weight = ihh_weight * isg_weight
|
|
|
|
# Déterminer vulnérabilité combinée
|
|
if combined_weight in [6, 9]:
|
|
vulnerability = "ÉLEVÉE à CRITIQUE"
|
|
elif combined_weight in [3, 4]:
|
|
vulnerability = "MOYENNE"
|
|
else: # 1, 2
|
|
vulnerability = "FAIBLE"
|
|
|
|
# Stocker résultats
|
|
results["ihh_isg_combined"][op_id] = {
|
|
"ihh_value": ihh_value,
|
|
"ihh_color": ihh_color,
|
|
"ihh_suffix": ihh_suffix,
|
|
"isg_combined": isg_combined,
|
|
"isg_color": isg_color,
|
|
"isg_suffix": isg_suffix,
|
|
"combined_weight": combined_weight,
|
|
"vulnerability": vulnerability
|
|
}
|
|
|
|
# 2. Calculer ICS_moyen pour chaque minerai
|
|
for mineral_id, mineral in data["minerals"].items():
|
|
ics_values = list(mineral["ics_values"].values())
|
|
ics_average = 0
|
|
|
|
if ics_values:
|
|
ics_average = sum(ics_values) / len(ics_values)
|
|
|
|
ivc_value = mineral.get("ivc", 0)
|
|
|
|
# Déterminer couleurs et poids
|
|
ics_color, ics_suffix = determine_threshold_color(ics_average, "ICS", thresholds)
|
|
ivc_color, ivc_suffix = determine_threshold_color(ivc_value, "IVC", thresholds)
|
|
|
|
# Calculer poids combiné
|
|
ics_weight = get_weight_for_color(ics_color)
|
|
ivc_weight = get_weight_for_color(ivc_color)
|
|
combined_weight = ics_weight * ivc_weight
|
|
|
|
# Déterminer vulnérabilité combinée
|
|
if combined_weight in [6, 9]:
|
|
vulnerability = "ÉLEVÉE à CRITIQUE"
|
|
elif combined_weight in [3, 4]:
|
|
vulnerability = "MOYENNE"
|
|
else: # 1, 2
|
|
vulnerability = "FAIBLE"
|
|
|
|
# Stocker résultats
|
|
results["ics_ivc_combined"][mineral_id] = {
|
|
"ics_average": ics_average,
|
|
"ics_color": ics_color,
|
|
"ics_suffix": ics_suffix,
|
|
"ivc_value": ivc_value,
|
|
"ivc_color": ivc_color,
|
|
"ivc_suffix": ivc_suffix,
|
|
"combined_weight": combined_weight,
|
|
"vulnerability": vulnerability
|
|
}
|
|
|
|
# 3. Identifier tous les chemins et leurs vulnérabilités
|
|
for product_id, product in data["products"].items():
|
|
for component_id in product["components"]:
|
|
component = data["components"][component_id]
|
|
|
|
for mineral_id in component["minerals"]:
|
|
mineral = data["minerals"][mineral_id]
|
|
|
|
# Collecter toutes les vulnérabilités dans ce chemin
|
|
path_vulnerabilities = []
|
|
|
|
# Assemblage (si présent)
|
|
assembly_id = product["assembly"]
|
|
if assembly_id and assembly_id in results["ihh_isg_combined"]:
|
|
path_vulnerabilities.append({
|
|
"type": "assemblage",
|
|
"vulnerability": results["ihh_isg_combined"][assembly_id]["vulnerability"],
|
|
"operation_id": assembly_id
|
|
})
|
|
|
|
# Fabrication (si présent)
|
|
manufacturing_id = component["manufacturing"]
|
|
if manufacturing_id and manufacturing_id in results["ihh_isg_combined"]:
|
|
path_vulnerabilities.append({
|
|
"type": "fabrication",
|
|
"vulnerability": results["ihh_isg_combined"][manufacturing_id]["vulnerability"],
|
|
"operation_id": manufacturing_id
|
|
})
|
|
|
|
# Minerai (ICS+IVC)
|
|
if mineral_id in results["ics_ivc_combined"]:
|
|
path_vulnerabilities.append({
|
|
"type": "minerai",
|
|
"vulnerability": results["ics_ivc_combined"][mineral_id]["vulnerability"],
|
|
"mineral_id": mineral_id
|
|
})
|
|
|
|
# Extraction (si présent)
|
|
extraction_id = mineral["extraction"]
|
|
if extraction_id and extraction_id in results["ihh_isg_combined"]:
|
|
path_vulnerabilities.append({
|
|
"type": "extraction",
|
|
"vulnerability": results["ihh_isg_combined"][extraction_id]["vulnerability"],
|
|
"operation_id": extraction_id
|
|
})
|
|
|
|
# Traitement (si présent)
|
|
treatment_id = mineral["treatment"]
|
|
if treatment_id and treatment_id in results["ihh_isg_combined"]:
|
|
path_vulnerabilities.append({
|
|
"type": "traitement",
|
|
"vulnerability": results["ihh_isg_combined"][treatment_id]["vulnerability"],
|
|
"operation_id": treatment_id
|
|
})
|
|
|
|
# Classifier le chemin
|
|
path_info = {
|
|
"product": product_id,
|
|
"component": component_id,
|
|
"mineral": mineral_id,
|
|
"vulnerabilities": path_vulnerabilities
|
|
}
|
|
|
|
# Déterminer le niveau de risque du chemin
|
|
critical_count = path_vulnerabilities.count({"vulnerability": "ÉLEVÉE à CRITIQUE"})
|
|
medium_count = path_vulnerabilities.count({"vulnerability": "MOYENNE"})
|
|
|
|
if any(v["vulnerability"] == "ÉLEVÉE à CRITIQUE" for v in path_vulnerabilities):
|
|
path_info["risk_level"] = "critique"
|
|
elif medium_count >= 3:
|
|
path_info["risk_level"] = "majeur"
|
|
elif any(v["vulnerability"] == "MOYENNE" for v in path_vulnerabilities):
|
|
path_info["risk_level"] = "moyen"
|
|
else:
|
|
path_info["risk_level"] = "faible"
|
|
|
|
results["chains"].append(path_info)
|
|
|
|
return results
|