gmao/watchdog_dnd.py

"""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()
    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()