# executives/main.py
import asyncio
import sqlite3
import datetime
import warnings
import sys
import re
import textwrap
import os
import difflib
from ollama import AsyncClient
from groq import AsyncGroq
import nexus_config
from nexus_qualification import QualificationEngine
from nexus_matrix import DOMAINES
warnings.filterwarnings("ignore")
# =====================================================================
# MODULE 1 : BASE DE DONNÉES & LOGGING (MUTLI-THREAD)
# =====================================================================
class ShadowLogger:
def __init__(self):
self.conn = sqlite3.connect(nexus_config.DB_PATH, check_same_thread=False)
self.cursor = self.conn.cursor()
self.cursor.execute('''CREATE TABLE IF NOT EXISTS interactions_log (
id INTEGER PRIMARY KEY AUTOINCREMENT, date_log TEXT, ticket_complet TEXT,
domaine_principal TEXT, domaines_secondaires TEXT, score REAL, statut TEXT)''')
self.conn.commit()
def log(self, ticket_final, domaine_principal, sec_list, score, statut="CLOS"):
date_now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
sec_str = ", ".join(sec_list) if sec_list else "AUCUN"
ticket_assaini = str(ticket_final).encode('utf-8', errors='ignore').decode('utf-8')
try:
self.cursor.execute('''INSERT INTO interactions_log (date_log, ticket_complet, domaine_principal, domaines_secondaires, score, statut)
VALUES (?, ?, ?, ?, ?, ?)''', (date_now, ticket_assaini, domaine_principal, sec_str, score, statut))
self.conn.commit()
except sqlite3.OperationalError:
self.cursor.execute('DROP TABLE IF EXISTS interactions_log')
self.cursor.execute('''CREATE TABLE IF NOT EXISTS interactions_log (
id INTEGER PRIMARY KEY AUTOINCREMENT, date_log TEXT, ticket_complet TEXT,
domaine_principal TEXT, domaines_secondaires TEXT, score REAL, statut TEXT)''')
self.cursor.execute('''INSERT INTO interactions_log (date_log, ticket_complet, domaine_principal, domaines_secondaires, score, statut)
VALUES (?, ?, ?, ?, ?, ?)''', (date_now, ticket_assaini, domaine_principal, sec_str, score, statut))
self.conn.commit()
print("\n" + "📊 " + "=" * 25 + " DATABASE COMMIT OVERSIGHT " + "=" * 25)
print(f" 📅 Horodatage : {date_now}")
print(f" 📂 Statut Log : {statut}")
print(f" 🎯 Domaine Maître : {domaine_principal} | Urgence Score : {score}/10")
print(f" 🚨 Renforts Unit : {sec_str}")
print(f" 📝 Transcript Final Reconstitué :\n {ticket_assaini}")
print("=" * 78 + "\n")
# =====================================================================
# MODULE 2 : EXÉCUTION TACTIQUE
# =====================================================================
class TaskExecutor:
@staticmethod
def declencher_protocoles(domaines_impliques, niveau_urgence, resume_ticket, motif_cloture="STANDARD"):
print("\n ⚡ DÉCLENCHEMENT DES TÂCHES OPÉRATIONNELLES (MULTI-SERVICES) :")
if motif_cloture == "SILENCE_CRITIQUE":
print(" [ALERTE] 🚨 RUPTURE DE LIAISON DÉTECTÉE - DÉPLOIEMENT SUR DERNIÈRE POSITION CONNUE")
elif motif_cloture == "CRI_DETECTE":
print(" [ALERTE] 🚨 PANIQUE ACOUSTIQUE DÉTECTÉE - INTERVENTION MAXIMUM IMMÉDIATE")
elif motif_cloture == "FRUSTRATION_CRITIQUE":
print(" [ALERTE] ⚠️ RUPTURE DE COOPÉRATION - ARBITRAGE D'URGENCE : ENVOI IMMÉDIAT DES SECOURS")
if niveau_urgence == "⚡ PROTOCOLE JUPITER":
print(" [CONSEIL DE DÉFENSE] 🎖️ Activation immédiate du protocole souverain -> ALERTÉ")
print(" [GIGN / RAID] 🚁 Déploiement des forces d'intervention spéciales -> EN COURS")
elif niveau_urgence == "🟣 CONTINUITÉ DE L'ÉTAT":
print(" [CIC REGLÉ] 📡 Activation de la Cellule Interministérielle de Crise -> ALERTÉ")
elif "PLAN NOVI" in niveau_urgence:
print(" [PLAN NOVI] ⛺ Déploiement Poste Médical Avancé (PMA) et Plan Blanc -> EN COURS")
print(" [COD PRÉFET] 📡 Activation du Centre Opérationnel Départemental -> ALERTÉ")
elif "URGENCE VITALE" in niveau_urgence:
print(" [SAMU 15] 🚑 Envoi d'une Unité Mobile Hospitalière (SMUR) -> EN COURS")
for domaine in domaines_impliques:
print(f"\n --- Unité En Alerte : {domaine} ---")
# VARIÉTÉ DES DOMAINES : Le dispatcher ignore la basse priorité
if motif_cloture == "ABANDON" or domaine in ["EN_ATTENTE", "NON_URGENT", "DIGITAL SUPPORT", "ANIMALIER"]:
print(f" [DISPATCH] 🗑️ Traitement basse priorité / Administratif ({domaine}) -> ARCHIVÉ SANS SIRÈNE")
elif domaine == "POMPIER":
print(" [WEBHOOK] 🚒 Transmission au SDIS local (Code Rouge) -> OK")
print(" [TASK] 🗺️ Extraction des coordonnées GPS pour les engins -> EN COURS")
elif domaine == "POLICE":
print(" [WEBHOOK] 🚓 Alerte patrouille secteur en cours (Sécurisation) -> OK")
elif domaine == "ÉNERGIE & INFRASTRUCTURES" or domaine == "CYBERSÉCURITÉ":
print(" [WEBHOOK] ⚡ Alerte Cellule de Crise Infrastructure -> OK")
elif domaine == "MÉDICAL" and "URGENCE VITALE" not in niveau_urgence and "PLAN NOVI" not in niveau_urgence:
print(" [WEBHOOK] 🚑 Transmission du bilan au SAMU (Régulation 15) -> OK")
else:
if domaine != "MÉDICAL":
print(f" [DISPATCH] 📨 Transmission standard au centre {domaine} -> OK")
# =====================================================================
# MODULE 3 : MOTEUR COGNITIF HYBRIDE (LLM)
# =====================================================================
class NexusAgenticSystem:
def __init__(self):
print(f"🧠 Initialisation NEXUS V_ULTIME (Cloud & Local {nexus_config.MODEL_LOCAL})...")
self.logger = ShadowLogger()
self.evaluator = QualificationEngine()
self.executor = TaskExecutor()
self.client_llm = AsyncClient()
self.local_model = nexus_config.MODEL_LOCAL
self.llm_en_ligne = False
self.groq_key = os.getenv("GROQ_API_KEY")
self.groq_disponible = bool(self.groq_key)
if self.groq_disponible:
self.client_groq = AsyncGroq(api_key=self.groq_key)
print("🌐 [CLOUD] Client Groq initialisé.")
else:
print("⚠️ [WARN] Clé GROQ_API_KEY introuvable. Souveraineté locale exclusive.")
async def prechauffer_cerveau(self):
print(f"🔥 Pré-chauffage du bouclier local {self.local_model} en cours...")
try:
await self.client_llm.chat(model=self.local_model, messages=[{'role': 'user', 'content': 'ping'}],
options={'num_predict': 1})
self.llm_en_ligne = True
print(f"✅ Moteur local {self.local_model} paré et monté en RAM.")
except Exception as e:
self.llm_en_ligne = False
print(f"⚠️ ERREUR CRITIQUE LOCAL : Impossible de joindre Ollama. {e}")
if self.groq_disponible:
print(f"📡 Vérification du pipeline Groq Cloud ({nexus_config.MODEL_CLOUD})...")
try:
await asyncio.wait_for(
self.client_groq.chat.completions.create(
model=nexus_config.MODEL_CLOUD, messages=[{'role': 'user', 'content': 'ping'}],
max_tokens=1
),
timeout=5.0
)
print(f"✅ [STATUS] Inférence Cloud active ({nexus_config.MODEL_CLOUD}).")
except Exception as e:
self.groq_disponible = False
def est_salutation_basique(self, texte):
t_clean = texte.strip().lower()
t_clean = re.sub(r'[^\w\s]', '', t_clean)
return t_clean in ["bonjour", "salut", "allo", "allô", "bonsoir", "oui", "non", "ok"]
def detecter_choc_acoustique(self, texte):
t = texte.upper()
if re.search(r'\b(CASSÉ MON TÉLÉPHONE|CASSE MON TELEPHONE|PÉTÉ MON ORDI|PASSEPORT|CLAVIER)\b', t):
return False
if re.search(r'\b(AU SECOURS|AIDEZ-MOI|ÇA EXPLOSE|AU FEU|JE MEURS|AAAA|VITE VITE)\b', t):
return True
if len(t) > 5 and sum(1 for c in texte if c.isupper()) / len(texte) > 0.6:
return True
return False
def verifier_presence_localisation(self, texte):
t = texte.lower()
if re.search(r'\b(gare|aéroport|aeroport|hôpital|hopital|clinique|mairie|préfecture|prefecture|super u|élysée|elysee|banque)\b', t):
return True
if re.search(r'\b(rue|avenue|boulevard|impasse|allée|chemin|route|place|square|pont)\b', t):
return True
if re.search(r'\b\d{5}\b', t):
return True
return False
def get_regex_urgence_vitale(self):
return r'\b(arrêt cardiaque|arret cardiaque|crise cardiaque|infarctus|mal au coeur|mal au c[oœ]ur|poitrine|coeur|cœur|cardiaque|' \
r'inconscient|coma|évanoui|evanoui|évanouie|evanouie|malaise|tomber dans les pommes|' \
r'étouffe|etouffe|respire plus|asphyxie|' \
r'saigne|sang|hémorragie|hemorragie|amputé|ampute|' \
r'mourir|meurt|vie en danger|vie et mort)\b'
def verifier_protocole_jupiter(self, texte):
t = str(texte).lower()
cible_president = re.search(r'\b(président|president|macron|élysée|elysée|elysee|centrale|nucléaire|nucleaire)\b', t)
if not cible_president:
return None
# FUZZY MATCHING : Tolérance aux erreurs de frappe en urgence souveraine
mots_menaces_internes = [
'arme', 'tir', 'tirs', 'assassinat', 'tuer', 'bombe', 'attaque',
'sang', 'inconscient', 'coma', 'évanoui', 'évanouie', 'étouffe', 'cardiaque',
'coeur', 'poitrine', 'infarctus', 'saigne', 'hémorragie', 'mourir',
'feu', 'incendie', 'explosion', 'fuite', 'toxique', 'chimique'
]
mots_texte = re.findall(r'\w+', t)
menace_valide = False
if re.search(self.get_regex_urgence_vitale() + r'|\b(arme|tir|tirs|assassinat|tuer|bombe|attaque|feu|incendie|explosion|fuite)\b', t):
menace_valide = True
else:
for mot in mots_texte:
if len(mot) < 4: continue
match = difflib.get_close_matches(mot, mots_menaces_internes, n=1, cutoff=0.8)
if match:
menace_valide = True
break
if menace_valide:
if re.search(self.get_regex_urgence_vitale(), t) or any(difflib.get_close_matches(m, mots_menaces_internes[8:20], 1, 0.8) for m in mots_texte):
return "MÉDICAL", 20.0, ["POLICE", "POMPIER", "TRANSPORT & MOBILITÉ", "RISQUES ENVIRONNEMENTAUX"]
return "POLICE", 20.0, ["MÉDICAL", "POMPIER", "CYBERSÉCURITÉ", "TRANSPORT & MOBILITÉ"]
return None
def verifier_mots_clefs_vitaux(self, texte):
t_lower = str(texte).lower()
if re.search(r'\b(pété mon|cassé mon|va me tuer|téléphone|telephone|ordi|clavier|souris)\b', t_lower):
return False
if re.search(self.get_regex_urgence_vitale(), t_lower):
return True
# FUZZY MATCHING : Pour le citoyen normal
mots_vitaux_isoles = [
'évanoui', 'évanouie', 'étouffe', 'inconscient', 'malaise', 'coeur',
'poitrine', 'infarctus', 'mourir', 'saigne',
'amputé', 'hémorragie', 'suicide', 'cardiaque'
]
mots_texte = re.findall(r'\w+', t_lower)
for mot in mots_texte:
if len(mot) < 4: continue
correspondances = difflib.get_close_matches(mot, mots_vitaux_isoles, n=1, cutoff=0.8)
if correspondances:
return True
return False
def verifier_urgences_combinees(self, texte):
t_lower = str(texte).lower()
if re.search(r'\b(clavier|souris|ordi|ordinateur|téléphone|telephone|connexion|wifi|mot de passe|imprimante)\b', t_lower):
return "DIGITAL SUPPORT", 1.5, []
has_transport = re.search(r'\b(bus|cars|car|train|métro|metro|tram|avion)\b', t_lower)
has_accident = re.search(r'\b(accident|crash|collision|déraillement|renversé)\b', t_lower)
has_groupe = re.search(r'\b(groupe|foule|plusieurs|masse|nombreux|personnes|gens|équipe|classe|centaine|cinquantaine)\b', t_lower)
has_feu = re.search(r'\b(feu|incendie|brûle|brule|flammes)\b', t_lower)
has_chimique = re.search(r'\b(chimique|fuite|gaz|toxique|nuage)\b', t_lower)
has_detresse = re.search(self.get_regex_urgence_vitale() + r'|\b(blessé|blesse|victime|victimes|tête|tete|fièvre|fievre)\b', t_lower)
bonus_masse = 0.0
if has_groupe: bonus_masse += 1.8
if has_transport: bonus_masse += 1.2
bonus_degats = 0.0
if has_accident: bonus_degats += 1.4
if has_feu: bonus_degats += 2.1
if has_chimique: bonus_degats += 2.7
if has_detresse: bonus_degats += 1.5
if (has_transport or has_groupe) and (has_accident or has_feu or has_chimique or has_detresse):
score_total_crise = 10.0 + bonus_masse + bonus_degats
domaine = "POMPIER" if (has_feu or has_accident) else "MÉDICAL"
if has_chimique: domaine = "RISQUES ENVIRONNEMENTAUX"
renforts = ["MÉDICAL", "POLICE"]
if has_transport: renforts.append("TRANSPORT & MOBILITÉ")
if has_feu or has_chimique: renforts.append("POMPIER")
return domaine, min(round(score_total_crise, 1), 19.9), list(set(renforts))
if re.search(r'\b(suicide|suicidaire|suicider)\b', t_lower):
return "MÉDICAL", 10.0, ["POLICE"]
if has_feu and has_detresse:
return "POMPIER", 11.5, ["MÉDICAL", "POLICE"]
return None
async def generer_question_bot(self, transcript, texte_utilisateur, domaine, score_actuel, state):
texte_utilisateur = str(texte_utilisateur).encode('utf-8', errors='ignore').decode('utf-8')
transcript = str(transcript).encode('utf-8', errors='ignore').decode('utf-8')
if self.est_salutation_basique(texte_utilisateur):
return "EN_COURS", "Ici les urgences. Quel est votre problème et où vous trouvez-vous ?"
if not self.groq_disponible and not self.llm_en_ligne:
return "EN_COURS", "Décrivez votre urgence et votre adresse exacte."
# RÈGLES DYNAMIQUES DE PROMPT (Fin de la boucle infinie pour l'IT)
is_basse_priorite = domaine in ["EN_ATTENTE", "NON_URGENT", "DIGITAL SUPPORT", "ANIMALIER"] or score_actuel < 3.0
if is_basse_priorite:
instruction_lieu = "- LIEU : INUTILE. C'est une basse priorité. Ne demande pas d'adresse. Fais un diagnostic technique ou administratif simple."
etat_dispatch = "NON APPLICABLE (Basse priorité)."
regles_dynamiques = """
- AUCUNE politesse. MAXIMUM 2 phrases.
- [DÉCLENCHEMENT] INTERDIT. Ne génère jamais le token ###DISPATCH###.
- [CLÔTURE] Dès que tu as réorienté l'utilisateur vers le bon service ou donné ton conseil final, réponds EXCLUSIVEMENT par le token : ###CLOS###"""
else:
if state.get("localisation_obtenue", False):
instruction_lieu = "- LIEU : [DÉJÀ OBTENU]. NE DEMANDE PLUS L'ADRESSE. Concentre-toi EXCLUSIVEMENT sur la nature du danger ou l'état de la victime."
else:
instruction_lieu = "- LIEU : Demande un repère ou le nom d'un bâtiment si aucune adresse n'est donnée."
etat_dispatch = "DÉJÀ DÉCLENCHÉS. Les unités sont en route. Concentre-toi sur le maintien en vie." if state.get("secours_declenches", False) else "NON DÉCLENCHÉS. Objectif: Obtenir Lieu + Nature pour déclencher les secours."
regles_dynamiques = """
- AUCUNE politesse. MAXIMUM 2 phrases.
- [DÉCLENCHEMENT] Si les secours sont NON DÉCLENCHÉS ET que tu as compris le lieu exact ET la nature du danger, commence OBLIGATOIREMENT ta réponse par le token ###DISPATCH###. Annonce que les secours sont en route.
- [CLÔTURE] Si les secours sont DÉJÀ DÉCLENCHÉS ET que l'arrivée physique des unités est confirmée, réponds EXCLUSIVEMENT par : ###CLOS###"""
prompt_base = f"""Tu es NEXUS, l'IA de régulation d'urgence de l'État français.
Ton rôle : Autorité bienveillante, directivité chirurgicale. Tu N'ES PAS un robot de discussion, tu es un opérateur tactique.
=== DONNÉES SYSTÈME ===
- Domaine actuel : {domaine}
- Gravité estimée (Score) : {score_actuel}/10
- Statut des Secours : {etat_dispatch}
=== HISTORIQUE DE LA LIAISON ===
{transcript}
=== PROTOCOLE D'ANALYSE ===
1. Qualification de l'appelant : Victime ou Témoin ?
2. Extraction : {instruction_lieu}
=== RÈGLES ABSOLUES ===
{regles_dynamiques}
Génère ta réponse finale directe à l'appelant :"""
contenu = ""
if self.groq_disponible:
try:
response = await asyncio.wait_for(
self.client_groq.chat.completions.create(
model=nexus_config.MODEL_CLOUD,
messages=[
{'role': 'system', 'content': prompt_base},
{'role': 'user', 'content': texte_utilisateur}
],
temperature=0.0, max_tokens=60
),
timeout=5.0
)
contenu = response.choices[0].message.content.strip()
except Exception:
self.groq_disponible = False
if not contenu and self.llm_en_ligne:
try:
prompt_complet = prompt_base + f"\n\nMessage du client : \"{texte_utilisateur}\""
reponse = await asyncio.wait_for(
self.client_llm.chat(
model=self.local_model,
messages=[{'role': 'user', 'content': prompt_complet}],
options={'temperature': 0.0}
),
timeout=45.0
)
contenu = reponse['message']['content'].strip()
contenu = re.sub(r'.*?', '', contenu, flags=re.DOTALL | re.IGNORECASE).strip()
except Exception:
pass
if not contenu:
if is_basse_priorite:
return "COMPLET", ""
if state.get("localisation_obtenue", False):
return "EN_COURS", "Analyse en cours. Précisez immédiatement l'état des victimes et la nature du danger."
else:
return "EN_COURS", "Analyse en cours. Où vous trouvez-vous exactement ?"
is_dispatch = False
if "###CLOS###" in contenu:
if is_basse_priorite:
return "COMPLET", ""
if not state.get("secours_declenches", False) and not state.get("localisation_obtenue", False):
return "EN_COURS", "Urgence identifiée. Donnez-moi un repère visuel précis ou une rue pour envoyer les secours."
return "COMPLET", ""
if "###DISPATCH###" in contenu:
if state.get("localisation_obtenue", False):
if not is_basse_priorite and score_actuel >= 5.0:
is_dispatch = True
contenu = contenu.replace("###DISPATCH###", "").strip()
contenu = re.sub(r'^(Régulateur\s*:|Client\s*:|Ta réponse\s*:|Réponse\s*:|:\s*|"\s*)', '', contenu, flags=re.IGNORECASE).strip('" ')
return "DISPATCH" if is_dispatch else "EN_COURS", contenu
async def extraire_domaines_secondaires(self, texte_utilisateur, domaine_principal):
texte_utilisateur = str(texte_utilisateur).encode('utf-8', errors='ignore').decode('utf-8')
if domaine_principal in ["NON_URGENT", "EN_ATTENTE", "DIGITAL SUPPORT", "ANIMALIER"]: return []
prompt = f"Analyse : '{texte_utilisateur}'. Principal : {domaine_principal}. Besoins de renforts ? Choix : {', '.join(DOMAINES)}. Règles : Pas le principal. Si rien : AUCUN. Si plusieurs, sépare par virgules. Aucun autre texte."
res = ""
if self.groq_disponible:
try:
response = await asyncio.wait_for(
self.client_groq.chat.completions.create(
model=nexus_config.MODEL_CLOUD, messages=[{'role': 'user', 'content': prompt}],
temperature=0.0, max_tokens=25
),
timeout=3.0
)
res = response.choices[0].message.content.strip().upper()
except:
pass
if not res and self.llm_en_ligne:
try:
reponse = await asyncio.wait_for(
self.client_llm.chat(
model=self.local_model,
messages=[{'role': 'user', 'content': prompt}],
options={'temperature': 0.0, 'num_predict': 30}
),
timeout=30.0
)
res = reponse['message']['content'].strip().upper()
res = re.sub(r'.*?', '', res, flags=re.DOTALL | re.IGNORECASE).strip()
except Exception:
pass
if not res or "AUCUN" in res: return []
return list(set([d.strip() for d in res.split(',') if d.strip() in DOMAINES and d.strip() != domaine_principal]))
# =====================================================================
# SOUS-ROUTINE ASYNCHRONE OPTIMISÉE
# =====================================================================
async def async_input(prompt: str, timeout: float):
loop = asyncio.get_event_loop()
print(prompt, end="", flush=True)
queue = asyncio.Queue()
def got_input():
line = sys.stdin.readline()
loop.call_soon_threadsafe(queue.put_nowait, line)
loop.add_reader(sys.stdin.fileno(), got_input)
try:
ligne = await asyncio.wait_for(queue.get(), timeout)
# Éradication de la faille de la touche d'effacement (\udcc3)
texte_propre = ligne.strip()
return str(texte_propre).encode('utf-8', errors='ignore').decode('utf-8')
except asyncio.TimeoutError:
print()
return ""
finally:
loop.remove_reader(sys.stdin.fileno())
# =====================================================================
# BOUCLE PRINCIPALE DE RÉGULATION
# =====================================================================
async def run_terminal():
nexus = NexusAgenticSystem()
await nexus.prechauffer_cerveau()
print("\n" + "=" * 70)
print(f"🚀 NEXUS COMMAND CENTER — PIPELINE TEMPS RÉEL V42 (RÉGULATION TOTALE)")
print("=" * 70 + "\n")
while True:
raw = await async_input("📝 Client (Début d'appel) : ", timeout=86400)
if raw.lower() in {"exit", "q", "quit"}: break
if not raw: continue
ticket_final = raw
transcript = f"Client : {raw}\n"
ticket_complet = False
silence_count = 0
motif_fermeture = "STANDARD"
skip_generation = False
forced_secondaries = []
session_state = {
"domaine_verrouille": "EN_ATTENTE",
"score_verrouille": 0.0,
"localisation_obtenue": False,
"secours_declenches": False,
"domaines_secondaires_lock": [],
"frustration_count": 0
}
if not nexus.est_salutation_basique(ticket_final):
dom_c, sco_c, _ = nexus.evaluator.evaluer_ticket(ticket_final)
redressement_jupiter = nexus.verifier_protocole_jupiter(ticket_final)
if redressement_jupiter:
dom_c, sco_c, forced_secondaries = redressement_jupiter
print(" [CONSEIL DE DÉFENSE] ⚡ PROTOCOLE JUPITER ENGAGÉ : URGENCE ABSOLUE D'ÉTAT (20/10).")
else:
redressement = nexus.verifier_urgences_combinees(ticket_final)
if redressement:
dom_c, sco_c, forced_secondaries = redressement
print(f" [ALERTE] ⚡ SURCHARGE DIRECTE INTER-CLASSES ACTIVÉE : {dom_c} ({sco_c}/10)")
elif nexus.verifier_mots_clefs_vitaux(ticket_final):
dom_c = "MÉDICAL"
sco_c = max(sco_c, 10.0)
print(" [ALERTE] 🛡️ BOUCLIER HEURISTIQUE ENGAGÉ : URGENCE VITALE FORCÉE.")
session_state["domaine_verrouille"] = dom_c
session_state["score_verrouille"] = sco_c
while not ticket_complet:
if not session_state["localisation_obtenue"] and nexus.verifier_presence_localisation(ticket_final):
session_state["localisation_obtenue"] = True
if nexus.detecter_choc_acoustique(ticket_final):
print(" [ALERTE] 💥 PANIQUE ACOUSTIQUE DÉTECTÉE - SEUIL PLANCHER D'URGENCE ACTIVÉ")
session_state["domaine_verrouille"] = "MÉDICAL"
session_state["score_verrouille"] = 10.0
motif_fermeture = "CRI_DETECTE"
if silence_count > 0:
ticket_complet = True
break
if re.search(r'\b(fausse alerte|je rigolais|un film)\b', ticket_final.lower()):
print(" [INFO] 📉 Désescalade détectée. Clôture immédiate pour appel nul.")
session_state["score_verrouille"] = 0.2
session_state["domaine_verrouille"] = "DIGITAL SUPPORT"
ticket_complet = True
break
pattern_frust = re.compile(r'\b(bordel|foutre|con|merde|putain|répéter|déjà dit|tu sers à rien)\b')
if pattern_frust.search(ticket_final.lower()):
session_state["frustration_count"] += 1
if session_state["frustration_count"] >= 2:
if session_state["score_verrouille"] >= 5.0:
motif_fermeture = "FRUSTRATION_CRITIQUE"
else:
motif_fermeture = "ABANDON"
ticket_complet = True
break
if not nexus.est_salutation_basique(ticket_final) and motif_fermeture != "CRI_DETECTE":
domaine_courant, score_courant, _ = nexus.evaluator.evaluer_ticket(ticket_final)
redressement_jupiter = nexus.verifier_protocole_jupiter(ticket_final)
if redressement_jupiter:
domaine_courant, score_courant, forced_secondaries = redressement_jupiter
else:
redressement = nexus.verifier_urgences_combinees(ticket_final)
if redressement:
domaine_courant, score_courant, forced_secondaries = redressement
elif nexus.verifier_mots_clefs_vitaux(ticket_final):
domaine_courant = "MÉDICAL"
score_courant = max(score_courant, 10.0)
domaines_regaliens = ["MÉDICAL", "POLICE", "POMPIER", "CYBERSÉCURITÉ", "ÉNERGIE & INFRASTRUCTURES",
"RISQUES ENVIRONNEMENTAUX"]
if score_courant >= 20.0:
session_state["domaine_verrouille"] = domaine_courant
session_state["score_verrouille"] = score_courant
else:
if session_state["domaine_verrouille"] in domaines_regaliens:
if score_courant > session_state["score_verrouille"]:
session_state["score_verrouille"] = score_courant
else:
session_state["domaine_verrouille"] = domaine_courant
session_state["score_verrouille"] = score_courant
domaine_maitre = session_state["domaine_verrouille"]
score_maitre = session_state["score_verrouille"]
if not skip_generation:
statut, question_bot = await nexus.generer_question_bot(transcript, ticket_final, domaine_maitre,
score_maitre, session_state)
if statut == "COMPLET":
ticket_complet = True
break
wrapped_bot = textwrap.fill(
f"🤖 NEXUS ({domaine_maitre} | Score: {score_maitre}/10) : {question_bot}",
width=70, initial_indent=" ", subsequent_indent=" "
)
print(wrapped_bot)
if statut == "DISPATCH" and not session_state["secours_declenches"]:
print("\n [SYSTÈME] ⚠️ TRANSMISSION ANTICIPÉE AUX UNITÉS ENGAGÉE (Liaison maintenue) ⚠️")
if score_maitre >= 20.0:
niveau = "⚡ PROTOCOLE JUPITER"
elif score_maitre >= 15.0:
niveau = "🟣 CONTINUITÉ DE L'ÉTAT"
elif score_maitre >= 10.1:
niveau = "⚫ PLAN NOVI (MASSE)"
elif score_maitre >= 10.0:
niveau = "🔴 URGENCE VITALE ABSOLUE"
elif score_maitre >= 5.0:
niveau = "🟠 URGENCE MAJEURE"
else:
niveau = "🟢 ROUTINE"
domaines_secondaires = await nexus.extraire_domaines_secondaires(ticket_final, domaine_maitre)
if forced_secondaries:
domaines_secondaires = list(set(domaines_secondaires + forced_secondaries))
domaines_secondaires = [d for d in domaines_secondaires if d != domaine_maitre]
domaines_impliques = [domaine_maitre] + domaines_secondaires
nexus.executor.declencher_protocoles(domaines_impliques, niveau, ticket_final, "ANTICIPÉ")
session_state["secours_declenches"] = True
session_state["domaines_secondaires_lock"] = domaines_secondaires
print(" " + "-" * 67 + "\n")
skip_generation = False
complement = await async_input(" 💬 Client : ", timeout=45.0)
if complement == "":
silence_count += 1
if domaine_maitre in ["MÉDICAL", "POLICE", "POMPIER", "RISQUES ENVIRONNEMENTAUX"] or score_maitre >= 5.0:
if silence_count == 1:
print(" 🤖 NEXUS (RELANCE CRITIQUE) : Allô ? Restez en ligne, répondez si vous m'entendez !")
skip_generation = True
continue
else:
motif_fermeture = "SILENCE_CRITIQUE"
ticket_complet = True
break
else:
print(" ⚠️ APPEL DE BASSE PRIORITÉ ABANDONNÉ PAR L'UTILISATEUR (Délai de réponse > 45s).")
motif_fermeture = "ABANDON"
ticket_complet = True
break
silence_count = 0
if complement.lower() in {"exit", "q", "quit"}:
ticket_final = "exit"
break
ticket_final += " " + complement
transcript += f"Régulateur : {question_bot}\nClient : {complement}\n"
if ticket_final == "exit": break
if not session_state.get("secours_declenches", False):
if score_maitre >= 20.0:
niveau = "⚡ PROTOCOLE JUPITER"
elif score_maitre >= 15.0:
niveau = "🟣 CONTINUITÉ DE L'ÉTAT"
elif score_maitre >= 10.1:
niveau = "⚫ PLAN NOVI (MASSE)"
elif score_maitre >= 10.0:
niveau = "🔴 URGENCE VITALE ABSOLUE"
elif score_maitre >= 5.0:
niveau = "🟠 URGENCE MAJEURE"
else:
niveau = "🟢 ROUTINE"
print(f"\n ✅ APPEL CLOS — TRANSMISSION AUX UNITÉS")
print(f" 🎯 Service Principal : {domaine_maitre} | 🔢 Score Tactique : {score_maitre}/10 → {niveau}")
domaines_secondaires = await nexus.extraire_domaines_secondaires(ticket_final, domaine_maitre)
if forced_secondaries:
domaines_secondaires = list(set(domaines_secondaires + forced_secondaries))
if motif_fermeture in ["CRI_DETECTE", "SILENCE_CRITIQUE", "FRUSTRATION_CRITIQUE"] or score_maitre >= 15.0:
domaines_secondaires.extend(["POLICE", "POMPIER", "MÉDICAL"])
domaines_secondaires = list(set(domaines_secondaires))
domaines_secondaires = [d for d in domaines_secondaires if d != domaine_maitre]
if domaines_secondaires:
print(f" 🚨 Renforts requis identifiés : {', '.join(domaines_secondaires)}")
domaines_impliques = [domaine_maitre] + domaines_secondaires
nexus.executor.declencher_protocoles(domaines_impliques, niveau, ticket_final, motif_fermeture)
else:
print(f"\n ✅ APPEL CLOS — UNITÉS DÉJÀ DÉPLOYÉES SUR LE TERRAIN")
domaines_secondaires = session_state.get("domaines_secondaires_lock", [])
nexus.logger.log(ticket_final, domaine_maitre, domaines_secondaires, score_maitre, statut=motif_fermeture)
print("-" * 70 + "\n📝 En attente du prochain appelant...\n")
if __name__ == "__main__":
asyncio.run(run_terminal())