575 lines
28 KiB
Python
575 lines
28 KiB
Python
import requests
|
||
import time
|
||
import argparse
|
||
import json
|
||
import sys
|
||
import uuid
|
||
from pathlib import Path
|
||
from typing import List, Dict, Any, Optional, Tuple
|
||
from enum import Enum
|
||
|
||
# Configuration de l'API PrivateGPT
|
||
PGPT_URL = "http://127.0.0.1:8001"
|
||
API_URL = f"{PGPT_URL}/v1"
|
||
DEFAULT_INPUT_PATH = "rapport_final.md"
|
||
DEFAULT_OUTPUT_PATH = "rapport_analyse_genere.md"
|
||
MAX_CONTEXT_LENGTH = 3000 # Taille maximale du contexte en caractères
|
||
TEMP_DIR = Path("temp_sections") # Répertoire pour les fichiers intermédiaires
|
||
|
||
# Stratégies de gestion du contexte
|
||
class ContextStrategy(str, Enum):
|
||
NONE = "none" # Pas de contexte
|
||
TRUNCATE = "truncate" # Tronquer le contexte
|
||
SUMMARIZE = "summarize" # Résumer le contexte
|
||
LATEST = "latest" # Uniquement la dernière section
|
||
FILE = "file" # Utiliser des fichiers ingérés (meilleure option)
|
||
|
||
# Les 5 étapes de génération du rapport
|
||
steps = [
|
||
{
|
||
"title": "Analyse Principale",
|
||
"prompt": """Vous êtes un assistant stratégique expert chargé de produire des rapports destinés à des décideurs de haut niveau (COMEX, directions risques, stratèges politiques ou industriels).
|
||
Objectif : Analyser les vulnérabilités systémiques dans une chaîne de valeur numérique décrite dans le fichier rapport_final.md en vous appuyant exclusivement sur ce rapport structuré fourni.
|
||
|
||
Votre tâche est de développer les sections suivantes en respectant toutes les contraintes décrites :
|
||
1. Analyse détaillée (vulnérabilités critiques → modérées)
|
||
2. Interdépendances (effets en cascade)
|
||
3. Points de vigilance (indicateurs à surveiller)
|
||
|
||
Interprétation des indices :
|
||
• IHH (Herfindahl-Hirschmann) :
|
||
- Échelle : 0-100
|
||
- Seuils : <25 = Vert (faible), 25-50 = Orange (modéré), >50 = Rouge (élevée)
|
||
• ICS (Substituabilité) :
|
||
- Échelle : 0-1
|
||
- Interprétation : 0 = facilement substituable, 1 = impossible
|
||
• ISG (Stabilité Géopolitique) :
|
||
- Échelle : 0-100
|
||
- Seuils : <40 = Vert, 40-60 = Orange, >60 = Rouge
|
||
• IVC (Concurrence) :
|
||
- Échelle : 0-150
|
||
- Seuils : <10 = Faible, 10-50 = Moyenne, >50 = Forte
|
||
|
||
Contraintes strictes :
|
||
1. N'utiliser aucune connaissance externe. Seul le contenu fourni compte.
|
||
2. Ne pas recommander de diversification minière, industrielle ou politique.
|
||
3. Concentrez-vous sur les impacts sur les produits numériques.
|
||
|
||
Votre mission :
|
||
- Interpréter les indices (IHH, ICS, ISG, IVC)
|
||
- Croiser et hiérarchiser les vulnérabilités
|
||
- Distinguer les horizons temporels
|
||
- Identifier les chaînes de vulnérabilités interdépendantes
|
||
- Identifier des indicateurs d'alerte pertinents
|
||
|
||
Style attendu :
|
||
- Narratif, fluide, impactant, structuré
|
||
- Paragraphes complets + encadrés synthétiques avec données chiffrées
|
||
|
||
Commencez par lire attentivement les indices et les fiches dans la section "Éléments factuels", puis produisez vos sections d'analyse."""
|
||
},
|
||
{
|
||
"title": "Synthèse et Conclusion",
|
||
"prompt": """Vous êtes un assistant stratégique expert chargé de produire des rapports destinés à des décideurs de haut niveau (COMEX, directions risques, stratèges politiques ou industriels).
|
||
|
||
Sur la base de l'analyse précédente, rédigez les deux sections finales du rapport :
|
||
1. Synthèse (narration + hiérarchie vulnérabilités)
|
||
2. Conclusion (scénarios d'impact)
|
||
|
||
Pour la Synthèse :
|
||
- Résumez de façon percutante les vulnérabilités identifiées dans la chaîne de valeur numérique
|
||
- Hiérarchisez les risques en fonction du croisement des indices (IHH, ISG, ICS, IVC)
|
||
- Orientez cette synthèse vers la prise de décision pour différents acteurs économiques
|
||
- Utilisez un ton direct, factuel et orienté décision
|
||
|
||
Pour la Conclusion :
|
||
- Proposez 3-4 scénarios d'impact précis avec déclencheur, horizon temporel, gravité et conséquences
|
||
- Basez chaque scénario sur les données factuelles de l'analyse
|
||
- Montrez comment les différents indices interagissent dans chaque scénario
|
||
- Concentrez-vous particulièrement sur les applications numériques identifiées
|
||
|
||
Interprétation des indices (rappel) :
|
||
• IHH (Herfindahl-Hirschmann) : <25 = Vert, 25-50 = Orange, >50 = Rouge
|
||
• ICS (Substituabilité) : 0 = facilement substituable, 1 = impossible
|
||
• ISG (Stabilité Géopolitique) : <40 = Vert, 40-60 = Orange, >60 = Rouge
|
||
• IVC (Concurrence) : <10 = Faible, 10-50 = Moyenne, >50 = Forte
|
||
|
||
Style attendu :
|
||
- Narratif, percutant, orienté décideur
|
||
- Encadrés synthétiques avec données chiffrées clés
|
||
- Focus sur les implications concrètes pour les entreprises"""
|
||
}
|
||
]
|
||
|
||
def check_api_availability() -> bool:
|
||
"""Vérifie si l'API PrivateGPT est disponible"""
|
||
try:
|
||
response = requests.get(f"{PGPT_URL}/health")
|
||
if response.status_code == 200:
|
||
print("✅ API PrivateGPT disponible")
|
||
return True
|
||
else:
|
||
print(f"❌ L'API PrivateGPT a retourné le code d'état {response.status_code}")
|
||
return False
|
||
except requests.RequestException as e:
|
||
print(f"❌ Erreur de connexion à l'API PrivateGPT: {e}")
|
||
return False
|
||
|
||
def ingest_document(file_path: Path, session_id: str = "") -> bool:
|
||
"""Ingère un document dans PrivateGPT"""
|
||
try:
|
||
with open(file_path, "rb") as f:
|
||
# Si un session_id est fourni, l'ajouter au nom du fichier pour le tracking
|
||
if session_id:
|
||
file_name = f"input_{session_id}_{file_path.name}"
|
||
else:
|
||
file_name = file_path.name
|
||
|
||
files = {"file": (file_name, f, "text/markdown")}
|
||
# Ajouter des métadonnées pour identifier facilement ce fichier d'entrée
|
||
metadata = {
|
||
"type": "input_file",
|
||
"session_id": session_id,
|
||
"document_type": "rapport_analyse_input"
|
||
}
|
||
response = requests.post(
|
||
f"{API_URL}/ingest/file",
|
||
files=files,
|
||
data={"metadata": json.dumps(metadata)} if "metadata" in requests.get(f"{API_URL}/ingest/file").text else None
|
||
)
|
||
response.raise_for_status()
|
||
print(f"✅ Document '{file_path}' ingéré avec succès sous le nom '{file_name}'")
|
||
return True
|
||
except FileNotFoundError:
|
||
print(f"❌ Fichier '{file_path}' introuvable")
|
||
return False
|
||
except requests.RequestException as e:
|
||
print(f"❌ Erreur lors de l'ingestion du document: {e}")
|
||
return False
|
||
|
||
def setup_temp_directory() -> None:
|
||
"""Crée le répertoire temporaire pour les fichiers intermédiaires"""
|
||
if not TEMP_DIR.exists():
|
||
TEMP_DIR.mkdir(parents=True)
|
||
print(f"📁 Répertoire temporaire '{TEMP_DIR}' créé")
|
||
|
||
def save_section_to_file(section: Dict[str, str], index: int, session_uuid: str) -> Path:
|
||
"""Sauvegarde une section dans un fichier temporaire et retourne le chemin"""
|
||
setup_temp_directory()
|
||
section_file = TEMP_DIR / f"temp_section_{session_uuid}_{index+1}_{section['title'].lower().replace(' ', '_')}.md"
|
||
|
||
# Contenu du fichier avec métadonnées et commentaire explicite
|
||
content = (
|
||
f"# SECTION TEMPORAIRE GÉNÉRÉE - {section['title']}\n\n"
|
||
f"Note: Ce document est une section temporaire du rapport d'analyse en cours de génération.\n"
|
||
f"UUID de session: {session_uuid}\n\n"
|
||
f"{section['output']}"
|
||
)
|
||
|
||
# Écrire dans le fichier
|
||
section_file.write_text(content, encoding="utf-8")
|
||
return section_file
|
||
|
||
def ingest_section_files(section_files: List[Path]) -> List[str]:
|
||
"""Ingère les fichiers de section et retourne leurs noms de fichiers"""
|
||
ingested_file_names = []
|
||
for file_path in section_files:
|
||
try:
|
||
with open(file_path, "rb") as f:
|
||
files = {"file": (file_path.name, f, "text/markdown")}
|
||
# Ajouter des métadonnées pour identifier facilement nos fichiers temporaires
|
||
metadata = {
|
||
"type": "temp_section",
|
||
"document_type": "rapport_analyse_section"
|
||
}
|
||
response = requests.post(
|
||
f"{API_URL}/ingest/file",
|
||
files=files,
|
||
data={"metadata": json.dumps(metadata)} if "metadata" in requests.get(f"{API_URL}/ingest/file").text else None
|
||
)
|
||
response.raise_for_status()
|
||
# Ajouter le nom du fichier à la liste pour pouvoir le retrouver et le supprimer plus tard
|
||
ingested_file_names.append(file_path.name)
|
||
print(f"✅ Section '{file_path.name}' ingérée")
|
||
except Exception as e:
|
||
print(f"⚠️ Erreur lors de l'ingestion de '{file_path.name}': {e}")
|
||
return ingested_file_names
|
||
|
||
def get_context(sections: List[Dict[str, str]], strategy: ContextStrategy, max_length: int) -> str:
|
||
"""Génère le contexte selon la stratégie choisie"""
|
||
if not sections or strategy == ContextStrategy.NONE:
|
||
return ""
|
||
|
||
# Stratégie: utiliser des fichiers ingérés
|
||
if strategy == ContextStrategy.FILE:
|
||
# Cette stratégie n'utilise pas de contexte dans le prompt
|
||
# Elle repose sur les fichiers ingérés et le paramètre use_context=True
|
||
section_names = [s["title"] for s in sections]
|
||
context_note = f"NOTE IMPORTANTE: Les sections précédentes ({', '.join(section_names)}) " + \
|
||
f"ont été ingérées sous forme de fichiers temporaires avec l'identifiant unique '{session_uuid}'. " + \
|
||
f"Utilisez UNIQUEMENT le document '{input_file.name}' et ces sections temporaires pour votre analyse. " + \
|
||
f"IGNOREZ tous les autres documents qui pourraient être présents dans la base de connaissances. " + \
|
||
f"Assurez la cohérence avec les sections déjà générées pour maintenir la continuité du rapport."
|
||
print(f"📄 Utilisation de {len(sections)} sections ingérées comme contexte")
|
||
return context_note
|
||
|
||
# Stratégie: uniquement la dernière section
|
||
if strategy == ContextStrategy.LATEST:
|
||
latest = sections[-1]
|
||
context = f"# {latest['title']}\n{latest['output']}"
|
||
if len(context) > max_length:
|
||
context = context[:max_length] + "..."
|
||
print(f"📄 Contexte basé sur la dernière section: {latest['title']} ({len(context)} caractères)")
|
||
return context
|
||
|
||
# Stratégie: tronquer toutes les sections
|
||
if strategy == ContextStrategy.TRUNCATE:
|
||
full_context = "\n\n".join([f"# {s['title']}\n{s['output']}" for s in sections])
|
||
if len(full_context) > max_length:
|
||
truncated = full_context[:max_length] + "..."
|
||
print(f"✂️ Contexte tronqué à {max_length} caractères")
|
||
return truncated
|
||
return full_context
|
||
|
||
# Stratégie: résumer intelligemment
|
||
if strategy == ContextStrategy.SUMMARIZE:
|
||
try:
|
||
# Construire un prompt pour demander un résumé des sections précédentes
|
||
sections_text = "\n\n".join([f"# {s['title']}\n{s['output']}" for s in sections])
|
||
|
||
# Limiter la taille du texte à résumer si nécessaire
|
||
if len(sections_text) > max_length * 2:
|
||
sections_text = sections_text[:max_length * 2] + "..."
|
||
|
||
summary_prompt = {
|
||
"messages": [
|
||
{"role": "system", "content": "Vous êtes un assistant spécialisé dans le résumé concis. Créez un résumé très court mais informatif qui conserve les points clés."},
|
||
{"role": "user", "content": f"Résumez les sections suivantes en {max_length // 10} mots maximum, en conservant les idées principales et les points essentiels:\n\n{sections_text}"}
|
||
],
|
||
"temperature": 0.1,
|
||
"stream": False
|
||
}
|
||
|
||
# Envoyer la requête pour obtenir un résumé
|
||
response = requests.post(
|
||
f"{API_URL}/chat/completions",
|
||
json=summary_prompt,
|
||
headers={"accept": "application/json"}
|
||
)
|
||
response.raise_for_status()
|
||
|
||
# Extraire et retourner le résumé
|
||
result = response.json()
|
||
summary = result["choices"][0]["message"]["content"]
|
||
print(f"✅ Résumé de contexte généré ({len(summary)} caractères)")
|
||
return summary
|
||
|
||
except Exception as e:
|
||
print(f"⚠️ Impossible de générer un résumé du contexte: {e}")
|
||
# En cas d'échec, revenir à la stratégie de troncature
|
||
return get_context(sections, ContextStrategy.TRUNCATE, max_length)
|
||
|
||
def generate_text(prompt: str, previous_context: str = "", use_context: bool = True, retry_on_error: bool = True) -> Optional[str]:
|
||
"""Génère du texte avec l'API PrivateGPT"""
|
||
try:
|
||
# Préparer le prompt avec le contexte précédent si disponible et demandé
|
||
full_prompt = prompt
|
||
if previous_context and use_context:
|
||
full_prompt += "\n\nContexte précédent (informations des sections déjà générées) :\n" + previous_context
|
||
|
||
# Définir les paramètres de la requête
|
||
system_message = "Vous êtes un assistant stratégique expert chargé de produire des rapports destinés à des décideurs de haut niveau (COMEX, directions risques, stratèges politiques ou industriels). " + \
|
||
f"Objectif : Analyser les vulnérabilités systémiques dans une chaîne de valeur numérique selon les données du fichier {input_file.name}. " + \
|
||
"Interprétation des indices : " + \
|
||
"• IHH (Herfindahl-Hirschmann) : Échelle 0-100, Seuils : <25 = Vert (faible), 25-50 = Orange (modéré), >50 = Rouge (élevée) " + \
|
||
"• ISG (Stabilité Géopolitique) : Échelle 0-100, Seuils : <40 = Vert (stable), 40-60 = Orange, >60 = Rouge (instable) " + \
|
||
"• ICS (Substituabilité) : Échelle 0-1, 0 = facilement substituable, 1 = impossible " + \
|
||
"• IVC (Concurrence) : Échelle 0-150, Seuils : <10 = Faible, 10-50 = Moyenne, >50 = Forte " + \
|
||
"Style attendu : Narratif, fluide, impactant, structuré avec paragraphes complets et données chiffrées. " + \
|
||
f"Concentrez-vous UNIQUEMENT sur le document principal '{input_file.name}' et sur les sections temporaires préfixées par 'temp_section_{session_uuid}_'. " + \
|
||
"IGNOREZ absolument tous les autres documents qui pourraient être présents dans la base de connaissances. " + \
|
||
"Les sections précédentes ont été générées et ingérées comme documents temporaires - " + \
|
||
"assurez la cohérence avec leur contenu pour maintenir la continuité du rapport."
|
||
|
||
# Définir les paramètres de la requête
|
||
payload = {
|
||
"messages": [
|
||
{"role": "system", "content": system_message},
|
||
{"role": "user", "content": full_prompt}
|
||
],
|
||
"use_context": True, # Active la recherche RAG dans les documents ingérés
|
||
"temperature": 0.2, # Température réduite pour plus de cohérence
|
||
"stream": False
|
||
}
|
||
|
||
# Tenter d'ajouter un filtre de contexte (fonctionnalité expérimentale qui peut ne pas être supportée)
|
||
try:
|
||
# Vérifier si le filtre de contexte est supporté sans faire de requête supplémentaire
|
||
filter_metadata = {
|
||
"document_name": [input_file.name] + [f.name for f in section_files]
|
||
}
|
||
payload["filter_metadata"] = filter_metadata
|
||
except Exception as e:
|
||
print(f"ℹ️ Remarque: Impossible d'appliquer le filtre de contexte: {e}")
|
||
|
||
# Envoyer la requête
|
||
response = requests.post(
|
||
f"{API_URL}/chat/completions",
|
||
json=payload,
|
||
headers={"accept": "application/json"}
|
||
)
|
||
response.raise_for_status()
|
||
|
||
# Extraire la réponse générée
|
||
result = response.json()
|
||
if "choices" in result and len(result["choices"]) > 0:
|
||
return result["choices"][0]["message"]["content"]
|
||
else:
|
||
print("⚠️ Format de réponse inattendu:", json.dumps(result, indent=2))
|
||
return None
|
||
|
||
except requests.RequestException as e:
|
||
print(f"❌ Erreur lors de la génération de texte: {e}")
|
||
if hasattr(e, 'response') and e.response is not None:
|
||
print(f"Détails: {e.response.text}")
|
||
return None
|
||
|
||
def cleanup_temp_files(temp_file_names: List[str] = None, remove_directory: bool = False, session_id: str = "") -> None:
|
||
"""Nettoie les fichiers temporaires et les documents ingérés"""
|
||
try:
|
||
# Supprimer les fichiers du répertoire temporaire
|
||
if TEMP_DIR.exists():
|
||
for temp_file in TEMP_DIR.glob("*.md"):
|
||
temp_file.unlink()
|
||
print(f"🗑️ Fichier temporaire supprimé : {temp_file.name}")
|
||
|
||
# Supprimer le répertoire s'il est vide et si demandé
|
||
if remove_directory and not any(TEMP_DIR.iterdir()):
|
||
TEMP_DIR.rmdir()
|
||
print(f"🗑️ Répertoire temporaire '{TEMP_DIR}' supprimé")
|
||
|
||
# Supprimer les documents ingérés via l'API de liste et suppression
|
||
try:
|
||
# Lister tous les documents ingérés
|
||
list_response = requests.get(f"{API_URL}/ingest/list")
|
||
if list_response.status_code == 200:
|
||
documents_data = list_response.json()
|
||
|
||
# Format de réponse OpenAI
|
||
if "data" in documents_data:
|
||
documents = documents_data.get("data", [])
|
||
# Format alternatif
|
||
else:
|
||
documents = documents_data.get("documents", [])
|
||
|
||
deleted_count = 0
|
||
# Parcourir les documents et supprimer ceux qui correspondent à nos fichiers temporaires
|
||
for doc in documents:
|
||
doc_metadata = doc.get("doc_metadata", {})
|
||
file_name = doc_metadata.get("file_name", "") or doc_metadata.get("filename", "")
|
||
|
||
# Vérifier si c'est un de nos fichiers temporaires ou le fichier d'entrée
|
||
is_our_file = False
|
||
if temp_file_names and file_name in temp_file_names:
|
||
is_our_file = True
|
||
elif f"temp_section_{session_uuid}_" in file_name:
|
||
is_our_file = True
|
||
elif session_id and f"input_{session_id}_" in file_name:
|
||
is_our_file = True
|
||
|
||
if is_our_file:
|
||
doc_id = doc.get("doc_id") or doc.get("id")
|
||
if doc_id:
|
||
delete_response = requests.delete(f"{API_URL}/ingest/{doc_id}")
|
||
if delete_response.status_code == 200:
|
||
deleted_count += 1
|
||
|
||
if deleted_count > 0:
|
||
print(f"🗑️ {deleted_count} documents supprimés de PrivateGPT")
|
||
|
||
except Exception as e:
|
||
print(f"⚠️ Erreur lors de la suppression des documents ingérés: {e}")
|
||
|
||
except Exception as e:
|
||
print(f"⚠️ Erreur lors du nettoyage des fichiers temporaires: {e}")
|
||
|
||
def clean_ai_thoughts(text: str) -> str:
|
||
"""Nettoie le texte généré en supprimant les 'pensées' de l'IA entre balises <think></think>"""
|
||
import re
|
||
|
||
# Supprimer tout le contenu entre les balises <think> et </think>, y compris les balises
|
||
text = re.sub(r'<think>.*?</think>', '', text, flags=re.DOTALL)
|
||
|
||
# Supprimer les lignes vides multiples
|
||
text = re.sub(r'\n{3,}', '\n\n', text)
|
||
|
||
return text
|
||
|
||
def main(input_path: str, output_path: str, context_strategy: ContextStrategy = ContextStrategy.FILE, context_length: int = MAX_CONTEXT_LENGTH):
|
||
"""Fonction principale qui exécute le processus complet"""
|
||
global input_file, section_files, session_uuid # Variables globales pour le filtre de contexte et l'UUID
|
||
|
||
# Générer un UUID unique pour cette session
|
||
session_uuid = str(uuid.uuid4())[:8] # Utiliser les 8 premiers caractères pour plus de concision
|
||
print(f"🔑 UUID de session généré: {session_uuid}")
|
||
|
||
# Vérifier la disponibilité de l'API
|
||
if not check_api_availability():
|
||
sys.exit(1)
|
||
|
||
# Convertir les chemins en objets Path (accessibles globalement)
|
||
input_file = Path(input_path)
|
||
output_file = Path(output_path)
|
||
|
||
# Ingérer le document principal avec l'UUID de session
|
||
if not ingest_document(input_file, session_uuid):
|
||
sys.exit(1)
|
||
|
||
# Récupérer la valeur du délai depuis args
|
||
delay = args.delay if 'args' in globals() else 5
|
||
|
||
# Attendre que l'ingestion soit complètement traitée
|
||
print(f"⏳ Attente du traitement de l'ingestion pendant {delay} secondes...")
|
||
time.sleep(delay)
|
||
|
||
print(f"🔧 Stratégie de contexte initiale: {context_strategy.value}, taille max: {context_length} caractères")
|
||
|
||
# Préparer le répertoire pour les fichiers temporaires
|
||
setup_temp_directory()
|
||
|
||
# Générer chaque section du rapport
|
||
step_outputs = []
|
||
section_files = [] # Chemins des fichiers temporaires
|
||
ingested_file_names = [] # Noms des fichiers ingérés
|
||
|
||
# Liste des stratégies de fallback dans l'ordre
|
||
fallback_strategies = [
|
||
ContextStrategy.FILE,
|
||
ContextStrategy.TRUNCATE,
|
||
ContextStrategy.SUMMARIZE,
|
||
ContextStrategy.LATEST,
|
||
ContextStrategy.NONE
|
||
]
|
||
|
||
for i, step in enumerate(steps):
|
||
print(f"\n🚧 Étape {i+1}/{len(steps)}: {step['title']}")
|
||
|
||
# Si nous avons des sections précédentes et utilisons la stratégie FILE
|
||
if step_outputs and context_strategy == ContextStrategy.FILE:
|
||
# Sauvegarder et ingérer toutes les sections précédentes qui ne l'ont pas encore été
|
||
for j, section in enumerate(step_outputs):
|
||
if j >= len(section_files): # Cette section n'a pas encore été sauvegardée
|
||
section_file = save_section_to_file(section, j, session_uuid)
|
||
section_files.append(section_file)
|
||
|
||
# Ingérer le fichier si nous utilisons la stratégie FILE
|
||
new_ids = ingest_section_files([section_file])
|
||
ingested_section_ids.extend(new_ids)
|
||
|
||
# Attendre que l'ingestion soit traitée
|
||
print(f"⏳ Attente du traitement de l'ingestion des sections précédentes...")
|
||
time.sleep(2)
|
||
|
||
# Essayer chaque stratégie jusqu'à ce qu'une réussisse
|
||
output = None
|
||
current_strategy_index = fallback_strategies.index(context_strategy)
|
||
|
||
while output is None and current_strategy_index < len(fallback_strategies):
|
||
current_strategy = fallback_strategies[current_strategy_index]
|
||
print(f"🔄 Tentative avec la stratégie: {current_strategy.value}")
|
||
|
||
# Préparer le contexte selon la stratégie actuelle
|
||
previous = get_context(step_outputs, current_strategy, context_length) if step_outputs else ""
|
||
|
||
# Générer le texte
|
||
try:
|
||
output = generate_text(step["prompt"], previous, current_strategy != ContextStrategy.NONE)
|
||
if output is None:
|
||
# Passer à la stratégie suivante
|
||
current_strategy_index += 1
|
||
print(f"⚠️ Échec avec la stratégie {current_strategy.value}, passage à la suivante")
|
||
time.sleep(1) # Pause avant nouvelle tentative
|
||
except Exception as e:
|
||
print(f"⚠️ Erreur lors de la génération avec {current_strategy.value}: {e}")
|
||
current_strategy_index += 1
|
||
time.sleep(1) # Pause avant nouvelle tentative
|
||
|
||
if output:
|
||
step_outputs.append({"title": step["title"], "output": output})
|
||
print(f"✅ Section '{step['title']}' générée ({len(output)} caractères)")
|
||
|
||
# Si nous sommes en mode FILE, sauvegarder immédiatement cette section
|
||
if context_strategy == ContextStrategy.FILE:
|
||
section_file = save_section_to_file(step_outputs[-1], len(step_outputs)-1, session_uuid)
|
||
section_files.append(section_file)
|
||
|
||
# Ingérer le fichier
|
||
new_file_names = ingest_section_files([section_file])
|
||
ingested_file_names.extend(new_file_names)
|
||
|
||
# Petite pause pour permettre l'indexation
|
||
time.sleep(1)
|
||
else:
|
||
print(f"❌ Échec de la génération pour '{step['title']}' après toutes les stratégies")
|
||
|
||
# Vérifier si nous avons généré toutes les sections
|
||
if len(step_outputs) != len(steps):
|
||
print(f"⚠️ Attention: Seules {len(step_outputs)}/{len(steps)} sections ont été générées")
|
||
|
||
# Écrire le rapport final dans l'ordre souhaité (Synthèse en premier)
|
||
# Réordonner les sections pour que la synthèse soit en premier
|
||
ordered_outputs = sorted(step_outputs, key=lambda x: 0 if x["title"] == "Synthèse" else
|
||
1 if x["title"] == "Analyse détaillée" else
|
||
2 if x["title"] == "Interdépendances" else
|
||
3 if x["title"] == "Points de vigilance" else 4)
|
||
|
||
# Assembler le rapport
|
||
report_sections = []
|
||
for s in ordered_outputs:
|
||
# Nettoyer les "pensées" de l'IA dans chaque section
|
||
cleaned_output = clean_ai_thoughts(s['output'])
|
||
report_sections.append(f"## {s['title']}\n\n{cleaned_output}")
|
||
|
||
report_text = "# Analyse des vulnérabilités de la chaine de fabrication du numériquee\n\n".join(report_sections)
|
||
|
||
# Écrire le rapport dans un fichier
|
||
try:
|
||
output_file.write_text(report_text, encoding="utf-8")
|
||
print(f"\n📄 Rapport final généré dans '{output_file}'")
|
||
except IOError as e:
|
||
print(f"❌ Erreur lors de l'écriture du fichier de sortie: {e}")
|
||
|
||
# Nettoyer les fichiers temporaires si demandé
|
||
if args.clean_temp:
|
||
print("\n🧹 Nettoyage des fichiers temporaires et documents ingérés...")
|
||
cleanup_temp_files(ingested_file_names, args.remove_temp_dir, session_uuid)
|
||
|
||
if __name__ == "__main__":
|
||
# Définir les arguments en ligne de commande
|
||
parser = argparse.ArgumentParser(description="Générer un rapport d'analyse structuré avec PrivateGPT")
|
||
parser.add_argument("-i", "--input", type=str, default=DEFAULT_INPUT_PATH,
|
||
help=f"Chemin du fichier d'entrée (défaut: {DEFAULT_INPUT_PATH})")
|
||
parser.add_argument("-o", "--output", type=str, default=DEFAULT_OUTPUT_PATH,
|
||
help=f"Chemin du fichier de sortie (défaut: {DEFAULT_OUTPUT_PATH})")
|
||
parser.add_argument("--context", type=str, choices=[s.value for s in ContextStrategy], default=ContextStrategy.FILE.value,
|
||
help=f"Stratégie de gestion du contexte (défaut: {ContextStrategy.FILE.value})")
|
||
parser.add_argument("--context-length", type=int, default=MAX_CONTEXT_LENGTH,
|
||
help=f"Taille maximale du contexte en caractères (défaut: {MAX_CONTEXT_LENGTH})")
|
||
parser.add_argument("--delay", type=int, default=5,
|
||
help="Délai d'attente en secondes après l'ingestion (défaut: 5)")
|
||
parser.add_argument("--clean-temp", action="store_true", default=True,
|
||
help="Nettoyer les fichiers temporaires et documents ingérés par le script (défaut: True)")
|
||
parser.add_argument("--keep-temp", action="store_true",
|
||
help="Conserver les fichiers temporaires (remplace --clean-temp)")
|
||
parser.add_argument("--remove-temp-dir", action="store_true",
|
||
help="Supprimer le répertoire temporaire après exécution (défaut: False)")
|
||
args = parser.parse_args()
|
||
|
||
# Gérer les options contradictoires
|
||
if args.keep_temp:
|
||
args.clean_temp = False
|
||
|
||
# Exécuter le programme principal avec les options configurées
|
||
context_strategy = ContextStrategy(args.context)
|
||
main(args.input, args.output, context_strategy, args.context_length)
|