"""Watchdog DND P520 - Surveillance continue - version autonome.""" import base64 import hashlib import sys import time import uuid import warnings import requests warnings.filterwarnings('ignore', message='Unverified HTTPS request') import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) _flask_app = None class YeastarClient: """Client minimal 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: timestamp = int(time.time() * 1000) random_str = uuid.uuid4().hex[:28] return f'{timestamp}-{random_str}' def _encode_password(self, password: str) -> str: md5_hash = hashlib.md5(password.encode()).hexdigest() return base64.b64encode(md5_hash.encode()).decode() def login(self, language: str='fr') -> dict: self._device_mark = self._generate_device_mark() params = {'username': self.username, 'language': language, 'supportcrx': 'true', 'linkus_devicemark': self._device_mark, 'login_link_type': 'all'} body = {'username': self.username, 'password': self._encode_password(self.password), 'language': language, 'supportcrx': True, 'linkus_devicemark': self._device_mark, 'login_link_type': 'all'} try: r = self.session.post(f'{self.base_url}/api/v1.0/login', params=params, json=body, verify=False, timeout=15) data = r.json() if r.status_code == 200 and data.get('errcode') == 0: self._logged_in = True return {'success': True, 'data': data} return {'success': False, 'error': data.get('errmsg', r.text)} except Exception as e: return {'success': False, 'error': str(e)} def logout(self) -> None: try: self.session.post(f'{self.base_url}/api/v1.0/logout', verify=False, timeout=10) except: pass self._logged_in = False def get_extensions(self) -> list: if not self._logged_in: raise RuntimeError('Non connecte') try: r = self.session.get(f'{self.base_url}/api/v1.0/extension/searchwebclientext', verify=False, timeout=15) data = r.json() if data.get('errcode') == 0: return data.get('ext_list', []) return [] except: return [] def get_personal_info(self) -> dict: if not self._logged_in: raise RuntimeError('Non connecte') try: r = self.session.get(f'{self.base_url}/api/v1.0/extension/getpersonal', verify=False, timeout=15) data = r.json() if data.get('errcode') == 0: return data.get('extension', {}) return {} except: return {} def set_presence(self, status: str, extension_id: int=None) -> bool: if not self._logged_in: raise RuntimeError('Non connecte') 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}') if extension_id is None: personal = self.get_personal_info() extension_id = personal.get('id') if not extension_id: raise RuntimeError("Impossible de determiner l'ID de l'extension") body = {'id': extension_id, 'presence_status': status} try: r = self.session.post(f'{self.base_url}/api/v1.0/extension/update', json=body, verify=False, timeout=15) data = r.json() return data.get('errcode') == 0 except: return False def load_config(): """Charge la configuration chiffree geree depuis l'interface GMAO.""" global _flask_app try: if _flask_app is None: from app_new import create_app _flask_app = create_app() from app_new.yeastar.models import YeastarConfig, YeastarDNDConfig with _flask_app.app_context(): connection = YeastarConfig.get_config() dnd = YeastarDNDConfig.query.first() if not connection or not dnd: raise RuntimeError('configuration absente') username = connection.get_username() password = connection.get_password() return { 'enabled': bool( dnd.is_enabled and connection.base_url and username and password ), 'refresh_seconds': max(dnd.refresh_seconds or 2, 1), 'reset_delay_seconds': max(dnd.reset_delay_seconds or 3, 0), 'excluded_extensions': dnd.excluded_list, 'base_url': connection.base_url, 'username': username, 'password': password, } except Exception: return { 'enabled': False, 'refresh_seconds': 300, 'reset_delay_seconds': 3, 'excluded_extensions': [], 'base_url': '', 'username': '', 'password': '', } def log_dnd(message, level='info'): from datetime import datetime timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') line = f"[{timestamp}] [DND] {message}" print(line) try: sys.path.insert(0, '/app') from app_new.watchdog_logger import log_watchdog log_watchdog('watchdog_dnd', message, level=level) except Exception: pass def run_watchdog(): """Boucle principale du watchdog.""" log_dnd('[DND] Demarrage watchdog') config = load_config() while not config['enabled']: log_dnd('[DND] Watchdog desactive ou configuration Yeastar incomplete') time.sleep(300) config = load_config() log_dnd(f"[DND] Config: enabled={config.get('enabled')}, url={config.get('base_url')}, user={config.get('username')}") client = YeastarClient(config['base_url'], config['username'], config['password']) result = client.login() if not result['success']: log_dnd(f"Login echoue: {result['error']}") return False log_dnd(f"Connecte a {config['base_url']}") prev_status = {} log_dnd('Verification etats DND existants...') extensions = client.get_extensions() for ext in extensions: ext_id = ext.get('ext_id') ext_num = ext.get('ext_num') name = ext.get('caller_id_name', '') status = ext.get('presence_status', '') prev_status[ext_id] = status if status == 'do_not_disturb': if str(ext_num) in config['excluded_extensions']: log_dnd(f'[IGNORE] {ext_num} - {name} (exclu)') else: log_dnd(f"[DND] {ext_num} - {name} -> wait {config['reset_delay_seconds']}s -> available") time.sleep(config['reset_delay_seconds']) try: client.set_presence('available', ext_id) log_dnd(f'[OK] {ext_num} remis en disponible') prev_status[ext_id] = 'available' except Exception as e: log_dnd(f'[ERREUR] {e}', level='error') log_dnd(f"Surveillance active - refresh={config['refresh_seconds']}s, delay={config['reset_delay_seconds']}s") last_excl = config['excluded_extensions'] try: while True: time.sleep(config['refresh_seconds']) config = load_config() if not config.get('enabled'): log_dnd('Watchdog desactive') break if config['excluded_extensions'] != last_excl: log_dnd(f"[CONFIG] Nouvelles exclus: {config['excluded_extensions']}") last_excl = config['excluded_extensions'] try: extensions = client.get_extensions() except Exception as e: log_dnd(f'Erreur API: {e}') time.sleep(5) try: client.logout() client.login() except: pass continue for ext in extensions: ext_id = ext.get('ext_id') ext_num = ext.get('ext_num') name = ext.get('caller_id_name', '') current_status = ext.get('presence_status', 'unknown') prev = prev_status.get(ext_id, 'unknown') if current_status == 'do_not_disturb' and prev != 'do_not_disturb': if str(ext_num) in config['excluded_extensions']: log_dnd(f'[IGNORE] {ext_num} - {name} (exclu)') else: log_dnd(f"[DND] {ext_num} - {name} -> wait {config['reset_delay_seconds']}s -> available") time.sleep(config['reset_delay_seconds']) try: ok = client.set_presence('available', ext_id) if ok: log_dnd(f'[OK] {ext_num} remis en disponible') prev_status[ext_id] = 'available' else: log_dnd(f'[ERREUR] {ext_num}', level='error') except Exception as e: log_dnd(f'[ERREUR] {e}', level='error') prev_status[ext_id] = current_status except KeyboardInterrupt: log_dnd('Arret watchdog') finally: client.logout() if __name__ == '__main__': run_watchdog()