gmao/app_new/lib_ext/api_client.py

"""
Client API Yeastar P520
Authentification et accès à l'API REST du P520
"""
import requests
import hashlib
import base64
import time
import uuid
from typing import Optional


class YeastarClient:
    """Client pour l'API Yeastar P-Series (P520)"""
    
    def __init__(self, base_url: str, username: str, password: str):
        self.base_url = base_url.rstrip('/')
        self.username = username
        self.password = password
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Accept': 'application/json, text/plain, */*',
            'Content-Type': 'application/json;charset=UTF-8',
            'Origin': self.base_url,
            'Referer': f'{self.base_url}/login'
        })
        self._logged_in = False
        self._device_mark = None
    
    def _generate_device_mark(self) -> str:
        """Génère un identifiant de device unique"""
        timestamp = int(time.time() * 1000)
        random_str = uuid.uuid4().hex[:28]
        return f"{timestamp}-{random_str}"
    
    def _encode_password(self, password: str) -> str:
        """Encode le mot de passe: MD5 → Base64"""
        md5_hash = hashlib.md5(password.encode()).hexdigest()
        return base64.b64encode(md5_hash.encode()).decode()
    
    def login(self, language: str = "fr") -> dict:
        """Authentification auprès du P520"""
        self._device_mark = self._generate_device_mark()
        
        # Préparer les paramètres de l'URL
        params = {
            "username": self.username,
            "language": language,
            "supportcrx": "true",
            "linkus_devicemark": self._device_mark,
            "login_link_type": "all"
        }
        
        # Préparer le body
        body = {
            "username": self.username,
            "password": self._encode_password(self.password),
            "language": language,
            "supportcrx": True,
            "linkus_devicemark": self._device_mark,
            "login_link_type": "all"
        }
        
        # Requête de login
        url = f"{self.base_url}/api/v1.0/login"
        resp = self.session.post(url, params=params, json=body, verify=False, timeout=10)
        resp.raise_for_status()
        
        data = resp.json()
        if data.get('errcode') == 0:
            self._logged_in = True
            return {"success": True, "data": data}
        else:
            return {"success": False, "error": data.get('errmsg', 'Unknown error')}
    
    def get_extensions(self) -> list:
        """Récupère la liste des extensions"""
        if not self._logged_in:
            raise RuntimeError("Non connecté. Appeler login() d'abord.")
        
        url = f"{self.base_url}/api/v1.0/extension/searchwebclientext"
        resp = self.session.get(url, verify=False, timeout=10)
        resp.raise_for_status()
        
        data = resp.json()
        if data.get('errcode') == 0:
            return data.get('ext_list', [])
        return []
    
    def get_personal_info(self) -> dict:
        """Récupère les infos de l'extension connectée"""
        if not self._logged_in:
            raise RuntimeError("Non connecté. Appeler login() d'abord.")
        
        url = f"{self.base_url}/api/v1.0/extension/getpersonal"
        resp = self.session.get(url, verify=False, timeout=10)
        resp.raise_for_status()
        
        data = resp.json()
        if data.get('errcode') == 0:
            return data.get('extension', {})
        return {}
    
    def get_pbx_info(self) -> dict:
        """Récupère les informations du PBX"""
        if not self._logged_in:
            raise RuntimeError("Non connecté. Appeler login() d'abord.")
        
        url = f"{self.base_url}/api/v1.0/pbx/getinformation"
        resp = self.session.get(url, verify=False, timeout=10)
        resp.raise_for_status()
        
        data = resp.json()
        if data.get('errcode') == 0:
            return data.get('information', {})
        return {}
    
    def set_presence(self, status: str, extension_id: int = None) -> bool:
        """Change le statut de présence d'une extension
        
        Args:
            status: available, away, business_trip, do_not_disturb, launch, off_work
            extension_id: ID de l'extension (défaut: extension connectée)
        
        Returns:
            True si succès
        """
        if not self._logged_in:
            raise RuntimeError("Non connecté. Appeler login() d'abord.")
        
        valid_statuses = ['available', 'away', 'business_trip', 'do_not_disturb', 'launch', 'off_work']
        if status not in valid_statuses:
            raise ValueError(f"Statut invalide. Valeurs: {valid_statuses}")
        
        # Si pas d'ID fourni, utiliser l'extension connectée
        if extension_id is None:
            personal = self.get_personal_info()
            extension_id = personal.get('id')
            if not extension_id:
                raise RuntimeError("Impossible de déterminer l'ID de l'extension")
        
        url = f"{self.base_url}/api/v1.0/extension/update"
        body = {
            "id": extension_id,
            "presence_status": status
        }
        
        resp = self.session.post(url, json=body, verify=False, timeout=10)
        resp.raise_for_status()
        
        data = resp.json()
        return data.get('errcode') == 0
    
    def logout(self):
        """Déconnexion"""
        if self._logged_in:
            url = f"{self.base_url}/api/v1.0/logout"
            try:
                self.session.post(url, verify=False, timeout=5)
            except:
                pass
        self.session.close()
        self._logged_in = False
    
    def __enter__(self):
        self.login()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.logout()