import os import json import logging import time import traceback from typing import List, Dict, Any, Optional, Tuple, cast from agents.base_agent import BaseAgent from agents.llama_vision.agent_image_sorter import AgentImageSorter from loaders.ticket_data_loader import TicketDataLoader from utils.image_dedup import filtrer_images_uniques from utils.ocr_utils import extraire_texte from utils.translate_utils import fr_to_en, en_to_fr, sauvegarder_ocr_traduction # Configuration du logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', filename='orchestrator_llama.log', filemode='w') logger = logging.getLogger("OrchestratorLlamaVision") class OrchestratorLlamaVision: """Orchestrateur pour l'analyse des tickets avec llama_vision.""" def __init__(self, output_dir: str = "output/", ticket_agent: Optional[BaseAgent] = None, image_sorter: Optional[BaseAgent] = None, image_analyser: Optional[BaseAgent] = None, report_generator: Optional[BaseAgent] = None, vision_ocr: Optional[BaseAgent] = None, config: Optional[Dict[str, Any]] = None): """Initialisation de l'orchestrateur.""" self.output_dir = output_dir self.ticket_agent = ticket_agent self.image_sorter = image_sorter self.image_analyser = image_analyser self.report_generator = report_generator self.vision_ocr = vision_ocr self.ticket_loader = TicketDataLoader() self.config = { "dedup_enabled": True, "dedup_threshold": 5, "save_results": True, "debug_mode": False, "reports_dir": "reports", "ocr_enabled": True, "ocr_llm_enabled": True, "english_only": True, "model_name": "llama3-vision-90b-instruct" # Nom du modèle par défaut } if config: self.config.update(config) # Assurer la cohérence des noms de modèles if "model_name" in self.config: self.config["model_name"] = self.config["model_name"].replace(".", "-").replace(":", "-").replace("_", "-") logger.info(f"OrchestratorLlamaVision initialisé avec les paramètres: {self.config}") def executer(self, ticket_id: Optional[str] = None): """ Point d'entrée principal pour l'analyse d'un ticket. Version optimisée selon le nouveau flux. Args: ticket_id: ID du ticket à analyser """ if not ticket_id: logger.error(f"ID de ticket requis pour l'exécution") return ticket_path = os.path.join(self.output_dir, f"ticket_{ticket_id}") if not os.path.exists(ticket_path): logger.error(f"Le ticket {ticket_id} est introuvable dans {ticket_path}") return try: # Exécuter le ticket avec toutes les options par défaut logger.info(f"[PIPELINE] Début d'analyse du ticket {ticket_id}") print(f"[PIPELINE] Début d'analyse du ticket {ticket_id}") resultats = self.executer_ticket( ticket_id=ticket_id, # Maintenant ticket_id est garanti non-None ocr_enabled=self.config.get("ocr_enabled", True), ocr_avance_enabled=self.config.get("ocr_llm_enabled", True), image_triage_enabled=True, image_analyse_enabled=True, report_generation_enabled=True ) # Vérifier si une erreur s'est produite if "error" in resultats: logger.error(f"[PIPELINE] Erreur lors de l'analyse du ticket {ticket_id}: {resultats['error']}") print(f"[PIPELINE] Erreur lors de l'analyse du ticket {ticket_id}: {resultats['error']}") else: logger.info(f"[PIPELINE] Analyse du ticket {ticket_id} terminée avec succès") print(f"[PIPELINE] Analyse du ticket {ticket_id} terminée avec succès") except Exception as e: logger.error(f"[PIPELINE] Erreur globale sur le ticket {ticket_id}: {e}") if self.config.get("debug_mode"): logger.error(traceback.format_exc()) def executer_ticket(self, ticket_id: str, ocr_enabled: bool = True, ocr_avance_enabled: bool = True, image_triage_enabled: bool = True, image_analyse_enabled: bool = True, report_generation_enabled: bool = True) -> Dict[str, Any]: """ Exécute l'analyse d'un ticket avec des options pour activer/désactiver spécifiquement chaque étape du processus d'analyse. Args: ticket_id: Identifiant du ticket à analyser ocr_enabled: Activer l'OCR standard sur les images ocr_avance_enabled: Activer l'OCR avancé via LLM sur les images image_triage_enabled: Activer le tri des images pertinentes image_analyse_enabled: Activer l'analyse des images report_generation_enabled: Activer la génération du rapport final Returns: Dictionnaire contenant les résultats du traitement """ logger.info(f"=== Exécution du ticket {ticket_id} ===") logger.info(f"Options: OCR={ocr_enabled}, OCR_Avancé={ocr_avance_enabled}, " f"Tri_Images={image_triage_enabled}, Analyse_Images={image_analyse_enabled}, " f"Rapport={report_generation_enabled}") # Affichage dans le terminal pour suivi print(f"=== Exécution du ticket {ticket_id} ===") print(f"Options activées: OCR={ocr_enabled}, OCR_Avancé={ocr_avance_enabled}, " f"Tri_Images={image_triage_enabled}, Analyse_Images={image_analyse_enabled}, " f"Rapport={report_generation_enabled}") ticket_path = os.path.join(self.output_dir, f"ticket_{ticket_id}") if not os.path.exists(ticket_path): logger.error(f"Le ticket {ticket_id} est introuvable dans {ticket_path}") print(f"⚠️ Erreur: Le ticket {ticket_id} est introuvable") return {"error": f"Ticket {ticket_id} introuvable"} # Sauvegarder les configurations originales original_ocr_enabled = self.config.get("ocr_enabled", True) original_ocr_llm_enabled = self.config.get("ocr_llm_enabled", True) # Mettre à jour temporairement les configurations self.config["ocr_enabled"] = ocr_enabled self.config["ocr_llm_enabled"] = ocr_avance_enabled # Variables pour collecter tous les résultats resultats = { "ticket_analysis": None, "image_triage": {}, "image_analysis": {}, "ocr_standard": {}, "ocr_advanced": {}, "report": None } try: # 1. Trouver le répertoire d'extraction extractions = self._trouver_extractions(ticket_path, ticket_id) if not extractions: logger.warning(f"Aucune extraction trouvée pour le ticket {ticket_id}") print(f"⚠️ Aucune extraction trouvée pour le ticket {ticket_id}") return {"error": "Aucune extraction trouvée"} extraction_path = extractions[0] logger.info(f"Répertoire d'extraction: {extraction_path}") # Chemins des dossiers principaux attachments_dir = os.path.join(extraction_path, "attachments") rapport_dir = os.path.join(extraction_path, f"{ticket_id}_rapports") os.makedirs(rapport_dir, exist_ok=True) # Créer le répertoire pipeline une seule fois pipeline_dir = os.path.join(rapport_dir, "pipeline") os.makedirs(pipeline_dir, exist_ok=True) # Récupérer le nom du modèle pour le logging model_name = self.config.get("model_name", "llama3-vision-90b-instruct") # Normaliser pour éviter les problèmes dans les noms de fichiers model_name = model_name.replace(".", "-").replace(":", "-").replace("_", "-") logger.info(f"Utilisation du modèle: {model_name}") print(f"Modèle utilisé: {model_name}") # Charger les données du ticket json_path = self.ticket_loader.trouver_ticket(extraction_path, ticket_id) logger.info(f"Fichier ticket JSON: {json_path}") ticket_data = self._charger_ticket(json_path) if not ticket_data: logger.error(f"Impossible de charger les données du ticket {ticket_id}") print(f"⚠️ Impossible de charger les données du ticket {ticket_id}") return {"error": "Impossible de charger les données du ticket"} # Ajouter le chemin du fichier JSON au ticket_data pour faciliter l'extraction du ticket_id if json_path: ticket_data["file_path"] = json_path # 1. ANALYSE DU TICKET - Directement en français (nouvelle approche) if self.ticket_agent: logger.info(f"1️⃣ Exécution de l'agent d'analyse de ticket pour {ticket_id}") print(f"1️⃣ Analyse du ticket {ticket_id} (directement en français)...") try: # Conserver le contenu original (français) sans traduire original_content = ticket_data.get("content", "") logger.info(f"[LANGUE] Analyse du ticket directement en français: {len(original_content)} caractères") # Analyse en français directement (sans traduction préalable) ticket_analysis = self.ticket_agent.executer(ticket_data) resultats["ticket_analysis"] = ticket_analysis # Sauvegarde des résultats d'analyse de ticket if ticket_analysis: from agents.utils.pipeline_logger import sauvegarder_donnees sauvegarder_donnees( ticket_id=ticket_id, step_name="analyse_ticket", data=ticket_analysis, base_dir=rapport_dir, is_resultat=True ) logger.info(f"Analyse du ticket en français terminée et sauvegardée") print(f"✅ Analyse du ticket en français terminée") else: logger.error(f"L'analyse du ticket n'a pas produit de résultat") print(f"❌ L'analyse du ticket n'a pas produit de résultat") except Exception as e: logger.error(f"Erreur lors de l'analyse du ticket: {e}") print(f"❌ Erreur lors de l'analyse du ticket: {e}") if self.config.get("debug_mode"): logger.error(traceback.format_exc()) # 2. DÉDUPLICATION ET OCR STANDARD # Lister et filtrer les images images = [] if os.path.exists(attachments_dir): images = self._lister_images(attachments_dir) logger.info(f"Trouvé {len(images)} images dans {attachments_dir}") # Déduplication des images if self.config.get("dedup_enabled", True): original_count = len(images) images = filtrer_images_uniques(images, seuil_hamming=self.config["dedup_threshold"], ticket_id=ticket_id) logger.info(f"Déduplication: {original_count} → {len(images)} images uniques") print(f"Déduplication d'images: {original_count} → {len(images)} images uniques") # OCR standard sur les images ocr_results = {} if ocr_enabled and images: logger.info(f"2️⃣ Traitement OCR standard de {len(images)} images") print(f"2️⃣ OCR standard sur {len(images)} images...") for img in images: try: logger.info(f"OCR sur image: {img}") ocr_fr, langue = extraire_texte(img, lang="auto") # Traduire le texte extrait en anglais si nécessaire ocr_en = fr_to_en(ocr_fr) if ocr_fr and langue in ["fr", "fra", "unknown"] else ocr_fr # Sauvegarder les résultats OCR sauvegarder_ocr_traduction(img, ticket_id, ocr_fr, ocr_en, "", base_dir=rapport_dir) # Stocker le résultat pour utilisation lors du tri (uniquement) ocr_results[img] = { "texte_fr": ocr_fr, "texte_en": ocr_en, "langue_detectee": langue } resultats["ocr_standard"][img] = ocr_results[img] logger.info(f"OCR terminé pour {os.path.basename(img)}: {len(ocr_fr)} caractères ({langue})") print(f" • OCR terminé: {os.path.basename(img)} ({len(ocr_fr)} caractères)") except Exception as e: logger.warning(f"Erreur OCR pour {os.path.basename(img)}: {e}") print(f" ⚠️ Erreur OCR pour {os.path.basename(img)}") print(f"✅ OCR standard terminé - utilisé uniquement pour le tri des images") # 3. TRI DES IMAGES relevant_images = [] images_analyses = {} if image_triage_enabled and self.image_sorter and images: logger.info(f"3️⃣ Tri de {len(images)} images") print(f"3️⃣ Tri de {len(images)} images (originales)...") for img in images: try: # Utiliser l'OCR comme contexte pour le tri uniquement ocr_context = ocr_results.get(img, {"texte_en": ""}).get("texte_en", "") logger.info(f"[TRI] Image originale: {img}") logger.info(f"[TRI] Contexte OCR fourni: {len(ocr_context)} caractères") result_sort = self.image_sorter.executer(img, ocr_context=ocr_context) # Vérifier si l'image est pertinente is_relevant = result_sort.get("is_relevant", True) if is_relevant: relevant_images.append(img) logger.info(f"[TRI] Image {os.path.basename(img)} marquée comme pertinente") # Stocker les résultats images_analyses[img] = { "sorting": result_sort or {"is_relevant": True}, "analysis": None } resultats["image_triage"][img] = result_sort print(f" • Image {os.path.basename(img)}: {'Pertinente ✅' if is_relevant else 'Non pertinente ❌'}") except Exception as e: logger.warning(f"Erreur tri image {os.path.basename(img)}: {e}") print(f" ⚠️ Erreur lors du tri de {os.path.basename(img)}") # Sauvegarder les résultats de tri if hasattr(self.image_sorter, "sauvegarder_resultats"): try: self.image_sorter.sauvegarder_resultats() logger.info(f"Résultats de tri sauvegardés ({len(relevant_images)} pertinentes sur {len(images)})") print(f"✅ Résultats de tri sauvegardés: {len(relevant_images)} images pertinentes sur {len(images)}") except Exception as e: logger.error(f"Erreur lors de la sauvegarde des résultats de tri: {e}") print(f"⚠️ Erreur lors de la sauvegarde des résultats de tri") else: # Si pas de tri, toutes les images sont considérées pertinentes relevant_images = images.copy() for img in images: images_analyses[img] = { "sorting": {"is_relevant": True}, "analysis": None } print(f"ℹ️ Tri des images désactivé, toutes les images ({len(images)}) sont considérées pertinentes") # IMPORTANT: Abandonner explicitement les résultats OCR standard logger.info(f"[NETTOYAGE] Abandon explicite des résultats OCR standard après tri des images") ocr_results = None # Libérer la mémoire # 4. OCR AVANCÉ sur les images pertinentes ocr_llm_results = {} if ocr_avance_enabled and self.vision_ocr and relevant_images: logger.info(f"4️⃣ OCR avancé (LLM) sur {len(relevant_images)} images pertinentes") print(f"4️⃣ OCR avancé (LLM) sur {len(relevant_images)} images pertinentes...") for img in relevant_images: try: # Exécuter l'OCR avancé via LLM sans contexte supplémentaire logger.info(f"[OCR-LLM] Analyse OCR avancé sur image originale: {img}") print(f" • Exécution de l'OCR avancé sur {os.path.basename(img)}...") # Utiliser le prompt natif sans données supplémentaires ocr_result = self.vision_ocr.executer(img, ocr_baseline="") if ocr_result: # Vérifier que le résultat contient du texte extrait has_text = "extracted_text" in ocr_result and ocr_result["extracted_text"] if has_text: # Stocker les résultats ocr_llm_results[img] = ocr_result resultats["ocr_advanced"][img] = ocr_result # Mettre à jour les informations d'OCR dans images_analyses if img in images_analyses: images_analyses[img]["ocr_llm"] = ocr_result extracted_text_len = len(ocr_result.get('extracted_text', '')) logger.info(f"OCR avancé terminé pour {os.path.basename(img)}: {extracted_text_len} caractères") print(f" ✅ OCR avancé terminé: {os.path.basename(img)} ({extracted_text_len} caractères)") else: logger.warning(f"OCR avancé sans texte extrait pour {os.path.basename(img)}") print(f" ⚠️ OCR avancé sans texte extrait pour {os.path.basename(img)}") else: logger.warning(f"Pas de résultat OCR avancé pour {os.path.basename(img)}") print(f" ❌ Pas de résultat OCR avancé pour {os.path.basename(img)}") except Exception as e: logger.error(f"Erreur OCR avancé pour {os.path.basename(img)}: {e}") print(f" ❌ Erreur OCR avancé pour {os.path.basename(img)}: {e}") # Sauvegarder les résultats d'OCR avancé if hasattr(self.vision_ocr, "sauvegarder_resultats"): try: self.vision_ocr.sauvegarder_resultats() logger.info(f"Résultats d'OCR avancé sauvegardés") print(f"✅ Résultats d'OCR avancé sauvegardés") except Exception as e: logger.error(f"Erreur lors de la sauvegarde des résultats d'OCR avancé: {e}") print(f"⚠️ Erreur lors de la sauvegarde des résultats d'OCR avancé") print(f"✅ OCR avancé terminé pour toutes les images pertinentes") # 5. ANALYSE DES IMAGES pertinentes if image_analyse_enabled and self.image_analyser and relevant_images: logger.info(f"5️⃣ Analyse de {len(relevant_images)} images pertinentes") print(f"5️⃣ Analyse de {len(relevant_images)} images pertinentes...") analyses_resultats = [] analyses_traduites = [] for img in relevant_images: try: # Préparer uniquement le contexte OCR avancé ocr_llm = ocr_llm_results.get(img, {}) # Vérifier que l'OCR LLM contient bien du texte has_llm_text = ocr_llm and "extracted_text" in ocr_llm and ocr_llm["extracted_text"] if has_llm_text: logger.info(f"[ANALYSE] Contexte OCR LLM disponible pour {os.path.basename(img)}: {len(ocr_llm.get('extracted_text', ''))} caractères") else: logger.warning(f"[ANALYSE] OCR LLM non disponible pour {os.path.basename(img)}") # Contexte minimaliste: uniquement l'analyse du ticket (EN) et l'OCR LLM si disponible contexte_enrichi = self._enrichir_contexte( ticket_analysis if ticket_analysis else {}, ocr_llm ) # Analyser l'image logger.info(f"[ANALYSE] Analyse de l'image originale: {img}") print(f" • Analyse de l'image {os.path.basename(img)}...") # Produire l'analyse en anglais (compatible avec LlamaVision) result = self.image_analyser.executer(img, contexte=contexte_enrichi) if result: # Traduire le résultat en français pour l'étape finale if "analyse_en" in result: logger.info(f"[TRADUCTION] Traduction du résultat d'analyse d'image EN → FR") if not "analyse" in result or not result["analyse"]: result["analyse"] = en_to_fr(result["analyse_en"]) logger.info(f"[TRADUCTION] Analyse traduite: {len(result['analyse'])} caractères") images_analyses[img]["analysis"] = result analyses_resultats.append(result) analyses_traduites.append(result) # Inclut la traduction française logger.info(f"Analyse terminée pour {os.path.basename(img)}") print(f" ✅ Analyse terminée pour {os.path.basename(img)}") else: logger.warning(f"Pas de résultat d'analyse pour {os.path.basename(img)}") except Exception as e: logger.error(f"Erreur analyse image {os.path.basename(img)}: {e}") print(f" ❌ Erreur lors de l'analyse de {os.path.basename(img)}: {e}") # Sauvegarder les résultats d'analyse (traduits en français) if hasattr(self.image_analyser, "sauvegarder_resultats"): try: # Utiliser sauvegarder_resultats sans arguments self.image_analyser.sauvegarder_resultats() logger.info(f"Résultats d'analyse d'images (FR) sauvegardés via agent") print(f"✅ Résultats d'analyse d'images sauvegardés via agent") except Exception as e: logger.error(f"Erreur lors de la sauvegarde via agent: {e}") print(f"⚠️ Erreur lors de la sauvegarde via agent") # En cas d'erreur, utiliser pipeline_logger directement try: from agents.utils.pipeline_logger import sauvegarder_donnees sauvegarder_donnees( ticket_id=ticket_id, step_name="analyse_image", data=analyses_traduites, # Utiliser les analyses traduites base_dir=rapport_dir, is_resultat=True ) logger.info(f"Analyse d'images (FR) sauvegardée via sauvegarder_donnees (fallback)") print(f"✅ Analyse d'images sauvegardée via sauvegarder_donnees (fallback)") except Exception as e2: logger.error(f"Erreur lors de la sauvegarde des analyses d'images: {e2}") print(f"⚠️ Erreur lors de la sauvegarde des analyses d'images") else: # Si la méthode n'existe pas, utiliser directement pipeline_logger try: from agents.utils.pipeline_logger import sauvegarder_donnees sauvegarder_donnees( ticket_id=ticket_id, step_name="analyse_image", data=analyses_traduites, # Utiliser les analyses traduites base_dir=rapport_dir, is_resultat=True ) logger.info(f"Analyse d'images (FR) sauvegardée via sauvegarder_donnees") print(f"✅ Analyse d'images sauvegardée via sauvegarder_donnees") except Exception as e: logger.error(f"Erreur lors de la sauvegarde des analyses d'images: {e}") print(f"⚠️ Erreur lors de la sauvegarde des analyses d'images") # 6. GÉNÉRATION DU RAPPORT FINAL if report_generation_enabled and self.report_generator and resultats["ticket_analysis"]: try: # Normaliser le nom du modèle pour éviter les doublons de rapports model_name = self.config.get("model_name", "").replace(".", "-").replace(":", "-").replace("_", "-") if not model_name: model_name = "llama3-vision-90b-instruct" # Préparer les données pour le rapport final (tout en français) rapport_data = { "ticket_id": ticket_id, "ticket_data": ticket_data, "ticket_analyse": resultats["ticket_analysis"], # Analyse en français "analyse_images": images_analyses, # Analyses traduites en français "metadata": { "model_name": model_name } } logger.info(f"6️⃣ Génération du rapport final entièrement en français pour le ticket {ticket_id}") print(f"6️⃣ Génération du rapport final en français pour le ticket {ticket_id}...") rapport_final = self.report_generator.executer(rapport_data) resultats["report"] = rapport_final logger.info(f"Rapport généré avec succès") print(f"✅ Rapport généré avec succès") except Exception as e: logger.error(f"Erreur lors de la génération du rapport: {e}") print(f"❌ Erreur lors de la génération du rapport: {e}") if self.config.get("debug_mode"): logger.error(traceback.format_exc()) # 7. EXTRACTION CSV (utilisation du processus existant) if report_generation_enabled and resultats["report"]: try: logger.info(f"7️⃣ Extraction CSV pour le ticket {ticket_id}") print(f"7️⃣ Extraction CSV pour le ticket {ticket_id}...") from agents.utils.report_csv_exporter import traiter_rapports_ticket traiter_rapports_ticket(ticket_id) logger.info(f"Extraction CSV terminée avec succès") print(f"✅ Extraction CSV terminée avec succès") except Exception as e: logger.error(f"Erreur lors de l'extraction CSV: {e}") print(f"⚠️ Erreur lors de l'extraction CSV: {e}") logger.info(f"=== Traitement du ticket {ticket_id} terminé ===") print(f"=== Traitement du ticket {ticket_id} terminé ===") except Exception as e: logger.error(f"Erreur globale sur le ticket {ticket_id}: {e}") print(f"❌ Erreur globale sur le ticket {ticket_id}: {e}") if self.config.get("debug_mode"): logger.error(traceback.format_exc()) resultats["error"] = str(e) finally: # Restaurer les configurations originales self.config["ocr_enabled"] = original_ocr_enabled self.config["ocr_llm_enabled"] = original_ocr_llm_enabled return resultats def _enrichir_contexte(self, ticket_analyse: Dict[str, Any], ocr_llm: Dict[str, Any]) -> Dict[str, Any]: """ Enrichit le contexte pour l'analyse d'image avec l'OCR avancé et l'analyse du ticket. Version simplifiée qui évite les transmissions inutiles. Args: ticket_analyse: Résultat de l'analyse du ticket ocr_llm: Résultat OCR avancé Returns: Dictionnaire contenant le contexte enrichi """ contexte = {} # 1. Extraire les informations pertinentes de l'analyse du ticket if ticket_analyse: # Inclure directement la réponse d'analyse if "response" in ticket_analyse: contexte["ticket_analyse"] = ticket_analyse["response"] elif "response_en" in ticket_analyse: contexte["ticket_analyse"] = ticket_analyse["response_en"] # Tracer les informations incluses logger.info(f"[CONTEXTE] Analyse de ticket incluse: {len(contexte.get('ticket_analyse', ''))} caractères") # 2. Ajouter le texte OCR avancé ocr_llm_len = 0 if ocr_llm: # Chercher le champ extracted_text qui est le standard pour l'agent OCR Vision if "extracted_text" in ocr_llm and ocr_llm["extracted_text"]: contexte["ocr_llm_text"] = ocr_llm["extracted_text"] ocr_llm_len = len(ocr_llm["extracted_text"]) logger.info(f"[CONTEXTE] OCR LLM inclus: {ocr_llm_len} caractères") else: # Chercher des champs alternatifs for field in ["text", "text_content", "content"]: if field in ocr_llm and ocr_llm[field]: contexte["ocr_llm_text"] = ocr_llm[field] ocr_llm_len = len(ocr_llm[field]) logger.info(f"[CONTEXTE] OCR LLM (champ {field}) inclus: {ocr_llm_len} caractères") break # Tracer le contexte total logger.info(f"[CONTEXTE] Contexte enrichi construit avec {len(contexte)} éléments") return contexte def _charger_ticket(self, json_path: Optional[str]) -> Optional[Dict[str, Any]]: """ Charge et prépare les données du ticket à partir d'un fichier JSON. Ajoute des logs détaillés sur le contenu chargé. Args: json_path: Chemin du fichier JSON Returns: Données du ticket chargées ou None en cas d'erreur """ if not json_path: logger.warning("[TICKET] Aucun chemin JSON fourni") return None try: logger.info(f"[TICKET] Chargement du ticket depuis {json_path}") ticket_data = self.ticket_loader.charger(json_path) # Vérifier le chargement if not ticket_data: logger.error(f"[TICKET] Échec du chargement: le fichier {json_path} a retourné des données vides") return None # Préparer le contenu du ticket à partir des messages messages = ticket_data.get("messages", []) contenu = [] # Ajouter le titre/description if "name" in ticket_data: contenu.append(f"TITRE: {ticket_data['name']}") logger.info(f"[TICKET] Titre: {ticket_data['name']}") if "description" in ticket_data and ticket_data["description"] != "*Contenu non extractible*": contenu.append(f"DESCRIPTION: {ticket_data['description']}") description_sample = ticket_data['description'][:100] + "..." if len(ticket_data['description']) > 100 else ticket_data['description'] logger.info(f"[TICKET] Description: {description_sample}") # Ajouter chaque message for msg in messages: auteur = msg.get("author_id", "Inconnu") date = msg.get("date", "") msg_type = msg.get("message_type", "") content = msg.get("content", "").strip() if content: contenu.append(f"\n[{date}] {auteur} ({msg_type}):") contenu.append(content) # Ajouter le contenu formaté au ticket_data ticket_data["content"] = "\n".join(contenu) # Ajouter le chemin du fichier pour référence ticket_data["file_path"] = json_path logger.info(f"[TICKET] Données chargées: {len(messages)} messages, {len(ticket_data['content'])} caractères") logger.debug(f"[TICKET] Extrait du contenu: {ticket_data['content'][:200]}...") return ticket_data except Exception as e: logger.error(f"[TICKET] Erreur chargement JSON: {e}") return None def _trouver_extractions(self, ticket_path: str, ticket_id: str) -> List[str]: return sorted( [os.path.join(ticket_path, d) for d in os.listdir(ticket_path) if os.path.isdir(os.path.join(ticket_path, d)) and d.startswith(ticket_id)], key=lambda x: os.path.getmtime(x), reverse=True ) def _lister_images(self, dossier: str) -> List[str]: """ Liste toutes les images dans un dossier avec une reconnaissance étendue des formats d'images et des logs détaillés. Args: dossier: Dossier contenant les images à analyser Returns: Liste des chemins absolus d'images trouvées """ # Liste étendue des extensions d'images courantes extensions = ['.png', '.jpg', '.jpeg', '.gif', '.bmp', '.webp', '.tiff', '.tif'] images = [] logger.info(f"[IMAGES] Recherche d'images dans {dossier}") # Parcourir le dossier pour trouver toutes les images if os.path.exists(dossier): for racine, _, fichiers in os.walk(dossier): for f in fichiers: # Vérifier l'extension du fichier (non sensible à la casse) if any(f.lower().endswith(ext) for ext in extensions): chemin_complet = os.path.abspath(os.path.join(racine, f)) # Vérifier que le fichier est bien une image valide et accessible try: from PIL import Image with Image.open(chemin_complet) as img: # S'assurer que c'est bien une image en vérifiant ses dimensions width, height = img.size if width > 0 and height > 0: images.append(chemin_complet) logger.debug(f"[IMAGES] Image valide trouvée: {chemin_complet} ({width}x{height})") except Exception as e: logger.warning(f"[IMAGES] Image ignorée {chemin_complet}: {str(e)}") if not images: logger.warning(f"[IMAGES] Aucune image trouvée dans {dossier}") else: logger.info(f"[IMAGES] {len(images)} images trouvées dans {dossier}") for idx, img in enumerate(images): logger.info(f"[IMAGES] {idx+1}/{len(images)}: {img}") return images def _sauvegarder_resultats(self, resultats_analyses_tickets, resultats_analyses_images, resultats_tri_images=None, ticket_id=None, resultats_ocr=None, resultats_ocr_llm=None, resultats_rapports=None): """ Sauvegarde les différents résultats générés pendant l'analyse du ticket Args: resultats_analyses_tickets: Résultats de l'analyse du ticket resultats_analyses_images: Résultats de l'analyse des images resultats_tri_images: Résultats du tri des images ticket_id: ID du ticket resultats_ocr: Résultats OCR standard resultats_ocr_llm: Résultats OCR avancé resultats_rapports: Résultats de la génération de rapports """ logger.info("Sauvegarde des résultats") os.makedirs(self.output_dir, exist_ok=True) # Sauvegarde de l'analyse du ticket if resultats_analyses_tickets: chemin_fichier = os.path.join(self.output_dir, f"analyse_ticket{self._format_fichier(ticket_id)}") try: with open(chemin_fichier + ".json", "w") as file: json.dump(resultats_analyses_tickets, file, indent=2) with open(chemin_fichier + ".txt", "w") as file: file.write(json.dumps(resultats_analyses_tickets, indent=2)) logger.debug(f"Analyse du ticket sauvegardée dans {chemin_fichier}(.json/.txt)") except Exception as e: logger.error(f"Erreur lors de la sauvegarde de l'analyse du ticket: {e}") # Sauvegarde de l'analyse des images if resultats_analyses_images: chemin_fichier = os.path.join(self.output_dir, f"analyse_image{self._format_fichier(ticket_id)}") try: with open(chemin_fichier + ".json", "w") as file: json.dump(resultats_analyses_images, file, indent=2) with open(chemin_fichier + ".txt", "w") as file: file.write(json.dumps(resultats_analyses_images, indent=2)) logger.debug(f"Analyse des images sauvegardée dans {chemin_fichier}(.json/.txt)") except Exception as e: logger.error(f"Erreur lors de la sauvegarde de l'analyse des images: {e}") # Sauvegarde du tri des images if resultats_tri_images: chemin_fichier = os.path.join(self.output_dir, f"tri_image{self._format_fichier(ticket_id)}") try: with open(chemin_fichier + ".json", "w") as file: json.dump(resultats_tri_images, file, indent=2) with open(chemin_fichier + ".txt", "w") as file: file.write(json.dumps(resultats_tri_images, indent=2)) logger.debug(f"Tri des images sauvegardé dans {chemin_fichier}(.json/.txt)") except Exception as e: logger.error(f"Erreur lors de la sauvegarde du tri des images: {e}") # Sauvegarde des résultats OCR standard if resultats_ocr: chemin_fichier = os.path.join(self.output_dir, f"ocr{self._format_fichier(ticket_id)}") try: with open(chemin_fichier + ".json", "w") as file: json.dump(resultats_ocr, file, indent=2) with open(chemin_fichier + ".txt", "w") as file: file.write(json.dumps(resultats_ocr, indent=2)) logger.debug(f"Résultats OCR sauvegardés dans {chemin_fichier}(.json/.txt)") except Exception as e: logger.error(f"Erreur lors de la sauvegarde des résultats OCR: {e}") # Sauvegarde des résultats OCR avancé if resultats_ocr_llm: chemin_fichier = os.path.join(self.output_dir, f"ocr_llm{self._format_fichier(ticket_id)}") try: with open(chemin_fichier + ".json", "w") as file: json.dump(resultats_ocr_llm, file, indent=2) with open(chemin_fichier + ".txt", "w") as file: file.write(json.dumps(resultats_ocr_llm, indent=2)) logger.debug(f"Résultats OCR avancé sauvegardés dans {chemin_fichier}(.json/.txt)") except Exception as e: logger.error(f"Erreur lors de la sauvegarde des résultats OCR avancé: {e}") # Sauvegarde des rapports if resultats_rapports: chemin_fichier = os.path.join(self.output_dir, f"rapport{self._format_fichier(ticket_id)}") try: with open(chemin_fichier + ".json", "w") as file: json.dump(resultats_rapports, file, indent=2) with open(chemin_fichier + ".txt", "w") as file: file.write(json.dumps(resultats_rapports, indent=2)) logger.debug(f"Rapport sauvegardé dans {chemin_fichier}(.json/.txt)") except Exception as e: logger.error(f"Erreur lors de la sauvegarde du rapport: {e}") def _format_fichier(self, ticket_id): return f"_{ticket_id}" if ticket_id else "" def _sauvegarder_resultat(self, data: Any, chemin_base: str) -> None: """ Sauvegarde des données dans des fichiers JSON et TXT. Args: data: Données à sauvegarder chemin_base: Chemin de base pour les fichiers (sans extension) """ try: # Extraire le nom de fichier pour le logging fichier_nom = os.path.basename(chemin_base) # Extraire le type de données et le nom du modèle depuis le chemin type_donnees = "inconnu" modele_nom = "inconnu" for prefix in ["analyse_ticket_", "tri_image_", "analyse_image_", "ocr_standard_", "ocr_avance_", "rapport_"]: if fichier_nom.startswith(prefix): type_donnees = prefix.replace("_", "") modele_nom = fichier_nom[len(prefix):] break logger.info(f"Sauvegarde des données '{type_donnees}' avec modèle '{modele_nom}'") # Vérifier si le répertoire existe, sinon le créer repertoire = os.path.dirname(chemin_base) if not os.path.exists(repertoire): os.makedirs(repertoire, exist_ok=True) logger.debug(f"Création du répertoire pour la sauvegarde: {repertoire}") # Normaliser le nom du modèle dans les métadonnées des données à sauvegarder if isinstance(data, dict) and "model_info" in data: # Vérifier si le nom du modèle est inconnu et le remplacer if data["model_info"].get("model", "").lower() in ["unknown", "unknown_model", "inconnu"]: # Utiliser le nom du modèle extrait du chemin data["model_info"]["model"] = modele_nom logger.info(f"Remplacement du nom de modèle inconnu par '{modele_nom}' dans les données") elif isinstance(data, list) and data: # Pour les listes, vérifier chaque élément for item in data: if isinstance(item, dict) and "model_info" in item: if item["model_info"].get("model", "").lower() in ["unknown", "unknown_model", "inconnu"]: item["model_info"]["model"] = modele_nom logger.info(f"Remplacement du nom de modèle inconnu par '{modele_nom}' dans un élément") # Estimer la taille des données pour le logging data_size = 0 if isinstance(data, dict): data_size = len(data) elif isinstance(data, list): data_size = len(data) # Sauvegarder au format JSON json_path = f"{chemin_base}_results.json" with open(json_path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) # Sauvegarder au format TXT pour lisibilité humaine txt_path = f"{chemin_base}_results.txt" with open(txt_path, "w", encoding="utf-8") as f: f.write(json.dumps(data, ensure_ascii=False, indent=2)) logger.info(f"Données sauvegardées avec succès: {type_donnees}_{modele_nom} ({data_size} éléments)") logger.debug(f"Fichiers créés: {json_path} et {txt_path}") except Exception as e: logger.error(f"Erreur lors de la sauvegarde des données: {e}") logger.error(f"Chemin de base: {chemin_base}") if self.config.get("debug_mode"): logger.error(traceback.format_exc())