from odoo_connection import OdooConnection import os import json from utils import save_json, ensure_export_directory, print_error from config import EXPORT_DIR from data_filter import filter_ticket_data class TicketManager: """Gestionnaire de tickets simplifié avec seulement les fonctionnalités essentielles""" def __init__(self): """Initialise le gestionnaire de tickets""" self.conn = OdooConnection() self.odoo = self.conn.get_odoo_instance() self.model_name = "project.task" def _ensure_connection(self): """Vérifie et établit la onnexion si nécessaire""" if not self.odoo: self.odoo = self.conn.get_odoo_instance() return self.odoo is not None def _safe_execute(self, model, method, *args): """ Exécute une méthode Odoo en toute sécurité via odoorpc. Vérifie la connexion avant d'exécuter. """ # Vérifier que la connexion est bien établie if not self._ensure_connection(): print_error("Connexion Odoo indisponible.") return None try: # Exécuter la méthode sur le modèle via OdooRPC return self.odoo.execute(model, method, *args) # type: ignore except odoorpc.error.RPCError as e: # type: ignore print_error(f" Erreur RPC lors de '{method}' sur '{model}': {e}") return None except Exception as e: print_error(f" Erreur inattendue lors de '{method}' sur '{model}': {e}") return None def get_ticket_by_id(self, ticket_id): """ Récupère les détails d'un ticket par son ID et applique le filtre """ fields_to_read = ['id', 'name', 'code', 'stage_id', 'date_deadline', 'description', 'message_ids'] # Vérifier la connexion avant d'exécuter la requête if not self._ensure_connection(): print_error("Connexion Odoo indisponible.") return None # Récupérer les données du ticket ticket_data = self._safe_execute(self.model_name, 'read', [ticket_id], fields_to_read) if not ticket_data: print_error(f"Aucun ticket trouvé avec l'ID {ticket_id}") return None # Nettoyer et filtrer les données du ticket return filter_ticket_data(ticket_data[0]) # Utilisation de data_filter.py def list_models(self): """Affiche la liste des modèles disponibles dans Odoo""" models = self._safe_execute('ir.model', 'search_read', [], ['model', 'name']) if not models: print_error("Aucun modèle disponible.") return [] print("\nListe des modèles disponibles:") for model in models: print(f"Modèle: {model['name']} (ID: {model['model']})") return models def list_model_fields(self, model_name): """Affiche les champs d'un modèle donné""" fields_info = self._safe_execute(model_name, 'fields_get') if not fields_info: print_error(f"Aucun champ trouvé pour le modèle {model_name}.") return [] print(f"\nChamps du modèle {model_name}:") for field_name, field_data in fields_info.items(): print(f"Champ: {field_name} - Type: {field_data['type']}") return fields_info def export_model_fields_to_json(self, model_name, filename): """Exporte les champs d'un modèle dans un fichier JSON""" fields_info = self._safe_execute(model_name, 'fields_get') if not fields_info: print_error(f"Aucun champ trouvé pour le modèle {model_name}.") return data = {field_name: field_data['type'] for field_name, field_data in fields_info.items()} filepath = os.path.join(EXPORT_DIR, filename) if save_json(filepath, data): print(f"Champs du modèle {model_name} exportés dans {filepath}") else: print_error(f"Erreur lors de l'exportation des champs du modèle {model_name} dans {filepath}") def export_tickets_by_project_and_stage(self, project_id, selected_stage_ids=None): """ Exporte les tickets d'un projet classés par étape """ # Vérifier la connexion Odoo if not self._ensure_connection(): print_error("Connexion Odoo indisponible.") return # Récupérer le projet depuis Odoo project_data = self._safe_execute('project.project', 'search_read', [('id', '=', project_id)], ['id', 'name']) if not project_data: print_error(f"Projet ID {project_id} introuvable.") return project_name = project_data[0]['name'] # Construire le domaine de recherche des tickets domain = [('project_id', '=', project_id)] if selected_stage_ids: domain.append(('stage_id', 'in', selected_stage_ids)) # Récupérer les tickets du projet ticket_ids = self._safe_execute(self.model_name, 'search', domain, 0, 1000) if not ticket_ids: print_error("Aucun ticket trouvé pour ce projet.") return # Lire les détails des tickets tickets = [self.get_ticket_by_id(ticket_id) for ticket_id in ticket_ids if self.get_ticket_by_id(ticket_id)] # Trier les tickets par étape tickets_by_stage = {} for ticket in tickets: #Vérifier que ticket est bien un dictionnaire if not isinstance(ticket, dict): print_error("Erreur: le format du ticket est invalide.") return None # Vérifier que "Champs Relationnels" existe champs_relationnels = ticket.get("Champs Relationnels", {}) # Vérifier que "stage_id" existe et est une liste avec au moins 2 éléments stage_data = champs_relationnels.get("stage_id", [0, "Non classé"]) stage_id = stage_data[0] if isinstance(stage_data, list) and len(stage_data) > 0 else 0 stage_name = stage_data[1] if isinstance(stage_data, list) and len(stage_data) > 1 else "Non classé" key = f"{stage_id}_{stage_name}" if key not in tickets_by_stage: tickets_by_stage[key] = [] tickets_by_stage[key].append(ticket) # Créer le répertoire du projet project_dir = ensure_export_directory(f"project_{project_id}_{project_name.replace(' ', '_')}") # Sauvegarder les tickets par étape for stage_key, stage_tickets in tickets_by_stage.items(): stage_dir = os.path.join(project_dir, stage_key) os.makedirs(stage_dir, exist_ok=True) save_json(os.path.join(stage_dir, "all_tickets.json"), stage_tickets) print(f"Exportation terminée. Les fichiers sont enregistrés dans : {project_dir}/")