nexus-api / executives /nexus_qualification.py
DominusRed824's picture
Update executives/nexus_qualification.py
f2bd7d5 verified
Raw
History Blame Contribute Delete
6.48 kB
# executives/nexus_qualification.py
import joblib
import os
import re
MODEL_UNIFIED_PATH = "../pickle_result/nexus_modele_final_V35.pkl"
# =====================================================================
# MATRICE LÉGALE DE L'ÉTAT FRANÇAIS (SGDSN / DISPOSITIF ORSEC)
# =====================================================================
MATRICE_CIBLES_ISOLEES = {
"MÉDICAL": {"CITOYEN": 1, "VIP_LOCAL": 2, "PRESIDENT": 5},
"POLICE": {"CITOYEN": 1, "VIP_LOCAL": 2, "PRESIDENT": 5},
"POMPIER": {"CITOYEN": 1, "VIP_LOCAL": 2, "PRESIDENT": 5},
"CYBERSÉCURITÉ": {"CITOYEN": 1, "VIP_LOCAL": 3, "PRESIDENT": 4},
"ÉNERGIE & INFRASTRUCTURES": {"CITOYEN": 1, "VIP_LOCAL": 3, "PRESIDENT": 4},
"TRANSPORT & MOBILITÉ": {"CITOYEN": 1, "VIP_LOCAL": 3, "PRESIDENT": 4},
"RISQUES ENVIRONNEMENTAUX": {"CITOYEN": 1, "VIP_LOCAL": 3, "PRESIDENT": 3},
"SERVICES URBAINS": {"CITOYEN": 1, "VIP_LOCAL": 2, "PRESIDENT": 2},
"ASSISTANCE SOCIALE": {"CITOYEN": 1, "VIP_LOCAL": 2, "PRESIDENT": 2},
"ANIMALIER": {"CITOYEN": 1, "VIP_LOCAL": 2, "PRESIDENT": 2},
"DIGITAL SUPPORT": {"CITOYEN": 1, "VIP_LOCAL": 1, "PRESIDENT": 1}
}
class QualificationEngine:
def __init__(self):
if os.path.exists(MODEL_UNIFIED_PATH):
self.model = joblib.load(MODEL_UNIFIED_PATH)
self.mode = "ML"
print(f"🔌 Moteur d'inférence chargé : {MODEL_UNIFIED_PATH}")
else:
self.model = None
self.mode = "SIMULATION"
print(f"⚠️ ERREUR FATALE : Modèle ML introuvable à {MODEL_UNIFIED_PATH}. Mode SIMULATION (0.2) activé.")
def traduire_cible_ml(self, cible_id):
cible_id = int(cible_id)
if cible_id == 0: return "CITOYEN"
elif cible_id == 2: return "VIP_LOCAL"
elif cible_id == 4: return "PRESIDENT"
return "CITOYEN"
def calculer_priorite_sgdsn(self, domaine, severite, impact, profil_cible, texte_brut=""):
t_lower = str(texte_brut).lower()
# 1. COUPE-CIRCUIT ABSOLU (PROTOCOLE JUPITER : 18.0 - 20.0)
# CORRECTION : Une centrale nucléaire ou un site souverain est traité au même niveau que le Chef de l'État
is_souverain = (profil_cible == "PRESIDENT") or bool(re.search(r'\b(centrale|nucléaire|nucleaire)\b', t_lower))
if is_souverain:
if domaine in ["MÉDICAL", "POLICE", "POMPIER", "RISQUES ENVIRONNEMENTAUX"] and severite >= 3:
return 20.0
elif domaine in ["CYBERSÉCURITÉ", "ÉNERGIE & INFRASTRUCTURES", "TRANSPORT & MOBILITÉ"]:
return 20.0
# 2. PLAFONDS DE VERRE (Niveau 1 : Domaines Bridés)
if domaine in ["NON_URGENT", "EN_ATTENTE"]:
return 0.2
if domaine == "DIGITAL SUPPORT":
return min((0.5 + severite * 0.5 + impact * 0.2), 2.9) if profil_cible == "CITOYEN" else 4.9
if domaine == "ANIMALIER":
return min((0.5 + severite * 1.0 + impact * 0.5), 2.9)
if domaine in ["SERVICES URBAINS", "ASSISTANCE SOCIALE"]:
return min((0.5 + severite * 1.0 + impact * 1.0), 4.9)
# SCANNER DE MASSE (PLAN NOVI)
has_groupe = re.search(r'\b(groupe|foule|plusieurs|masse|nombreux|personnes|gens|équipe|classe|centaine|cinquantaine)\b', t_lower)
has_transport = re.search(r'\b(bus|cars|car|train|métro|metro|tram|avion|navire)\b', t_lower)
# 3. DOCTRINE DE GUERRE / MASSE (PLAN NOVI / CIBLES STRATÉGIQUES : 10.1 - 15.0)
if (severite == 4 and domaine in ["MÉDICAL", "POMPIER", "POLICE", "RISQUES ENVIRONNEMENTAUX", "CYBERSÉCURITÉ"]) or \
(severite >= 3 and profil_cible == "VIP_LOCAL" and domaine in ["CYBERSÉCURITÉ", "ÉNERGIE & INFRASTRUCTURES", "MÉDICAL", "TRANSPORT & MOBILITÉ"]) or \
(has_groupe and (domaine in ["POMPIER", "MÉDICAL", "POLICE"])):
base_vitale = 10.0
delta_masse = 0.0
if impact == 2: delta_masse = 1.5
elif impact == 3: delta_masse = 3.0
elif impact == 4: delta_masse = 4.5
if has_groupe: delta_masse = max(delta_masse, 3.5)
if has_transport: delta_masse = max(delta_masse, 4.0)
delta_infra = 0.0
if profil_cible == "VIP_LOCAL" or re.search(r'\b(hôpital|hopital|clinique|école|ecole|gare|aéroport)\b', t_lower):
delta_infra = 4.0
score_final = base_vitale + delta_masse + delta_infra
if domaine in ["CYBERSÉCURITÉ", "ÉNERGIE & INFRASTRUCTURES"]:
return min(round(score_final, 1), 15.0)
return min(round(score_final, 1), 19.9)
# 4. ÉCHELLE CIVILE DYNAMIQUE (Urgence Vitale Individuelle : 5.0 - 10.0)
poids_cible = MATRICE_CIBLES_ISOLEES.get(domaine, {}).get(profil_cible, 1)
score_base = 0.5 + (severite * 1.5) + (impact * 1.0) + (poids_cible * 0.5)
bonus_gravite = 0.0
mots_stress = ['très', 'tres', 'ultra', 'grave', 'gravement', 'fort', 'forte', 'énorme', 'enorme', 'extrême', 'extreme', 'pire', 'vite', 'urgence']
if any(mot in t_lower for mot in mots_stress):
bonus_gravite += 1.2
mots_cliniques = ['sang', 'saigne', 'douleur', 'pression', 'brûlure', 'brule', 'coupé', 'coupe', 'cassé', 'casse', 'tomber', 'tombé']
if any(mot in t_lower for mot in mots_cliniques):
bonus_gravite += 1.8
mots_liste = t_lower.split()
if len(mots_liste) <= 4 and bonus_gravite > 0:
bonus_gravite += 0.8
score_normalise = score_base + bonus_gravite
return min(round(score_normalise, 1), 10.0)
def evaluer_ticket(self, texte):
if self.mode == "ML":
prediction = self.model.predict([texte])[0]
domaine = str(prediction[0]).upper()
severite = int(float(prediction[1]))
impact = int(float(prediction[2]))
cible_id = int(float(prediction[3]))
friction = str(prediction[4]).upper()
profil_cible = self.traduire_cible_ml(cible_id)
score = self.calculer_priorite_sgdsn(domaine, severite, impact, profil_cible, str(texte))
return domaine, score, friction
else:
return "DIGITAL SUPPORT", 0.2, "MANQUE_LIEU"