import os import json from datetime import datetime from typing import Dict, Any, Optional, Union, List def determiner_repertoire_ticket(ticket_id: str) -> Optional[str]: """ Détermine dynamiquement le répertoire du ticket. Args: ticket_id: str, le code du ticket Returns: str, le chemin du répertoire pour ce ticket ou None si non trouvé """ # Base de recherche des tickets output_dir = "output" # Normaliser le ticket_id (retirer les préfixes "ticket_" éventuels) if ticket_id.startswith("ticket_"): ticket_id = ticket_id[7:] # Retire "ticket_" # Si "UNKNOWN", chercher les tickets disponibles et utiliser T11143 comme fallback if ticket_id == "UNKNOWN": print(f"ID de ticket 'UNKNOWN' reçu, recherche de tickets disponibles ou utilisation de la valeur par défaut") # Vérifier si T11143 existe (cas spécifique testé) test_path = os.path.join(output_dir, f"ticket_T11143") if os.path.exists(test_path): print(f"Utilisation du ticket par défaut: T11143") ticket_id = "T11143" else: # Sinon chercher le premier ticket disponible tickets = [d[7:] for d in os.listdir(output_dir) if os.path.isdir(os.path.join(output_dir, d)) and d.startswith("ticket_T")] if tickets: ticket_id = tickets[0] print(f"Utilisation du premier ticket disponible: {ticket_id}") else: print("Aucun ticket trouvé dans le répertoire output/") return None # Format attendu du répertoire de ticket ticket_dir = f"ticket_{ticket_id}" ticket_path = os.path.join(output_dir, ticket_dir) if not os.path.exists(ticket_path): print(f"Répertoire de ticket non trouvé: {ticket_path}") # Essayer de trouver un répertoire avec un nom similaire tickets = [d for d in os.listdir(output_dir) if os.path.isdir(os.path.join(output_dir, d)) and d.startswith("ticket_")] closest_match = None for t in tickets: if ticket_id in t: closest_match = t break if closest_match: ticket_path = os.path.join(output_dir, closest_match) print(f"Utilisation du répertoire alternatif trouvé: {ticket_path}") else: print(f"Aucun répertoire alternatif trouvé pour le ticket {ticket_id}") return None # Trouver la dernière extraction (par date) extractions = [] for extraction in os.listdir(ticket_path): extraction_path = os.path.join(ticket_path, extraction) if os.path.isdir(extraction_path) and extraction.startswith(ticket_id): extractions.append(extraction_path) # Si pas d'extraction avec le format exact, essayer des formats similaires if not extractions: for extraction in os.listdir(ticket_path): extraction_path = os.path.join(ticket_path, extraction) if os.path.isdir(extraction_path): extractions.append(extraction_path) if not extractions: print(f"Aucune extraction trouvée pour le ticket {ticket_id} dans {ticket_path}") return None # Trier par date de modification (plus récente en premier) extractions.sort(key=lambda x: os.path.getmtime(x), reverse=True) # Retourner le chemin de la dernière extraction print(f"Répertoire d'extraction trouvé: {extractions[0]}") return extractions[0] def extraire_ticket_id(data: Dict[str, Any]) -> Optional[str]: """ Extrait l'ID du ticket à partir des métadonnées ou du chemin de l'image. Args: data: dict, données contenant potentiellement des métadonnées avec l'ID du ticket Returns: str, l'ID du ticket extrait ou None si non trouvé """ if not data: return None # Essayer d'extraire le ticket_id des métadonnées metadata = data.get("metadata", {}) image_path = metadata.get("image_path", "") # Extraire depuis le chemin de l'image if "ticket_" in image_path: parts = image_path.split("ticket_") if len(parts) > 1: ticket_parts = parts[1].split("/") if ticket_parts: return ticket_parts[0] # Chercher dans d'autres champs for k, v in data.items(): if isinstance(v, str) and "ticket_" in v.lower(): parts = v.lower().split("ticket_") if len(parts) > 1: ticket_parts = parts[1].split("/") if ticket_parts and ticket_parts[0].startswith("t"): return ticket_parts[0].upper() return None def generer_version_texte(data: Union[Dict[str, Any], list], ticket_id: str, step_name: str, file_path: str) -> None: """ Génère une version texte lisible du fichier JSON pour faciliter la revue humaine. Args: data: Données à convertir en texte lisible ticket_id: ID du ticket step_name: Nom de l'étape ou de l'agent file_path: Chemin du fichier JSON """ try: # Créer le chemin du fichier texte txt_path = file_path.replace('.json', '.txt') with open(txt_path, 'w', encoding='utf-8') as f: f.write(f"RÉSULTATS DE L'ANALYSE {step_name.upper()} - TICKET {ticket_id}\n") f.write("="*80 + "\n\n") # Si c'est une liste, traiter chaque élément if isinstance(data, list): for i, item in enumerate(data, 1): f.write(f"--- ÉLÉMENT {i} ---\n\n") # Pour les résultats OCR if isinstance(item, dict): if "extracted_text" in item: f.write(f"Image: {item.get('image_name', 'N/A')}\n") f.write(f"Texte extrait:\n{item['extracted_text']}\n\n") # Priorité 1: Champ "response" (pour les rapports) elif "response" in item: f.write(f"{item['response']}\n\n") # Priorité 2: Champ "analyse" (pour les analyses d'images) elif "analyse" in item and isinstance(item["analyse"], str): f.write(f"{item['analyse']}\n\n") # Priorité 2b: Champ "analyse" avec langues elif "analyse" in item and isinstance(item["analyse"], dict): if "fr" in item["analyse"] and item["analyse"]["fr"]: f.write(f"{item['analyse']['fr']}\n\n") elif "en" in item["analyse"] and item["analyse"]["en"]: f.write(f"{item['analyse']['en']}\n\n") # Priorité 3: Champ "raw_response" (pour les analyses brutes) elif "raw_response" in item: f.write(f"{item['raw_response']}\n\n") # Priorité 4: Structure imbriquée d'analyse elif "analysis" in item: if isinstance(item["analysis"], dict): if "analyse" in item["analysis"]: f.write(f"{item['analysis']['analyse']}\n\n") elif "raw_response" in item["analysis"]: f.write(f"{item['analysis']['raw_response']}\n\n") # Priorité 5: Pour le tri d'images elif "is_relevant" in item: f.write(f"Image pertinente: {item['is_relevant']}\n") if "reason" in item: f.write(f"Raison: {item['reason']}\n\n") else: f.write("Aucun contenu d'analyse trouvé.\n\n") else: f.write(str(item) + "\n\n") f.write("-"*40 + "\n\n") # Si c'est un dictionnaire unique elif isinstance(data, dict): if "extracted_text" in data: f.write(f"Image: {data.get('image_name', 'N/A')}\n") f.write(f"Texte extrait:\n{data['extracted_text']}\n\n") elif "response" in data: f.write(f"{data['response']}\n\n") elif "analyse" in data and isinstance(data["analyse"], str): f.write(f"{data['analyse']}\n\n") elif "raw_response" in data: f.write(f"{data['raw_response']}\n\n") else: for key, value in data.items(): if key != "prompt" and not isinstance(value, dict): f.write(f"{key}: {value}\n") f.write("\n" + "="*80 + "\n") f.write(f"Fichier original: {os.path.basename(file_path)}") print(f"Version texte générée dans {txt_path}") except Exception as e: print(f"Erreur lors de la génération de la version texte: {e}") def sauvegarder_donnees(ticket_id: Optional[str] = None, step_name: str = "", data: Optional[Union[Dict[str, Any], List[Dict[str, Any]]]] = None, base_dir: Optional[str] = None, is_resultat: bool = False) -> None: """ Sauvegarde des données sous forme de fichier JSON. Args: ticket_id: str, le code du ticket (optionnel, sera extrait automatiquement si None) step_name: str, le nom de l'étape ou de l'agent data: dict ou liste, les données à sauvegarder base_dir: str, le répertoire de base pour les fichiers de logs (optionnel) is_resultat: bool, indique si les données sont des résultats d'agent """ if data is None: print("Aucune donnée à sauvegarder") return # Convertir les données en liste si ce n'est pas déjà le cas data_list = data if isinstance(data, list) else [data] # Si ticket_id n'est pas fourni, essayer de l'extraire des métadonnées du premier élément if not ticket_id and data_list: first_item = data_list[0] ticket_id = extraire_ticket_id(first_item) if ticket_id: print(f"Ticket ID extrait: {ticket_id}") # Si on n'a toujours pas de ticket_id, on ne peut pas continuer if not ticket_id: print("Impossible de déterminer l'ID du ticket, sauvegarde impossible") return # Si base_dir n'est pas fourni ou est un répertoire générique comme "reports", le déterminer automatiquement if base_dir is None or base_dir == "reports": extraction_dir = determiner_repertoire_ticket(ticket_id) if not extraction_dir: print(f"Impossible de déterminer le répertoire pour le ticket {ticket_id}") return # Utiliser le répertoire rapports pour stocker les résultats rapports_dir = os.path.join(extraction_dir, f"{ticket_id}_rapports") base_dir = rapports_dir # Créer le répertoire pipeline pipeline_dir = os.path.join(base_dir, "pipeline") os.makedirs(pipeline_dir, exist_ok=True) def extraire_nom_modele(nom_fichier, etape_name): """ Extrait le nom du modèle à partir d'un nom de fichier. Args: nom_fichier: Nom du fichier etape_name: Nom de l'étape ou de l'agent Returns: Nom du modèle extrait """ if not nom_fichier.startswith(f"{etape_name}_") or not nom_fichier.endswith("_results.json"): return None # Extraire la partie centrale (modèle) modele = nom_fichier.replace(f"{etape_name}_", "").replace("_results.json", "") return modele # Fonction pour normaliser le nom du modèle et éviter les variations syntaxiques def normaliser_nom_modele(nom_modele): if not nom_modele: print("Attention: nom de modèle vide, utilisation de llama3-vision-90b-instruct") return "llama3-vision-90b-instruct" # Normaliser les chaînes de caractères communes nom_normalise = nom_modele.lower() # Supprimer les préfixes "image_" redondants if nom_normalise.startswith("image_"): nom_normalise = nom_normalise[6:] # Normaliser les noms de modèles connus pour éviter les doublons conversions = { "llama3.2-vision:90b-instruct-q8_0": "llama3-vision-90b-instruct", "llama3_2-vision:90b-instruct-q8_0": "llama3-vision-90b-instruct", "llama3.2-vision:90b": "llama3-vision-90b-instruct", "llama3_2-vision_90b-instruct-q8_0": "llama3-vision-90b-instruct", "llama3.2-vision_90b-instruct-q8_0": "llama3-vision-90b-instruct", "llama3-2-vision-90b-instruct-q8-0": "llama3-vision-90b-instruct", "llama-vision": "llama3-vision-90b-instruct", "mistral-large": "mistral-large-latest", "pixtral-large": "pixtral-large-latest", "unknown": "llama3-vision-90b-instruct", "unknown_model": "llama3-vision-90b-instruct", "inconnu": "llama3-vision-90b-instruct" } # Appliquer les conversions exactes if nom_normalise in conversions: nom_normalise = conversions[nom_normalise] print(f"Nom de modèle normalisé via conversion exacte: {nom_normalise}") else: # Appliquer les conversions partielles for original, nouveau in conversions.items(): if original in nom_normalise: nom_normalise = nouveau print(f"Nom de modèle normalisé via conversion partielle: {nom_normalise}") break # Nettoyer les caractères spéciaux pour le nom de fichier # On remplace tous les caractères problématiques par des tirets safe_nom = "" for c in nom_normalise: if c.isalnum() or c == '-': safe_nom += c else: safe_nom += '-' # Si après toutes ces conversions, on a toujours "unknown_model", utiliser la valeur par défaut if safe_nom in ["unknown-model", "unknown", "inconnu"]: print(f"Nom de modèle {safe_nom} remplacé par la valeur par défaut") safe_nom = "llama3-vision-90b-instruct" return safe_nom # Nom du fichier if is_resultat and data_list: # Extraire le nom du modèle LLM des métadonnées du premier élément first_item = data_list[0] # Vérifier d'abord s'il existe un champ model_info directement dans l'élément if "model_info" in first_item and "model" in first_item["model_info"]: llm_name = first_item["model_info"]["model"] print(f"Nom du modèle trouvé dans model_info: {llm_name}") else: # Sinon chercher dans les métadonnées llm_name = first_item.get("metadata", {}).get("model_info", {}).get("model", "") print(f"Nom du modèle trouvé dans metadata: {llm_name}") # Si le modèle est vide ou 'unknown', vérifier s'il existe déjà un fichier pour cette étape # afin d'éviter de créer des doublons avec "unknown_model" if not llm_name or llm_name.lower() == "unknown": print(f"Le nom du modèle est vide ou 'unknown', recherche d'un fichier existant pour {step_name}") # Rechercher un fichier existant pour la même étape existing_files = [f for f in os.listdir(pipeline_dir) if f.startswith(f"{step_name}_") and f.endswith("_results.json")] if existing_files: # Utiliser le même nom de modèle que le fichier existant for file in existing_files: if "unknown_model" not in file: # Extraire le nom du modèle du fichier existant parts = file.split('_') if len(parts) > 2: # Reconstruire le nom du modèle (tout ce qui est entre step_name_ et _results.json) model_part = '_'.join(parts[1:-1]) llm_name = model_part print(f"Utilisation du nom de modèle existant: {llm_name}") break # Normaliser le nom du modèle safe_llm_name = normaliser_nom_modele(llm_name) print(f"Nom du modèle après normalisation: {safe_llm_name}") file_name = f"{step_name}_{safe_llm_name}_results.json" else: file_name = f"{step_name}.json" file_path = os.path.join(pipeline_dir, file_name) try: # Vérifier s'il existe déjà des fichiers similaires pour cette étape et ce modèle # Cela permet d'éviter les doublons dus à des légères variations dans les noms de modèles similar_files = [] normalized_model_files = {} # Première passe: normaliser les noms de tous les fichiers existants for existing_file in os.listdir(pipeline_dir): if existing_file.startswith(f"{step_name}_") and existing_file.endswith("_results.json"): # Extraire le nom du modèle model_part = extraire_nom_modele(existing_file, step_name) if model_part: normalized_model = normaliser_nom_modele(model_part) normalized_model_files[existing_file] = normalized_model # Normaliser le modèle actuel pour comparaison safe_model = normaliser_nom_modele(llm_name) # Deuxième passe: trouver les fichiers avec un modèle normalisé identique for existing_file, normalized_model in normalized_model_files.items(): if normalized_model == safe_model: similar_files.append(existing_file) print(f"Fichier similaire trouvé: {existing_file} (modèle normalisé: {normalized_model})") # Si des fichiers similaires existent, utiliser le premier if similar_files and is_resultat: file_name = similar_files[0] file_path = os.path.join(pipeline_dir, file_name) print(f"Utilisation du fichier existant: {file_name}") # Vérifier si le fichier existe déjà if os.path.exists(file_path): print(f"Le fichier {file_path} existe déjà, mise à jour des données") # Charger les données existantes try: with open(file_path, "r", encoding="utf-8") as f: file_content = f.read().strip() existing_data = json.loads(file_content) if file_content else [] # Si les données existantes ne sont pas une liste, les convertir if not isinstance(existing_data, list): existing_data = [existing_data] except Exception as e: print(f"Erreur lors de la lecture des données existantes: {e}") existing_data = [] else: existing_data = [] # Fusionner les nouvelles données avec les données existantes combined_data = existing_data.copy() # Pour chaque nouvel élément, vérifier s'il existe déjà un élément similaire for new_item in data_list: # Vérifier si un élément similaire existe déjà is_duplicate = False # Pour la déduplication, on compare certains champs clés if isinstance(new_item, dict): for idx, existing_item in enumerate(existing_data): if isinstance(existing_item, dict): # Comparer les métadonnées pour détecter les doublons new_meta = new_item.get("metadata", {}) existing_meta = existing_item.get("metadata", {}) # Si les deux éléments ont des chemins d'image et que ces chemins sont identiques, # on considère que c'est le même élément et on le met à jour if (new_meta.get("image_path") and existing_meta.get("image_path") and new_meta["image_path"] == existing_meta["image_path"]): is_duplicate = True # Mettre à jour l'élément existant avec le nouveau combined_data[idx] = new_item print(f"Mise à jour de l'élément existant pour l'image {os.path.basename(new_meta['image_path'])}") break # Pour les rapports finaux, comparer par ticket_id et source_agent elif (step_name == "rapport_final" and new_meta.get("ticket_id") and existing_meta.get("ticket_id") and new_meta["ticket_id"] == existing_meta["ticket_id"] and new_meta.get("source_agent") == existing_meta.get("source_agent")): is_duplicate = True # Mettre à jour l'élément existant avec le nouveau combined_data[idx] = new_item break # Pour les analyses de tickets, analyses d'images ou tri d'images, vérifier les métadonnées du modèle elif (step_name in ["analyse_ticket", "analyse_image", "tri_image"] and new_meta.get("ticket_id") and existing_meta.get("ticket_id") and new_meta.get("timestamp") and existing_meta.get("timestamp") and new_meta["ticket_id"] == existing_meta["ticket_id"]): # Si les modèles sont les mêmes ou après normalisation if (new_meta.get("model_info", {}).get("model") and existing_meta.get("model_info", {}).get("model") and normaliser_nom_modele(new_meta["model_info"]["model"]) == normaliser_nom_modele(existing_meta["model_info"]["model"])): is_duplicate = True # Mettre à jour l'élément existant avec le nouveau combined_data[idx] = new_item break # Si ce n'est pas un doublon, l'ajouter aux données combinées if not is_duplicate: combined_data.append(new_item) print(f"Ajout d'un nouvel élément aux données combinées") # Si on n'a qu'un seul élément à la fin, le sortir de la liste pour garder # la même structure que précédemment pour les fichiers non multiples if not step_name.startswith("tri_image") and len(combined_data) == 1: final_data = combined_data[0] else: # Pour tri_image, toujours garder une liste pour accumuler les résultats final_data = combined_data # Sauvegarder les données combinées with open(file_path, "w", encoding="utf-8") as f: json.dump(final_data, f, ensure_ascii=False, indent=2) print(f"Données sauvegardées dans {file_path} ({len(combined_data)} entrées)") # Générer également une version texte pour faciliter la lecture generer_version_texte(final_data, ticket_id, step_name, file_path) except Exception as e: print(f"Erreur lors de la sauvegarde des données dans {file_path}: {e}")