import os import base64 import logging from typing import List, Dict, Any, Optional from .auth_manager import AuthManager from core.utils import save_json, normalize_filename class AttachmentManager: """ Gestionnaire de pièces jointes pour extraire et sauvegarder les fichiers attachés aux tickets. """ def __init__(self, auth: AuthManager): """ Initialise le gestionnaire de pièces jointes. Args: auth: Gestionnaire d'authentification """ self.auth = auth self.model_name = "project.task" self.excluded_mime_types = [] # Types MIME à exclure si nécessaire def get_ticket_attachments(self, ticket_id: int) -> List[Dict[str, Any]]: """ Récupère les pièces jointes associées à un ticket. Args: ticket_id: ID du ticket Returns: Liste des pièces jointes avec leurs métadonnées """ params = { "model": "ir.attachment", "method": "search_read", "args": [[["res_id", "=", ticket_id], ["res_model", "=", self.model_name]]], "kwargs": { "fields": ["id", "name", "mimetype", "file_size", "create_date", "create_uid", "datas", "description", "res_name"] } } attachments = self.auth._rpc_call("/web/dataset/call_kw", params) # Résoudre les informations sur le créateur for attachment in attachments: if "create_uid" in attachment and isinstance(attachment["create_uid"], list) and len(attachment["create_uid"]) >= 2: attachment["creator_name"] = attachment["create_uid"][1] attachment["creator_id"] = attachment["create_uid"][0] elif "create_uid" in attachment and isinstance(attachment["create_uid"], int): # Récupérer le nom du créateur params = { "model": "res.users", "method": "name_get", "args": [[attachment["create_uid"]]], "kwargs": {} } result = self.auth._rpc_call("/web/dataset/call_kw", params) if result and isinstance(result, list) and result[0] and len(result[0]) >= 2: attachment["creator_name"] = result[0][1] attachment["creator_id"] = result[0][0] return attachments if isinstance(attachments, list) else [] def download_attachment(self, attachment: Dict[str, Any], output_dir: str) -> Dict[str, Any]: """ Télécharge et sauvegarde une pièce jointe dans le répertoire spécifié. Args: attachment: Dictionnaire contenant les métadonnées de la pièce jointe output_dir: Répertoire où sauvegarder la pièce jointe Returns: Dictionnaire avec les informations sur le fichier sauvegardé """ result = { "id": attachment.get("id"), "name": attachment.get("name", "Sans nom"), "mimetype": attachment.get("mimetype", "application/octet-stream"), "file_size": attachment.get("file_size", 0), "create_date": attachment.get("create_date"), "creator": attachment.get("creator_name", "Inconnu"), "status": "error", "file_path": "", "error": "" } if not attachment.get("datas"): result["error"] = "Données de pièce jointe manquantes" return result try: # Créer le dossier attachments s'il n'existe pas attachments_dir = os.path.join(output_dir, "attachments") os.makedirs(attachments_dir, exist_ok=True) # Construire un nom de fichier sécurisé safe_filename = normalize_filename(attachment.get("name", f"attachment_{attachment.get('id')}.bin")) file_path = os.path.join(attachments_dir, safe_filename) # Vérifier si un fichier avec le même nom existe déjà if os.path.exists(file_path): base, ext = os.path.splitext(safe_filename) counter = 1 while os.path.exists(file_path): new_filename = f"{base}_{counter}{ext}" file_path = os.path.join(attachments_dir, new_filename) counter += 1 # Décoder et sauvegarder le contenu file_content = base64.b64decode(attachment["datas"]) with open(file_path, "wb") as f: f.write(file_content) result["status"] = "success" result["file_path"] = file_path return result except Exception as e: logging.error(f"Erreur lors du téléchargement de la pièce jointe {attachment.get('name', '')}: {e}") result["error"] = str(e) return result def save_attachments(self, ticket_id: int, output_dir: str, download: bool = True) -> List[Dict[str, Any]]: """ Récupère et sauvegarde toutes les pièces jointes d'un ticket. Args: ticket_id: ID du ticket output_dir: Répertoire de sortie download: Si True, télécharge les pièces jointes, sinon récupère seulement les métadonnées Returns: Liste des informations sur les pièces jointes """ # Récupérer les pièces jointes attachments = self.get_ticket_attachments(ticket_id) if not attachments: logging.info(f"Aucune pièce jointe trouvée pour le ticket {ticket_id}") return [] logging.info(f"Traitement de {len(attachments)} pièces jointes pour le ticket {ticket_id}") # Préparer les résultats attachments_info = [] # Télécharger chaque pièce jointe for i, attachment in enumerate(attachments): # Ne pas inclure le contenu binaire dans les métadonnées attachment_meta = {key: value for key, value in attachment.items() if key != "datas"} if download: # Télécharger et sauvegarder la pièce jointe download_result = self.download_attachment(attachment, output_dir) attachment_meta.update({ "download_status": download_result.get("status"), "local_path": download_result.get("file_path", ""), "error": download_result.get("error", "") }) if download_result.get("status") == "success": logging.info(f"Pièce jointe téléchargée: {attachment_meta.get('name')} ({i+1}/{len(attachments)})") else: logging.warning(f"Échec du téléchargement de la pièce jointe: {attachment_meta.get('name')} - {download_result.get('error')}") else: # Seulement récupérer les métadonnées attachment_meta.update({ "download_status": "not_attempted", "local_path": "", "error": "" }) attachments_info.append(attachment_meta) # Sauvegarder les informations sur les pièces jointes attachments_info_path = os.path.join(output_dir, "attachments_info.json") save_json(attachments_info, attachments_info_path) return attachments_info