llm_ticket3/odoo/auth_manager.py

import json
import logging
import os
import requests
from typing import Dict, Any, Optional

# Variable globale pour stocker l'instance du gestionnaire d'authentification
_auth_manager_instance = None

class AuthManager:
    """
    Gestionnaire d'authentification pour l'API Odoo.
    Gère la connexion et les appels RPC à l'API Odoo.
    """
    
    def __init__(self, url: str, db: str, username: str, api_key: str):
        """
        Initialise le gestionnaire d'authentification.
        
        Args:
            url: URL de l'instance Odoo
            db: Nom de la base de données Odoo
            username: Nom d'utilisateur pour la connexion
            api_key: Clé API ou mot de passe pour l'authentification
        """
        self.url = url.rstrip('/')
        self.db = db
        self.username = username
        self.api_key = api_key
        self.uid = None
        self.session = requests.Session()
        self.session.headers.update({
            'Content-Type': 'application/json',
            'Accept': 'application/json'
        })
        self.max_retries = 3
        self.timeout = 30  # secondes
    
    def login(self) -> bool:
        """
        Se connecte à l'API Odoo en utilisant les identifiants fournis.
        
        Returns:
            True si l'authentification réussie, False sinon
        """
        try:
            logging.info(f"Tentative de connexion à {self.url} avec l'utilisateur {self.username}")
            endpoint = '/web/session/authenticate'
            
            payload = {
                "jsonrpc": "2.0",
                "params": {
                    "db": self.db,
                    "login": self.username,
                    "password": self.api_key
                }
            }
            
            response = self.session.post(
                f"{self.url}{endpoint}",
                data=json.dumps(payload),
                timeout=self.timeout
            )
            response.raise_for_status()
            
            result = response.json()
            if 'error' in result:
                error = result['error']
                logging.error(f"Erreur d'authentification: {error.get('message', 'Erreur inconnue')}")
                return False
                
            self.uid = result.get('result', {}).get('uid')
            if not self.uid:
                logging.error("Erreur: UID non trouvé dans la réponse d'authentification")
                return False
                
            logging.info(f"Authentification réussie. UID: {self.uid}")
            return True
            
        except requests.RequestException as e:
            logging.error(f"Erreur de connexion à l'API Odoo: {e}")
            return False
        except json.JSONDecodeError as e:
            logging.error(f"Erreur de décodage JSON: {e}")
            return False
        except Exception as e:
            logging.error(f"Erreur inattendue lors de l'authentification: {e}")
            return False
    
    def _rpc_call(self, endpoint: str, params: Dict[str, Any], retry_count: int = 0) -> Any:
        """
        Effectue un appel RPC à l'API Odoo.
        
        Args:
            endpoint: Point de terminaison de l'API
            params: Paramètres de l'appel
            retry_count: Nombre de tentatives actuelles (pour les nouvelles tentatives)
            
        Returns:
            Résultat de l'appel RPC ou None en cas d'erreur
        """
        if not self.uid and endpoint != '/web/session/authenticate':
            logging.warning("Tentative d'appel RPC sans être authentifié. Reconnexion...")
            if not self.login():
                logging.error("Échec de la reconnexion")
                return None
        
        try:
            payload = {
                "jsonrpc": "2.0",
                "params": params
            }
            
            response = self.session.post(
                f"{self.url}{endpoint}",
                data=json.dumps(payload),
                timeout=self.timeout
            )
            response.raise_for_status()
            
            result = response.json()
            if 'error' in result:
                error = result['error']
                error_msg = error.get('message', 'Erreur inconnue')
                error_data = error.get('data', {})
                error_name = error_data.get('name', 'UnknownError')
                logging.error(f"Erreur RPC: {error_name} - {error_msg}")
                
                # Gérer les erreurs d'authentification
                if "session expired" in error_msg or "Access denied" in error_msg:
                    if retry_count < self.max_retries:
                        logging.info("Session expirée, nouvelle tentative d'authentification...")
                        if self.login():
                            return self._rpc_call(endpoint, params, retry_count + 1)
                
                return None
                
            return result.get('result')
            
        except requests.RequestException as e:
            logging.error(f"Erreur de requête RPC: {e}")
            if retry_count < self.max_retries:
                logging.info(f"Nouvelle tentative ({retry_count + 1}/{self.max_retries})...")
                return self._rpc_call(endpoint, params, retry_count + 1)
            return None
        except json.JSONDecodeError as e:
            logging.error(f"Erreur de décodage JSON dans la réponse RPC: {e}")
            return None
        except Exception as e:
            logging.error(f"Erreur inattendue lors de l'appel RPC: {e}")
            return None
    
    def search_read(self, model: str, domain: list, fields: list, **kwargs) -> list:
        """
        Effectue une recherche et lecture sur le modèle spécifié.
        
        Args:
            model: Nom du modèle Odoo
            domain: Domaine de recherche (filtres)
            fields: Liste des champs à récupérer
            **kwargs: Arguments supplémentaires (limit, offset, etc.)
            
        Returns:
            Liste des enregistrements trouvés
        """
        params = {
            "model": model,
            "method": "search_read",
            "args": [domain, fields],
            "kwargs": kwargs
        }
        
        return self._rpc_call("/web/dataset/call_kw", params) or []
    
    def read(self, model: str, ids: list, fields: list) -> list:
        """
        Lit les enregistrements spécifiés par leurs IDs.
        
        Args:
            model: Nom du modèle Odoo
            ids: Liste des IDs des enregistrements à lire
            fields: Liste des champs à récupérer
            
        Returns:
            Liste des enregistrements lus
        """
        if not ids:
            return []
            
        params = {
            "model": model,
            "method": "read",
            "args": [ids, fields],
            "kwargs": {}
        }
        
        return self._rpc_call("/web/dataset/call_kw", params) or []
    
    def get_fields(self, model: str) -> Dict[str, Any]:
        """
        Récupère les informations sur les champs d'un modèle.
        
        Args:
            model: Nom du modèle Odoo
            
        Returns:
            Dictionnaire avec les informations sur les champs
        """
        params = {
            "model": model,
            "method": "fields_get",
            "args": [],
            "kwargs": {}
        }
        
        return self._rpc_call("/web/dataset/call_kw", params) or {}


