380 lines
No EOL
13 KiB
Python
380 lines
No EOL
13 KiB
Python
"""
|
||
Status Système - GMAO Collège
|
||
Page de surveillance de l'état des systèmes avec vérification de connexion réelle.
|
||
API JSON pour rafraîchissement asynchrone.
|
||
"""
|
||
from flask import Blueprint, render_template, jsonify
|
||
from flask_login import login_required
|
||
from datetime import datetime, timezone
|
||
import os
|
||
import requests
|
||
|
||
status_bp = Blueprint('status', __name__, url_prefix='/status')
|
||
|
||
|
||
def _find_pids_by_cmd(pattern):
|
||
"""Trouve les PID des processus dont la cmdline contient pattern."""
|
||
pids = []
|
||
try:
|
||
for pid in os.listdir('/proc'):
|
||
if not pid.isdigit():
|
||
continue
|
||
try:
|
||
cmdline_path = f'/proc/{pid}/cmdline'
|
||
with open(cmdline_path, 'rb') as f:
|
||
cmdline = f.read().decode('utf-8', errors='replace').replace('\x00', ' ')
|
||
if pattern in cmdline and 'grep' not in cmdline:
|
||
pids.append(pid)
|
||
except (FileNotFoundError, PermissionError):
|
||
continue
|
||
except Exception:
|
||
pass
|
||
return pids
|
||
|
||
|
||
def _read_last_run(path):
|
||
"""Lit et formate la dernière exécution depuis un fichier."""
|
||
if not os.path.exists(path):
|
||
return None
|
||
try:
|
||
with open(path, 'r') as f:
|
||
last_run = f.read().strip()
|
||
dt = datetime.fromisoformat(last_run)
|
||
return dt.strftime('%d/%m/%Y %H:%M:%S')
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def check_database():
|
||
"""Vérifie la connexion à la base de données MariaDB."""
|
||
result = {
|
||
'name': 'MariaDB',
|
||
'icon': 'bi-database',
|
||
'configured': True,
|
||
'connected': False,
|
||
'message': ''
|
||
}
|
||
try:
|
||
from app_new.extensions import db
|
||
result_sql = db.session.execute(db.text('SELECT 1')).scalar()
|
||
if result_sql == 1:
|
||
result['connected'] = True
|
||
version = db.session.execute(db.text('SELECT VERSION()')).scalar()
|
||
result['message'] = f'Connecté — {version}'
|
||
else:
|
||
result['message'] = 'Réponse inattendue'
|
||
except Exception as e:
|
||
result['message'] = f'Erreur: {str(e)[:80]}'
|
||
return result
|
||
|
||
|
||
def check_outlook():
|
||
"""Vérifie la connexion Outlook/Microsoft Graph API."""
|
||
result = {
|
||
'name': 'Outlook / Microsoft',
|
||
'icon': 'bi-envelope',
|
||
'configured': False,
|
||
'connected': False,
|
||
'message': ''
|
||
}
|
||
try:
|
||
from app_new.outlook.models import OutlookAccount
|
||
from app_new.extensions import db
|
||
|
||
accounts = OutlookAccount.query.filter_by(is_active=True).all()
|
||
if not accounts:
|
||
result['message'] = 'Aucun compte configuré'
|
||
return result
|
||
|
||
result['configured'] = True
|
||
# Prendre le compte avec le refresh_token le plus récent
|
||
account = max(accounts, key=lambda a: a.last_connected_at or datetime.min.replace(tzinfo=timezone.utc))
|
||
result['message'] = f'{account.email}'
|
||
|
||
# Vérifier le statut de connexion stocké
|
||
if account.connection_status == 'active':
|
||
result['connected'] = True
|
||
if account.last_connected_at:
|
||
dt = account.last_connected_at
|
||
if dt.tzinfo is None:
|
||
dt = dt.replace(tzinfo=timezone.utc)
|
||
result['message'] += f' — Dernière sync: {dt.strftime("%d/%m/%Y %H:%M")}'
|
||
|
||
# Tester réellement l'API Graph
|
||
try:
|
||
from app_new.outlook.auth import get_access_token
|
||
token = get_access_token(account)
|
||
if token:
|
||
resp = requests.get(
|
||
'https://graph.microsoft.com/v1.0/me',
|
||
headers={'Authorization': f'Bearer {token}'},
|
||
timeout=10
|
||
)
|
||
if resp.status_code == 200:
|
||
me = resp.json()
|
||
result['connected'] = True
|
||
result['message'] += f' — API OK ({me.get("displayName", "")})'
|
||
elif resp.status_code == 401:
|
||
result['connected'] = False
|
||
result['message'] += ' — Token expiré, reconnexion nécessaire'
|
||
else:
|
||
result['connected'] = False
|
||
result['message'] += f' — API erreur {resp.status_code}'
|
||
else:
|
||
result['connected'] = False
|
||
result['message'] += ' — Impossible d\'obtenir un token'
|
||
except Exception as e:
|
||
result['message'] += f' — Test API: {str(e)[:40]}'
|
||
else:
|
||
result['connected'] = False
|
||
status_txt = account.connection_status or 'inconnu'
|
||
result['message'] += f' — Statut: {status_txt}'
|
||
|
||
# Tenter quand même de rafraîchir le token
|
||
try:
|
||
from app_new.outlook.auth import get_access_token
|
||
token = get_access_token(account)
|
||
if token:
|
||
result['connected'] = True
|
||
result['message'] += ' — Token rafraîchi automatiquement'
|
||
except Exception:
|
||
pass
|
||
except Exception as e:
|
||
result['message'] = f'Erreur: {str(e)[:80]}'
|
||
return result
|
||
|
||
|
||
def check_ent():
|
||
"""Vérifie la connexion ENT77."""
|
||
result = {
|
||
'name': 'ENT77',
|
||
'icon': 'bi-person-badge',
|
||
'configured': False,
|
||
'connected': False,
|
||
'message': ''
|
||
}
|
||
try:
|
||
from app_new.ent.models import EntCredential
|
||
from app_new.extensions import db
|
||
|
||
cred = EntCredential.query.filter_by(is_active=True).first()
|
||
if not cred:
|
||
result['message'] = 'Non configuré'
|
||
return result
|
||
|
||
result['configured'] = True
|
||
|
||
if cred.last_check:
|
||
dt = cred.last_check
|
||
if dt.tzinfo is None:
|
||
dt = dt.replace(tzinfo=timezone.utc)
|
||
result['message'] = f'Dernière vérification: {dt.strftime("%d/%m/%Y %H:%M")}'
|
||
|
||
# Tester l'accessibilité du serveur ENT
|
||
try:
|
||
resp = requests.get('https://ent77.seine-et-marne.fr', timeout=10, allow_redirects=True)
|
||
if resp.status_code in (200, 302, 301):
|
||
result['connected'] = True
|
||
result['message'] += ' — Serveur accessible'
|
||
else:
|
||
result['message'] += f' — HTTP {resp.status_code}'
|
||
except requests.exceptions.Timeout:
|
||
result['message'] += ' — Timeout'
|
||
except Exception as e:
|
||
result['message'] += f' — {str(e)[:40]}'
|
||
except Exception as e:
|
||
result['message'] = f'Erreur: {str(e)[:80]}'
|
||
return result
|
||
|
||
|
||
def check_pronote():
|
||
"""Vérifie la connexion PRONOTE."""
|
||
result = {
|
||
'name': 'PRONOTE',
|
||
'icon': 'bi-calendar-week',
|
||
'configured': False,
|
||
'connected': False,
|
||
'message': ''
|
||
}
|
||
try:
|
||
from app_new.pronote.models import PronoteSession
|
||
from app_new.extensions import db
|
||
|
||
session = PronoteSession.get()
|
||
if not session or not session.session_url:
|
||
result['message'] = 'Non configuré'
|
||
return result
|
||
|
||
result['configured'] = True
|
||
result['message'] = f'{session.username or "N/A"}'
|
||
|
||
# Tester l'URL PRONOTE
|
||
try:
|
||
resp = requests.get(session.session_url, timeout=10, allow_redirects=True)
|
||
if resp.status_code == 200:
|
||
result['connected'] = True
|
||
result['message'] += ' — Serveur accessible'
|
||
else:
|
||
result['message'] += f' — HTTP {resp.status_code}'
|
||
except requests.exceptions.Timeout:
|
||
result['message'] += ' — Timeout'
|
||
except Exception as e:
|
||
result['message'] += f' — {str(e)[:40]}'
|
||
|
||
if session.last_connection:
|
||
dt = session.last_connection
|
||
if dt.tzinfo is None:
|
||
dt = dt.replace(tzinfo=timezone.utc)
|
||
result['message'] += f', dernière connexion: {dt.strftime("%d/%m/%Y %H:%M")}'
|
||
except Exception as e:
|
||
result['message'] = f'Erreur: {str(e)[:80]}'
|
||
return result
|
||
|
||
|
||
def check_yeastar():
|
||
"""Vérifie la connexion Yeastar P520 avec test de login API."""
|
||
result = {
|
||
'name': 'Yeastar P520',
|
||
'icon': 'bi-telephone',
|
||
'configured': False,
|
||
'connected': False,
|
||
'message': ''
|
||
}
|
||
try:
|
||
from app_new.yeastar.models import YeastarConfig
|
||
from app_new.extensions import db
|
||
|
||
config = YeastarConfig.query.filter_by(is_active=True).first()
|
||
if not config or not config.base_url:
|
||
result['message'] = 'Non configuré'
|
||
return result
|
||
|
||
result['configured'] = True
|
||
result['message'] = f'{config.base_url}'
|
||
|
||
# Test de connexion API
|
||
try:
|
||
from app_new.lib_ext.api_client import YeastarClient
|
||
username = config.get_username()
|
||
password = config.get_password()
|
||
|
||
client = YeastarClient(config.base_url, username, password)
|
||
login_result = client.login()
|
||
|
||
if login_result.get('success'):
|
||
result['connected'] = True
|
||
try:
|
||
exts = client.get_extensions()
|
||
result['message'] += f' — Connecté, {len(exts)} extensions'
|
||
client.logout()
|
||
except Exception:
|
||
result['message'] += ' — Connecté (erreur liste extensions)'
|
||
else:
|
||
result['message'] += f' — Login échoué: {login_result.get("error", "inconnu")[:40]}'
|
||
except Exception as e:
|
||
result['message'] += f' — Erreur: {str(e)[:40]}'
|
||
except Exception as e:
|
||
result['message'] = f'Erreur: {str(e)[:80]}'
|
||
return result
|
||
|
||
|
||
def check_gmao_context():
|
||
"""Vérifie la configuration GMAO (auto-interprétation, sync PRONOTE)."""
|
||
result = {
|
||
'name': 'Contexte GMAO',
|
||
'icon': 'bi-gear',
|
||
'configured': False,
|
||
'connected': False,
|
||
'message': ''
|
||
}
|
||
try:
|
||
from app_new.outlook.models import GmaoContext
|
||
from app_new.extensions import db
|
||
|
||
ctx = GmaoContext.get_for_user(2)
|
||
if not ctx:
|
||
result['message'] = 'Non configuré'
|
||
return result
|
||
|
||
result['configured'] = True
|
||
result['connected'] = True
|
||
parts = []
|
||
if ctx.auto_interpret_enabled:
|
||
parts.append('Auto-interprétation ON')
|
||
else:
|
||
parts.append('Auto-interprétation OFF')
|
||
|
||
if ctx.pronote_sync_enabled:
|
||
parts.append(f'Sync PRONOTE ON (×{ctx.pronote_sync_interval or 360}min)')
|
||
else:
|
||
parts.append('Sync PRONOTE OFF')
|
||
|
||
result['message'] = ' | '.join(parts)
|
||
except Exception as e:
|
||
result['message'] = f'Erreur: {str(e)[:80]}'
|
||
return result
|
||
|
||
|
||
def check_watchdog(name, pattern, last_run_path, label_sync=''):
|
||
"""Vérifie si un watchdog est en cours d'exécution."""
|
||
result = {
|
||
'name': name,
|
||
'running': False,
|
||
'pid': None,
|
||
'last_run': None,
|
||
'message': ''
|
||
}
|
||
try:
|
||
pids = _find_pids_by_cmd(pattern)
|
||
if pids:
|
||
result['running'] = True
|
||
result['pid'] = pids[0]
|
||
result['last_run'] = _read_last_run(last_run_path)
|
||
msg = f'Actif (PID: {result["pid"]})'
|
||
if result['last_run']:
|
||
msg += f', {label_sync}: {result["last_run"]}'
|
||
result['message'] = msg
|
||
else:
|
||
result['message'] = 'Inactif'
|
||
except Exception as e:
|
||
result['message'] = f'Erreur: {str(e)[:30]}'
|
||
return result
|
||
|
||
|
||
@status_bp.route('/')
|
||
@login_required
|
||
def index():
|
||
"""Page de status du système — rendu HTML rapide, données chargées en async."""
|
||
return render_template('status/index.html')
|
||
|
||
|
||
@status_bp.route('/api/check')
|
||
@login_required
|
||
def api_check():
|
||
"""API JSON pour rafraîchissement asynchrone — vérifie les connexions en temps réel."""
|
||
systems = [
|
||
check_database(),
|
||
check_outlook(),
|
||
check_ent(),
|
||
check_pronote(),
|
||
check_yeastar(),
|
||
check_gmao_context(),
|
||
]
|
||
|
||
configured_count = sum(1 for s in systems if s.get('configured'))
|
||
connected_count = sum(1 for s in systems if s.get('connected'))
|
||
|
||
watchdogs = [
|
||
check_watchdog('Watchdog DND Yeastar', 'watchdog_dnd.py', '/tmp/dnd_last_run.txt', 'dernière action'),
|
||
check_watchdog('Watchdog GMAO Mail', 'gmao_watchdog.py', '/tmp/gmao_last_run.txt', 'dernière exécution'),
|
||
check_watchdog('Watchdog ENT', 'ent_watchdog.py', '/tmp/ent_last_run.txt', 'dernière exécution'),
|
||
check_watchdog('Watchdog PRONOTE', 'pronote_watchdog.py', '/tmp/pronote_last_run.txt', 'dernière sync'),
|
||
]
|
||
|
||
return jsonify({
|
||
'systems': systems,
|
||
'watchdogs': watchdogs,
|
||
'configured_count': configured_count,
|
||
'connected_count': connected_count,
|
||
'timestamp': datetime.now(timezone.utc).isoformat()
|
||
}) |