"""Watchdog DND P520 - Surveillance continue - version autonome.""" import base64 import hashlib import os 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) 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 depuis l'environnement, sans secret dans le code.""" base_url = os.environ.get('YEASTAR_BASE_URL', '').strip() username = os.environ.get('YEASTAR_USERNAME', '').strip() password = os.environ.get('YEASTAR_PASSWORD', '') explicitly_enabled = os.environ.get('YEASTAR_DND_ENABLED', '0') == '1' return { 'enabled': explicitly_enabled and bool(base_url and username and password), 'refresh_seconds': int(os.environ.get('YEASTAR_DND_REFRESH_SECONDS', '2')), 'reset_delay_seconds': int(os.environ.get('YEASTAR_DND_RESET_DELAY_SECONDS', '3')), 'excluded_extensions': [ value.strip() for value in os.environ.get('YEASTAR_DND_EXCLUDED_EXTENSIONS', '').split(',') if value.strip() ], 'base_url': base_url, 'username': username, 'password': 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() if not config['enabled']: log_dnd('[DND] Watchdog desactive ou configuration Yeastar incomplete') return True 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()