# Fonctions d'aide pour centraliser l'authentification
def load_config(config_file: str = "config.json") -> Dict[str, Any]:
    """
    Charge le fichier de configuration.
    
    Args:
        config_file: Chemin vers le fichier de configuration
        
    Returns:
        Dictionnaire contenant les paramètres de configuration
    """
    try:
        with open(config_file, "r", encoding='utf-8') as f:
            return json.load(f)
    except Exception as e:
        logging.error(f"Erreur lors du chargement du fichier de configuration: {e}")
        return {}
    
def get_auth_manager(config_file: str = "config.json", force_new: bool = False) -> Optional[AuthManager]:
    """
    Obtient une instance unique du gestionnaire d'authentification.
    
    Args:
        config_file: Chemin vers le fichier de configuration
        force_new: Si True, force la création d'une nouvelle instance
        
    Returns:
        Instance du gestionnaire d'authentification ou None en cas d'erreur
    """
    global _auth_manager_instance
    
    # Si une instance existe et que force_new est False, retourner l'instance existante
    if _auth_manager_instance is not None and not force_new:
        return _auth_manager_instance
    
    # Charger la configuration
    config = load_config(config_file)
    
    # Extraire les informations de connexion
    odoo_config = config.get("odoo", {})
    url = odoo_config.get("url")
    db = odoo_config.get("db")
    username = odoo_config.get("username")
    api_key = odoo_config.get("api_key")
    
    if not all([url, db, username, api_key]):
        logging.error("Informations de connexion Odoo manquantes dans le fichier de configuration")
        return None
       
    # Créer une nouvelle instance
    try:
        auth_manager = AuthManager(
            url=url,
            db=db,
            username=username,
            api_key=api_key
        )
        
        # Tenter de se connecter
        if not auth_manager.login():
            logging.error("Échec de la connexion à l'API Odoo")
            return None
            
        # Stocker l'instance pour les appels futurs
        _auth_manager_instance = auth_manager
        return auth_manager
    
    except Exception as e:
        logging.exception(f"Erreur lors de l'initialisation du gestionnaire d'authentification: {e}")
        return None

def get_output_dir(config_file: str = "config.json", subdir: Optional[str] = None) -> str:
    """
    Obtient le répertoire de sortie à partir de la configuration.
    
    Args:
        config_file: Chemin vers le fichier de configuration
        subdir: Sous-répertoire à ajouter au chemin (optionnel)
        
    Returns:
        Chemin du répertoire de sortie
    """
    config = load_config(config_file)
    output_dir = config.get("output_dir", "output")
    
    if subdir:
        return os.path.join(output_dir, subdir)
    
    return output_dir