337 lines
11 KiB
Python
337 lines
11 KiB
Python
"""
|
|
Yeastar Routes - GMAO Collège
|
|
Surveillance téléphone P520
|
|
"""
|
|
from flask import Blueprint, render_template, jsonify, request, redirect, url_for, flash
|
|
from flask_login import login_required
|
|
from app_new.extensions import db
|
|
from app_new.core.system_security import system_admin_required
|
|
from app_new.yeastar.models import YeastarConfig, YeastarDNDConfig
|
|
from app_new.lib_ext.api_client import YeastarClient
|
|
from cryptography.fernet import InvalidToken
|
|
import subprocess
|
|
from datetime import datetime, timezone
|
|
|
|
yeastar_bp = Blueprint('yeastar', __name__, template_folder='templates')
|
|
|
|
|
|
def get_client():
|
|
"""Crée et retourne un client P520 connecté en utilisant la config DB."""
|
|
config = YeastarConfig.get_config()
|
|
|
|
if not config:
|
|
return None, "Configuration Yeastar non trouvée. Veuillez configurer les identifiants."
|
|
|
|
try:
|
|
username = config.get_username()
|
|
password = config.get_password()
|
|
except (InvalidToken, ValueError, RuntimeError):
|
|
return None, "Les identifiants Yeastar doivent etre saisis de nouveau."
|
|
|
|
if not all([config.base_url, username, password]):
|
|
return None, "Identifiants incomplets dans la configuration."
|
|
|
|
client = YeastarClient(config.base_url, username, password)
|
|
result = client.login()
|
|
|
|
if result['success']:
|
|
return client, None
|
|
return None, result.get('error')
|
|
|
|
|
|
@yeastar_bp.route('/')
|
|
@login_required
|
|
def index():
|
|
"""Dashboard surveillance P520."""
|
|
client, error = get_client()
|
|
|
|
extensions = []
|
|
pbx_info = {}
|
|
|
|
if client:
|
|
try:
|
|
extensions = client.get_extensions()
|
|
pbx_info = client.get_pbx_info()
|
|
client.logout()
|
|
except Exception as e:
|
|
error = str(e)
|
|
|
|
status_map = {
|
|
'available': 'Disponible',
|
|
'away': 'Absent',
|
|
'business_trip': 'En déplacement',
|
|
'do_not_disturb': 'Ne pas déranger',
|
|
'launch': 'Pause déjeuner',
|
|
'off_work': 'Hors service'
|
|
}
|
|
|
|
# Status de connexion réseau (status field)
|
|
# 0 = Enregistré/Connecté, 4 = Non enregistré/Non connecté
|
|
conn_map = {
|
|
0: ('Connecté', 'success'),
|
|
4: ('Non connecté', 'warning'),
|
|
}
|
|
|
|
ext_display = []
|
|
for ext in sorted(extensions, key=lambda x: int(x.get('ext_num', 0))):
|
|
status_code = ext.get('status', -1)
|
|
conn_info = conn_map.get(status_code, ('Inconnu', 'secondary'))
|
|
|
|
ext_display.append({
|
|
'num': ext['ext_num'],
|
|
'name': ext.get('caller_id_name', 'N/A'),
|
|
'status': ext.get('presence_status', 'unknown'),
|
|
'status_fr': status_map.get(ext.get('presence_status', ''), ext.get('presence_status', '')),
|
|
'status_code': status_code,
|
|
'conn_label': conn_info[0],
|
|
'conn_class': conn_info[1],
|
|
'ip': ext.get('ip_address', 'N/A'),
|
|
'device': ext.get('device', 'N/A')
|
|
})
|
|
|
|
# Configuration DND
|
|
dnd_config = YeastarDNDConfig.query.first()
|
|
if not dnd_config:
|
|
dnd_config = YeastarDNDConfig()
|
|
db.session.add(dnd_config)
|
|
db.session.commit()
|
|
|
|
# Configuration Yeastar
|
|
yeastar_config = YeastarConfig.get_config()
|
|
|
|
return render_template('yeastar/index.html',
|
|
extensions=ext_display,
|
|
pbx_info=pbx_info,
|
|
error=error,
|
|
dnd_config=dnd_config,
|
|
yeastar_config=yeastar_config)
|
|
|
|
|
|
@yeastar_bp.route('/config', methods=['GET', 'POST'])
|
|
@system_admin_required
|
|
def config_page():
|
|
"""Page de configuration des identifiants Yeastar."""
|
|
config = YeastarConfig.get_config()
|
|
|
|
if request.method == 'POST':
|
|
base_url = request.form.get('base_url', '').strip()
|
|
username = request.form.get('username', '').strip()
|
|
password = request.form.get('password', '').strip()
|
|
|
|
if not base_url.startswith(('https://', 'http://')):
|
|
flash("L'URL Yeastar doit commencer par https:// ou http://.", "danger")
|
|
return redirect(url_for('yeastar.config_page'))
|
|
|
|
if not all([base_url, username]):
|
|
flash("L'URL et le nom d'utilisateur sont obligatoires.", "danger")
|
|
return redirect(url_for('yeastar.config_page'))
|
|
|
|
# Créer ou mettre à jour la configuration
|
|
if not config:
|
|
config = YeastarConfig()
|
|
db.session.add(config)
|
|
|
|
config.base_url = base_url
|
|
|
|
# Chiffrer les identifiants
|
|
config.set_username(username)
|
|
if password:
|
|
config.set_password(password)
|
|
else:
|
|
try:
|
|
existing_password = config.get_password()
|
|
except (InvalidToken, ValueError, RuntimeError):
|
|
existing_password = None
|
|
if not existing_password:
|
|
flash(
|
|
"Le mot de passe est obligatoire pour cette configuration.",
|
|
"danger",
|
|
)
|
|
return redirect(url_for('yeastar.config_page'))
|
|
|
|
config.updated_at = datetime.now(timezone.utc)
|
|
db.session.commit()
|
|
|
|
flash("Configuration Yeastar sauvegardée avec succès.", "success")
|
|
return redirect(url_for('yeastar.config_page'))
|
|
|
|
# Préparer les données pour l'affichage (ne pas afficher le mot de passe)
|
|
display_config = None
|
|
password_is_valid = False
|
|
if config:
|
|
try:
|
|
username = config.get_username()
|
|
config.get_password()
|
|
password_is_valid = True
|
|
except (InvalidToken, ValueError, RuntimeError):
|
|
username = ''
|
|
display_config = {
|
|
'base_url': config.base_url,
|
|
'username': username,
|
|
'is_active': config.is_active,
|
|
'updated_at': config.updated_at,
|
|
'password_is_valid': password_is_valid,
|
|
}
|
|
|
|
return render_template('yeastar/config.html', config=display_config)
|
|
|
|
|
|
@yeastar_bp.route('/config/test', methods=['POST'])
|
|
@system_admin_required
|
|
def test_connection():
|
|
"""Teste la connexion Yeastar avec les identifiants actuels."""
|
|
client, error = get_client()
|
|
|
|
if error:
|
|
return jsonify({"success": False, "error": error})
|
|
|
|
try:
|
|
# Test simple : récupérer les extensions
|
|
extensions = client.get_extensions()
|
|
pbx_info = client.get_pbx_info()
|
|
client.logout()
|
|
|
|
return jsonify({
|
|
"success": True,
|
|
"extension_count": len(extensions),
|
|
"pbx_type": pbx_info.get('product_type', 'P520'),
|
|
"version": pbx_info.get('version', 'N/A')
|
|
})
|
|
except Exception as e:
|
|
return jsonify({"success": False, "error": str(e)})
|
|
|
|
|
|
@yeastar_bp.route('/dnd-config', methods=['GET', 'POST'])
|
|
@system_admin_required
|
|
def dnd_config_page():
|
|
"""Page de configuration du watchdog DND."""
|
|
config = YeastarDNDConfig.query.first()
|
|
if not config:
|
|
config = YeastarDNDConfig()
|
|
db.session.add(config)
|
|
db.session.commit()
|
|
|
|
if request.method == 'POST':
|
|
config.is_enabled = request.form.get('is_enabled') == 'on'
|
|
config.refresh_seconds = int(request.form.get('refresh_seconds', 1))
|
|
config.reset_delay_seconds = int(request.form.get('reset_delay_seconds', 3))
|
|
config.excluded_extensions = request.form.get('excluded_extensions', '')
|
|
config.updated_at = datetime.now(timezone.utc)
|
|
|
|
db.session.commit()
|
|
|
|
# Redémarrer le watchdog
|
|
try:
|
|
subprocess.run(['pkill', '-f', 'watchdog_dnd'], capture_output=True)
|
|
except:
|
|
pass
|
|
if config.is_enabled:
|
|
script_path = '/app/watchdog_dnd.py'
|
|
subprocess.Popen(['python3', '/app/watchdog_gate.py', 'yeastar', script_path],
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
|
|
flash("Configuration DND sauvegardée", "success")
|
|
return redirect(url_for('yeastar.dnd_config_page'))
|
|
|
|
return render_template('yeastar/dnd_config.html', config=config)
|
|
|
|
|
|
@yeastar_bp.route('/api/extensions')
|
|
@login_required
|
|
def api_extensions():
|
|
"""API JSON pour les statuts des extensions."""
|
|
client, error = get_client()
|
|
|
|
if error:
|
|
return jsonify({"error": error}), 500
|
|
|
|
try:
|
|
extensions = client.get_extensions()
|
|
pbx_info = client.get_pbx_info()
|
|
client.logout()
|
|
|
|
status_map = {
|
|
'available': 'Disponible',
|
|
'away': 'Absent',
|
|
'business_trip': 'En déplacement',
|
|
'do_not_disturb': 'Ne pas déranger',
|
|
'launch': 'Pause déjeuner',
|
|
'off_work': 'Hors service'
|
|
}
|
|
|
|
# Status de connexion réseau
|
|
conn_map = {
|
|
0: ('Connecté', 'success'),
|
|
4: ('Non connecté', 'warning'),
|
|
}
|
|
|
|
ext_list = []
|
|
for ext in sorted(extensions, key=lambda x: int(x.get('ext_num', 0))):
|
|
status_code = ext.get('status', -1)
|
|
conn_info = conn_map.get(status_code, ('Inconnu', 'secondary'))
|
|
|
|
ext_list.append({
|
|
'num': ext['ext_num'],
|
|
'name': ext.get('caller_id_name', 'N/A'),
|
|
'status': ext.get('presence_status', 'unknown'),
|
|
'status_fr': status_map.get(ext.get('presence_status', ''), ext.get('presence_status', '')),
|
|
'status_code': status_code,
|
|
'conn_label': conn_info[0],
|
|
'conn_class': conn_info[1],
|
|
'ip': ext.get('ip_address', 'N/A'),
|
|
'device': ext.get('device', 'N/A')
|
|
})
|
|
|
|
return jsonify({
|
|
"pbx": pbx_info.get('product_type', 'P520'),
|
|
"version": pbx_info.get('version', 'N/A'),
|
|
"extensions": ext_list
|
|
})
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
|
|
@yeastar_bp.route('/api/watchdog/toggle', methods=['POST'])
|
|
@system_admin_required
|
|
def api_watchdog_toggle():
|
|
"""Active ou désactive le watchdog."""
|
|
config = YeastarDNDConfig.query.first()
|
|
if config:
|
|
config.is_enabled = request.json.get('enabled', False)
|
|
db.session.commit()
|
|
return jsonify({"success": True, "enabled": config.is_enabled})
|
|
return jsonify({"error": "Config non trouvée"}), 404
|
|
|
|
|
|
@yeastar_bp.route('/api/watchdog/status')
|
|
@login_required
|
|
def api_watchdog_status():
|
|
"""Statut du watchdog."""
|
|
config = YeastarDNDConfig.query.first()
|
|
if config:
|
|
return jsonify({
|
|
"enabled": config.is_enabled,
|
|
"refresh_seconds": config.refresh_seconds,
|
|
"reset_delay_seconds": config.reset_delay_seconds,
|
|
"excluded_extensions": config.excluded_extensions
|
|
})
|
|
return jsonify({"error": "Config non trouvée"}), 404
|
|
|
|
|
|
@yeastar_bp.route('/api/watchdog/restart', methods=['POST'])
|
|
@system_admin_required
|
|
def api_watchdog_restart():
|
|
"""Redémarre le watchdog (tue et relance)."""
|
|
# Tuer les processus existants
|
|
try:
|
|
subprocess.run(['pkill', '-f', 'watchdog_dnd'], capture_output=True)
|
|
except:
|
|
pass
|
|
|
|
# Relancer
|
|
script_path = '/app/watchdog_dnd.py'
|
|
subprocess.Popen(['python3', '/app/watchdog_gate.py', 'yeastar', script_path],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL)
|
|
|
|
return jsonify({"success": True, "message": "Watchdog relancé"})
|