llm_ticket3/odoo/auth_manager.py

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

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 {}