#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
NeuroPulse Monitor Pro v2.0 - Version Simplifiée
Application Flask pour la surveillance système
"""
import os
import sys
import json
import time
import sqlite3
import logging
import subprocess
import threading
from datetime import datetime, timedelta
from functools import wraps
from collections import defaultdict, deque
from flask import Flask, render_template, request, jsonify, session, redirect, url_for, flash
from flask_cors import CORS
from werkzeug.security import generate_password_hash, check_password_hash
# Import du module de monitoring
try:
from utils.monitoring import SystemMonitor, get_system_metrics
except ImportError:
print("ATTENTION: Module monitoring non trouvé, utilisation de données fictives")
def get_system_metrics():
return {
'cpu': 45.2,
'ram': 67.8,
'disk': 23.1,
'network': {'kb_in_per_sec': 1024.5, 'kb_out_per_sec': 512.3},
'services': [],
'system': {'hostname': 'localhost'},
'anomalies': []
}
# Configuration Flask
app = Flask(__name__)
CORS(app)
# Configuration par défaut
DEFAULT_CONFIG = {
"version": "2.0",
"application": {
"name": "NeuroPulse Monitor Pro",
"port": 5000,
"host": "0.0.0.0",
"debug": True,
"secret_key": "neuropulse-secret-key-2024"
},
"monitoring": {
"refresh_interval": 3000,
"metrics_retention_days": 30,
"enable_real_time_alerts": True
},
"alert_thresholds": {
"cpu": {"warning": 70, "critical": 90},
"ram": {"warning": 80, "critical": 95},
"disk": {"warning": 85, "critical": 95}
},
"services_to_monitor": [
{
"name": "apache2",
"title": "Apache HTTP Server",
"port": 80,
"enabled": True,
"criticality": "Critique"
},
{
"name": "nginx",
"title": "Nginx Web Server",
"port": 80,
"enabled": True,
"criticality": "Critique"
}
],
"notifications": {
"email": {"enabled": False},
"slack": {"enabled": False}
},
"backup": {
"enabled": True,
"schedule": "02:00",
"retention_days": 30,
"backup_directory": "/var/www/html/backups"
}
}
# Variables globales
config = DEFAULT_CONFIG.copy()
metrics_cache = {"timestamp": 0, "data": {}}
alert_history = deque(maxlen=1000)
command_history = deque(maxlen=100)
# Paths configurables
BASE_PATH = "/var/www/html"
CONFIG_PATH = os.path.join(BASE_PATH, "config", "config.json")
DB_PATH = os.path.join(BASE_PATH, "config", "neuropulse.db")
LOG_PATH = os.path.join(BASE_PATH, "logs", "app.log")
# Configuration des logs
os.makedirs(os.path.dirname(LOG_PATH), exist_ok=True)
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(LOG_PATH),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
# Commandes interdites pour sécurité
FORBIDDEN_COMMANDS = [
'rm -rf', 'mkfs', 'dd if=', 'shutdown', 'reboot', 'halt',
'passwd', 'userdel', 'fdisk', 'parted', 'crontab -r',
'init 0', 'init 6', 'poweroff', 'systemctl poweroff',
'systemctl reboot', 'format', 'del /q', 'deltree'
]
def load_config():
"""Charger la configuration depuis le fichier"""
global config
try:
if os.path.exists(CONFIG_PATH):
with open(CONFIG_PATH, 'r', encoding='utf-8') as f:
loaded_config = json.load(f)
config.update(loaded_config)
logger.info(f"Configuration chargée depuis {CONFIG_PATH}")
else:
# Créer le fichier de config par défaut
os.makedirs(os.path.dirname(CONFIG_PATH), exist_ok=True)
with open(CONFIG_PATH, 'w', encoding='utf-8') as f:
json.dump(config, f, indent=2, ensure_ascii=False)
logger.info(f"Configuration par défaut créée: {CONFIG_PATH}")
except Exception as e:
logger.error(f"Erreur lors du chargement de la configuration: {e}")
config = DEFAULT_CONFIG.copy()
def save_config():
"""Sauvegarder la configuration"""
try:
os.makedirs(os.path.dirname(CONFIG_PATH), exist_ok=True)
with open(CONFIG_PATH, 'w', encoding='utf-8') as f:
json.dump(config, f, indent=2, ensure_ascii=False)
logger.info("Configuration sauvegardée")
return True
except Exception as e:
logger.error(f"Erreur lors de la sauvegarde: {e}")
return False
def init_database():
"""Initialiser la base de données SQLite"""
try:
os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
# Table des utilisateurs
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
password_hash TEXT NOT NULL,
email TEXT,
role TEXT DEFAULT 'admin',
last_login TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Table des tickets
cursor.execute('''
CREATE TABLE IF NOT EXISTS tickets (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
description TEXT,
severity TEXT DEFAULT 'medium',
status TEXT DEFAULT 'open',
service TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
resolved_at TIMESTAMP,
auto_created BOOLEAN DEFAULT 0
)
''')
# Table des métriques historiques
cursor.execute('''
CREATE TABLE IF NOT EXISTS metrics_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
cpu_percent REAL,
memory_percent REAL,
disk_percent REAL,
network_in REAL,
network_out REAL
)
''')
# Table des commandes exécutées
cursor.execute('''
CREATE TABLE IF NOT EXISTS command_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
command TEXT NOT NULL,
user TEXT,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
success BOOLEAN,
output TEXT,
error TEXT
)
''')
# Créer l'utilisateur admin par défaut
cursor.execute("SELECT COUNT(*) FROM users WHERE username = 'admin'")
if cursor.fetchone()[0] == 0:
admin_hash = generate_password_hash('admin')
cursor.execute(
"INSERT INTO users (username, password_hash, email, role) VALUES (?, ?, ?, ?)",
('admin', admin_hash, 'admin@neuropulse.local', 'admin')
)
logger.info("Utilisateur admin créé avec mot de passe par défaut")
conn.commit()
conn.close()
logger.info(f"Base de données initialisée: {DB_PATH}")
except Exception as e:
logger.error(f"Erreur lors de l'initialisation de la base de données: {e}")
def get_db():
"""Obtenir une connexion à la base de données"""
return sqlite3.connect(DB_PATH)
def login_required(f):
"""Décorateur simple pour vérifier l'authentification"""
@wraps(f)
def decorated_function(*args, **kwargs):
if 'user_id' not in session or 'username' not in session:
if request.path.startswith('/api/'):
return jsonify({'error': 'Authentification requise'}), 401
return redirect(url_for('login'))
return f(*args, **kwargs)
return decorated_function
def is_command_forbidden(command):
"""Vérifier si une commande est interdite"""
command_lower = command.lower().strip()
for forbidden in FORBIDDEN_COMMANDS:
if forbidden in command_lower:
return True
return False
def execute_command(command, user='unknown'):
"""Exécuter une commande système de manière sécurisée"""
if is_command_forbidden(command):
return {
'success': False,
'error': 'Commande interdite pour des raisons de sécurité',
'output': '',
'return_code': 1
}
try:
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=30,
cwd='/tmp'
)
success = result.returncode == 0
# Enregistrer dans l'historique
try:
conn = get_db()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO command_history (command, user, success, output, error) VALUES (?, ?, ?, ?, ?)",
(command, user, success, result.stdout, result.stderr)
)
conn.commit()
conn.close()
except Exception as e:
logger.error(f"Erreur lors de l'enregistrement de la commande: {e}")
return {
'success': success,
'output': result.stdout,
'error': result.stderr,
'return_code': result.returncode
}
except subprocess.TimeoutExpired:
return {
'success': False,
'error': 'Commande timeout (>30s)',
'output': '',
'return_code': 124
}
except Exception as e:
return {
'success': False,
'error': str(e),
'output': '',
'return_code': 1
}
def create_ticket(title, description, severity='medium', service='System', auto_created=False):
"""Créer un nouveau ticket"""
try:
conn = get_db()
cursor = conn.cursor()
# Vérifier si un ticket similaire existe déjà (pour éviter les doublons)
if auto_created:
cursor.execute(
"SELECT id FROM tickets WHERE title = ? AND status = 'open' AND DATE(created_at) = DATE('now')",
(title,)
)
if cursor.fetchone():
conn.close()
return None # Ticket déjà existant
cursor.execute(
"INSERT INTO tickets (title, description, severity, service, auto_created) VALUES (?, ?, ?, ?, ?)",
(title, description, severity, service, auto_created)
)
ticket_id = cursor.lastrowid
conn.commit()
conn.close()
logger.info(f"Ticket créé: #{ticket_id} - {title}")
return ticket_id
except Exception as e:
logger.error(f"Erreur lors de la création du ticket: {e}")
return None
def get_cached_metrics():
"""Obtenir les métriques avec cache et debug amélioré"""
global metrics_cache
current_time = time.time()
# Cache de 3 secondes
if current_time - metrics_cache['timestamp'] > 3:
try:
logger.info("Collecte des métriques en cours...")
# Essayer d'utiliser le module monitoring personnalisé
try:
from utils.monitoring import get_system_metrics
metrics_data = get_system_metrics()
logger.info(f"Métriques collectées via module personnalisé: CPU={metrics_data.get('cpu', 'N/A')}, RAM={metrics_data.get('ram', 'N/A')}")
except ImportError as e:
logger.warning(f"Module monitoring personnalisé non disponible: {e}")
# Fallback sur psutil direct
metrics_data = get_system_metrics_fallback()
logger.info(f"Métriques collectées via fallback: CPU={metrics_data.get('cpu', 'N/A')}, RAM={metrics_data.get('ram', 'N/A')}")
metrics_cache['data'] = metrics_data
metrics_cache['timestamp'] = current_time
except Exception as e:
logger.error(f"Erreur lors de la collecte des métriques: {e}")
# Utiliser des données de test en cas d'erreur
metrics_cache['data'] = get_test_metrics()
metrics_cache['timestamp'] = current_time
return metrics_cache['data']
def get_system_metrics_fallback():
"""Collecte de métriques de secours utilisant psutil directement"""
try:
import psutil
import socket
import time
logger.info("Utilisation du fallback psutil pour les métriques")
# CPU avec un délai pour avoir une mesure précise
cpu_percent = psutil.cpu_percent(interval=1.0) # Augmenter l'intervalle
if cpu_percent == 0:
# Fallback si cpu_percent retourne 0
cpu_times = psutil.cpu_times()
total_time = sum([cpu_times.user, cpu_times.system, cpu_times.idle])
if total_time > 0:
cpu_percent = ((cpu_times.user + cpu_times.system) / total_time) * 100
logger.info(f"CPU collecté: {cpu_percent}%")
# Mémoire
memory = psutil.virtual_memory()
memory_percent = memory.percent
logger.info(f"RAM collectée: {memory_percent}%")
# Disque - Utiliser la partition racine et ignorer les loop devices
disk_percent = 0
try:
# Chercher la partition principale (/, /home, ou plus grande partition)
partitions = psutil.disk_partitions()
main_partition = None
max_size = 0
for partition in partitions:
# Ignorer les devices virtuels
if any(x in partition.device for x in ['/dev/loop', '/dev/snap', 'tmpfs', 'devtmpfs']):
continue
try:
usage = psutil.disk_usage(partition.mountpoint)
# Priorité à la partition racine
if partition.mountpoint == '/':
main_partition = partition
break
# Sinon, prendre la plus grande partition
elif usage.total > max_size:
max_size = usage.total
main_partition = partition
except (PermissionError, FileNotFoundError):
continue
if main_partition:
disk_usage = psutil.disk_usage(main_partition.mountpoint)
disk_percent = (disk_usage.used / disk_usage.total) * 100
logger.info(f"Disque principal ({main_partition.mountpoint}): {disk_percent:.1f}%")
else:
# Fallback sur /
disk_usage = psutil.disk_usage('/')
disk_percent = (disk_usage.used / disk_usage.total) * 100
logger.info(f"Disque fallback (/): {disk_percent:.1f}%")
except Exception as e:
logger.warning(f"Erreur calcul disque: {e}")
disk_percent = 0
# Réseau avec calcul de vitesse
net_io = psutil.net_io_counters()
network_speeds = {'kb_in_per_sec': 0, 'kb_out_per_sec': 0}
if net_io:
# Utiliser une approche simple de calcul de vitesse
current_time = time.time()
# Stocker les données précédentes pour calculer la vitesse
if not hasattr(get_system_metrics_fallback, 'last_net_data'):
get_system_metrics_fallback.last_net_data = {
'bytes_recv': net_io.bytes_recv,
'bytes_sent': net_io.bytes_sent,
'timestamp': current_time
}
else:
time_diff = current_time - get_system_metrics_fallback.last_net_data['timestamp']
if time_diff > 0:
bytes_recv_diff = net_io.bytes_recv - get_system_metrics_fallback.last_net_data['bytes_recv']
bytes_sent_diff = net_io.bytes_sent - get_system_metrics_fallback.last_net_data['bytes_sent']
network_speeds['kb_in_per_sec'] = max(0, bytes_recv_diff / time_diff / 1024)
network_speeds['kb_out_per_sec'] = max(0, bytes_sent_diff / time_diff / 1024)
# Mettre à jour les données pour le prochain calcul
get_system_metrics_fallback.last_net_data = {
'bytes_recv': net_io.bytes_recv,
'bytes_sent': net_io.bytes_sent,
'timestamp': current_time
}
logger.info(f"Réseau: ↓{network_speeds['kb_in_per_sec']:.1f} KB/s ↑{network_speeds['kb_out_per_sec']:.1f} KB/s")
# Services (basique)
services = []
for service_config in config.get('services_to_monitor', []):
if not service_config.get('enabled', True):
continue
service_name = service_config['name']
try:
# Test simple de port si disponible
port = service_config.get('port')
active = False
if port:
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(2)
result = sock.connect_ex(('localhost', port))
active = result == 0
sock.close()
except:
pass
else:
# Test avec systemctl si pas de port
try:
result = subprocess.run(
['systemctl', 'is-active', service_name],
capture_output=True, text=True, timeout=3
)
active = result.stdout.strip() == 'active'
except:
pass
services.append({
'name': service_name,
'title': service_config.get('title', service_name),
'active': active,
'port': port,
'criticality': service_config.get('criticality', 'Moyenne')
})
logger.info(f"Service {service_name}: {'Actif' if active else 'Inactif'}")
except Exception as e:
logger.warning(f"Erreur test service {service_name}: {e}")
# Générer uniquement les alertes légitimes (pas de loop devices)
anomalies = []
# Alerte CPU
if cpu_percent > 90:
anomalies.append({
'type': 'cpu_critical',
'severity': 'critical',
'message': f'CPU usage critically high: {cpu_percent:.1f}%',
'timestamp': datetime.now().isoformat()
})
elif cpu_percent > 70:
anomalies.append({
'type': 'cpu_warning',
'severity': 'warning',
'message': f'CPU usage high: {cpu_percent:.1f}%',
'timestamp': datetime.now().isoformat()
})
# Alerte RAM
if memory_percent > 95:
anomalies.append({
'type': 'memory_critical',
'severity': 'critical',
'message': f'Memory usage critically high: {memory_percent:.1f}%',
'timestamp': datetime.now().isoformat()
})
elif memory_percent > 80:
anomalies.append({
'type': 'memory_warning',
'severity': 'warning',
'message': f'Memory usage high: {memory_percent:.1f}%',
'timestamp': datetime.now().isoformat()
})
# Alerte disque principal uniquement
if disk_percent > 95:
anomalies.append({
'type': 'disk_critical',
'severity': 'critical',
'message': f'Main disk usage critically high: {disk_percent:.1f}%',
'timestamp': datetime.now().isoformat()
})
elif disk_percent > 85:
anomalies.append({
'type': 'disk_warning',
'severity': 'warning',
'message': f'Main disk usage high: {disk_percent:.1f}%',
'timestamp': datetime.now().isoformat()
})
metrics = {
'cpu': round(cpu_percent, 1),
'ram': round(memory_percent, 1),
'disk': round(disk_percent, 1),
'network': {
'kb_in_per_sec': round(network_speeds['kb_in_per_sec'], 1),
'kb_out_per_sec': round(network_speeds['kb_out_per_sec'], 1),
'bytes_sent': net_io.bytes_sent if net_io else 0,
'bytes_recv': net_io.bytes_recv if net_io else 0,
'speeds': network_speeds,
'global': {
'bytes_sent': net_io.bytes_sent if net_io else 0,
'bytes_recv': net_io.bytes_recv if net_io else 0,
'packets_sent': net_io.packets_sent if net_io else 0,
'packets_recv': net_io.packets_recv if net_io else 0
} if net_io else {}
},
'services': services,
'system': {
'hostname': socket.gethostname(),
'uptime': time.time() - psutil.boot_time(),
'boot_time': datetime.fromtimestamp(psutil.boot_time()).isoformat(),
'platform': {
'system': 'linux',
'machine': 'x86_64',
'release': 'ubuntu'
}
},
'anomalies': anomalies,
'timestamp': datetime.now().isoformat(),
'source': 'fallback_psutil_corrected'
}
logger.info(f"Métriques fallback corrigées: CPU={metrics['cpu']}%, RAM={metrics['ram']}%, Disk={metrics['disk']}%")
return metrics
except Exception as e:
logger.error(f"Erreur dans le fallback psutil corrigé: {e}")
return get_test_metrics()
def get_test_metrics():
"""Générer des métriques de test pour le debug"""
import random
logger.warning("Utilisation de métriques de test")
return {
'cpu': round(random.uniform(20, 80), 1),
'ram': round(random.uniform(30, 70), 1),
'disk': round(random.uniform(15, 45), 1),
'network': {
'kb_in_per_sec': round(random.uniform(100, 1000), 1),
'kb_out_per_sec': round(random.uniform(50, 500), 1),
'bytes_sent': random.randint(1000000, 10000000),
'bytes_recv': random.randint(1000000, 10000000)
},
'services': [
{
'name': 'test-service',
'title': 'Service de Test',
'active': True,
'port': 80,
'criticality': 'Élevée'
}
],
'system': {
'hostname': 'neuropulse-test',
'uptime': 86400
},
'anomalies': [],
'timestamp': datetime.now().isoformat(),
'source': 'test_data'
}
# ========================
# ROUTES PRINCIPALES
# ========================
@app.route('/')
def index():
"""Page d'accueil - redirige vers dashboard si connecté, sinon login"""
if 'user_id' in session:
return redirect(url_for('dashboard'))
return redirect(url_for('login'))
@app.route('/login', methods=['GET', 'POST'])
def login():
"""Page de connexion"""
if request.method == 'POST':
username = request.form.get('username', '').strip()
password = request.form.get('password', '').strip()
if not username or not password:
flash('Nom d\'utilisateur et mot de passe requis', 'error')
return render_template('login.html')
try:
conn = get_db()
cursor = conn.cursor()
cursor.execute("SELECT id, password_hash FROM users WHERE username = ?", (username,))
user = cursor.fetchone()
if user and check_password_hash(user[1], password):
# Connexion réussie
session['user_id'] = user[0]
session['username'] = username
session.permanent = True
# Mettre à jour la dernière connexion
cursor.execute("UPDATE users SET last_login = CURRENT_TIMESTAMP WHERE id = ?", (user[0],))
conn.commit()
conn.close()
logger.info(f"Connexion réussie pour l'utilisateur: {username}")
flash(f'Bienvenue {username} !', 'success')
return redirect(url_for('dashboard'))
else:
conn.close()
flash('Identifiants invalides', 'error')
logger.warning(f"Tentative de connexion échouée pour: {username}")
except Exception as e:
logger.error(f"Erreur lors de la connexion: {e}")
flash('Erreur interne', 'error')
return render_template('login.html')
@app.route('/logout')
def logout():
"""Déconnexion"""
username = session.get('username', 'inconnu')
session.clear()
logger.info(f"Déconnexion de l'utilisateur: {username}")
flash('Vous avez été déconnecté', 'info')
return redirect(url_for('login'))
@app.route('/dashboard')
@login_required
def dashboard():
"""Dashboard principal"""
try:
metrics = get_cached_metrics()
# Obtenir les tickets récents
conn = get_db()
cursor = conn.cursor()
cursor.execute(
"SELECT id, title, severity, status, created_at FROM tickets ORDER BY created_at DESC LIMIT 5"
)
recent_tickets = cursor.fetchall()
conn.close()
return render_template('dashboard.html',
user=session['username'],
metrics=metrics,
config=config,
tickets=recent_tickets)
except Exception as e:
logger.error(f"Erreur dashboard: {e}")
return render_template('error.html', error=str(e)), 500
@app.route('/dashboard_debug')
@login_required
def dashboard_debug():
"""Dashboard de debug avec diagnostic complet"""
try:
metrics = get_cached_metrics()
# Obtenir quelques tickets pour test
conn = get_db()
cursor = conn.cursor()
cursor.execute(
"SELECT id, title, severity, status, created_at FROM tickets ORDER BY created_at DESC LIMIT 3"
)
recent_tickets = cursor.fetchall()
conn.close()
return render_template('dashboard_debug.html',
user=session['username'],
metrics=metrics,
config=config,
tickets=recent_tickets)
except Exception as e:
logger.error(f"Erreur dashboard debug: {e}")
return f"""
Erreur Dashboard Debug
❌ Erreur Dashboard Debug
Erreur: {str(e)}
Utilisateur: {session.get('username', 'Non défini')}
← Retour Login
🐛 Debug App
""", 500
@app.route('/api/debug/monitoring')
@login_required
def api_debug_monitoring():
"""API de debug pour tester le module monitoring"""
try:
debug_info = {
'module_loaded': False,
'functions_available': False,
'test_data': {},
'error': None,
'import_path': 'utils.monitoring'
}
try:
# Test d'import du module
from utils.monitoring import SystemMonitor, get_system_metrics
debug_info['module_loaded'] = True
debug_info['functions_available'] = True
# Test de la fonction get_system_metrics
test_data = get_system_metrics()
debug_info['test_data'] = test_data
# Test détaillé de psutil
import psutil
debug_info['psutil_test'] = {
'cpu_percent': psutil.cpu_percent(interval=0.1),
'memory_percent': psutil.virtual_memory().percent,
'disk_usage': psutil.disk_usage('/').percent if hasattr(psutil.disk_usage('/'), 'percent') else 'Error',
'boot_time': psutil.boot_time()
}
except ImportError as e:
debug_info['error'] = f'Import Error: {str(e)}'
debug_info['module_loaded'] = False
# Test psutil directement
try:
import psutil
debug_info['psutil_direct'] = {
'available': True,
'cpu': psutil.cpu_percent(interval=0.1),
'memory': psutil.virtual_memory().percent,
'disk': psutil.disk_usage('/').percent
}
debug_info['test_data'] = {
'cpu': psutil.cpu_percent(interval=0.1),
'ram': psutil.virtual_memory().percent,
'disk': psutil.disk_usage('/').percent,
'network': {
'kb_in_per_sec': 0,
'kb_out_per_sec': 0
},
'services': [],
'system': {'hostname': 'localhost'},
'anomalies': []
}
except Exception as psutil_error:
debug_info['psutil_direct'] = {'available': False, 'error': str(psutil_error)}
except Exception as e:
debug_info['error'] = f'Runtime Error: {str(e)}'
return jsonify(debug_info)
except Exception as e:
logger.error(f"Erreur debug monitoring: {e}")
return jsonify({
'module_loaded': False,
'functions_available': False,
'test_data': {},
'error': str(e)
}), 500
@app.route('/metrics')
@login_required
def metrics_page():
"""Page des métriques détaillées"""
return render_template('metrics.html',
user=session['username'],
config=config)
@app.route('/services')
@login_required
def services_page():
"""Page de gestion des services"""
try:
metrics = get_cached_metrics()
services = metrics.get('services', [])
return render_template('services.html',
user=session['username'],
services=services,
config=config)
except Exception as e:
logger.error(f"Erreur page services: {e}")
return render_template('error.html', error=str(e)), 500
@app.route('/tickets')
@login_required
def tickets_page():
"""Page de gestion des tickets"""
try:
conn = get_db()
cursor = conn.cursor()
# Filtres
status_filter = request.args.get('status', 'all')
severity_filter = request.args.get('severity', 'all')
query = "SELECT id, title, description, severity, status, service, created_at, resolved_at FROM tickets"
params = []
conditions = []
if status_filter != 'all':
conditions.append("status = ?")
params.append(status_filter)
if severity_filter != 'all':
conditions.append("severity = ?")
params.append(severity_filter)
if conditions:
query += " WHERE " + " AND ".join(conditions)
query += " ORDER BY created_at DESC LIMIT 100"
cursor.execute(query, params)
tickets = cursor.fetchall()
conn.close()
return render_template('tickets.html',
user=session['username'],
tickets=tickets,
status_filter=status_filter,
severity_filter=severity_filter,
config=config)
except Exception as e:
logger.error(f"Erreur page tickets: {e}")
return render_template('error.html', error=str(e)), 500
@app.route('/terminal')
@login_required
def terminal_page():
"""Page terminal web"""
return render_template('terminal.html',
user=session['username'],
config=config)
@app.route('/settings')
@login_required
def settings_page():
"""Page de configuration"""
return render_template('settings.html',
user=session['username'],
config=config)
# ========================
# API REST ENDPOINTS
# ========================
@app.route('/api/health')
def api_health():
"""API de vérification de santé"""
try:
# Vérifications de base
checks = {
'database': 'ok',
'config': 'ok',
'monitoring': 'ok'
}
# Test base de données
try:
conn = get_db()
cursor = conn.cursor()
cursor.execute("SELECT 1")
conn.close()
except:
checks['database'] = 'error'
# Test monitoring
try:
get_cached_metrics()
except:
checks['monitoring'] = 'error'
status = 'healthy' if all(check == 'ok' for check in checks.values()) else 'degraded'
return jsonify({
'status': status,
'timestamp': datetime.now().isoformat(),
'checks': checks,
'version': config.get('version', '2.0'),
'authenticated': 'user_id' in session
})
except Exception as e:
logger.error(f"Erreur health check: {e}")
return jsonify({
'status': 'unhealthy',
'timestamp': datetime.now().isoformat(),
'error': str(e)
}), 500
@app.route('/api/metrics/current')
@login_required
def api_current_metrics():
"""API pour obtenir les métriques actuelles"""
try:
metrics = get_cached_metrics()
return jsonify({
'success': True,
'data': metrics,
'timestamp': datetime.now().isoformat()
})
except Exception as e:
logger.error(f"Erreur API métriques: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/services')
@login_required
def api_services_list():
"""API pour lister les services"""
try:
metrics = get_cached_metrics()
services = metrics.get('services', [])
return jsonify({
'success': True,
'services': services
})
except Exception as e:
logger.error(f"Erreur API services: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/execute', methods=['POST'])
@login_required
def api_execute_command():
"""API pour exécuter des commandes"""
try:
data = request.get_json()
command = data.get('command', '').strip()
if not command:
return jsonify({'success': False, 'error': 'Commande vide'}), 400
result = execute_command(command, session['username'])
return jsonify({
'success': result['success'],
'command': command,
'output': result['output'],
'error': result['error'],
'return_code': result['return_code']
})
except Exception as e:
logger.error(f"Erreur exécution commande: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
# ========================
# ROUTES DE DEBUG
# ========================
@app.route('/debug')
def debug_info():
"""Page de debug pour diagnostiquer les problèmes"""
debug_data = {
'session': dict(session),
'config_loaded': bool(config),
'database_exists': os.path.exists(DB_PATH),
'templates_path': app.template_folder,
'static_path': app.static_folder,
'base_path': BASE_PATH,
'python_version': sys.version,
'flask_version': getattr(Flask, '__version__', 'Unknown'),
'current_time': datetime.now().isoformat(),
'working_directory': os.getcwd(),
'environment': dict(os.environ),
'request_info': {
'method': request.method,
'path': request.path,
'args': dict(request.args),
'headers': dict(request.headers)
}
}
# Test des métriques
try:
metrics = get_cached_metrics()
debug_data['metrics_test'] = 'OK'
debug_data['sample_metrics'] = {
'cpu': metrics.get('cpu', 'N/A'),
'ram': metrics.get('ram', 'N/A')
}
except Exception as e:
debug_data['metrics_test'] = f'ERROR: {str(e)}'
# Test de la base de données
try:
conn = get_db()
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM users")
user_count = cursor.fetchone()[0]
conn.close()
debug_data['database_test'] = f'OK - {user_count} utilisateurs'
except Exception as e:
debug_data['database_test'] = f'ERROR: {str(e)}'
return f"""
🐛 Debug NeuroPulse
🐛 Debug NeuroPulse Monitor Pro
📊 État de l'Application
Base de données: {'✅ Trouvée' if debug_data['database_exists'] else '❌ Manquante'}
Configuration: {'✅ Chargée' if debug_data['config_loaded'] else '❌ Erreur'}
Métriques: {debug_data.get('metrics_test', 'N/A')}
Base de données: {debug_data.get('database_test', 'N/A')}
👤 Session Utilisateur
Connecté: {'✅ Oui' if debug_data['session'].get('user_id') else '⚠️ Non'}
Utilisateur: {debug_data['session'].get('username', 'Non connecté')}
{json.dumps(debug_data['session'], indent=2)}
🔧 Informations Techniques
Dossier de travail: {debug_data['working_directory']}
Chemin base: {debug_data['base_path']}
Templates: {debug_data['templates_path']}
Python: {debug_data['python_version'].split()[0]}
📈 Métriques Sample
{json.dumps(debug_data.get('sample_metrics', {}), indent=2)}
🌐 Informations Requête
{json.dumps(debug_data['request_info'], indent=2)}
NeuroPulse Monitor Pro v2.0 - Debug généré à {debug_data['current_time']}
"""
def save_metrics_to_history():
"""Sauvegarder les métriques dans l'historique"""
try:
metrics = get_cached_metrics()
conn = get_db()
cursor = conn.cursor()
# Extraire les données importantes
cpu_percent = metrics.get('cpu', 0)
memory_percent = metrics.get('ram', 0)
disk_percent = metrics.get('disk', 0)
network = metrics.get('network', {})
cursor.execute('''
INSERT INTO metrics_history
(cpu_percent, memory_percent, disk_percent, network_in, network_out)
VALUES (?, ?, ?, ?, ?)
''', (
cpu_percent,
memory_percent,
disk_percent,
network.get('kb_in_per_sec', 0),
network.get('kb_out_per_sec', 0)
))
conn.commit()
conn.close()
except Exception as e:
logger.error(f"Erreur lors de la sauvegarde des métriques: {e}")
def background_tasks():
"""Tâches exécutées en arrière-plan"""
while True:
try:
# Sauvegarder les métriques toutes les minutes
save_metrics_to_history()
# Nettoyer l'historique des métriques (garder 30 jours)
if datetime.now().hour == 2 and datetime.now().minute < 1: # 2h du matin
conn = get_db()
cursor = conn.cursor()
cursor.execute(
"DELETE FROM metrics_history WHERE timestamp < datetime('now', '-30 days')"
)
conn.commit()
conn.close()
logger.info("Nettoyage historique métriques effectué")
except Exception as e:
logger.error(f"Erreur tâche arrière-plan: {e}")
time.sleep(60) # Attendre 1 minute
def initialize_app():
"""Initialiser l'application"""
logger.info("Initialisation de NeuroPulse Monitor Pro v2.0...")
# Charger la configuration
load_config()
# Configurer Flask
app.config['SECRET_KEY'] = config['application']['secret_key']
app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(hours=8)
app.config['SESSION_COOKIE_SECURE'] = False # True en production avec HTTPS
app.config['SESSION_COOKIE_HTTPONLY'] = True
app.config['SESSION_COOKIE_SAMESITE'] = 'Lax'
# Initialiser la base de données
init_database()
# Démarrer les tâches en arrière-plan
background_thread = threading.Thread(target=background_tasks, daemon=True)
background_thread.start()
logger.info("✅ NeuroPulse Monitor Pro v2.0 initialisé avec succès")
logger.info(f"📁 Base path: {BASE_PATH}")
logger.info(f"🗄️ Database: {DB_PATH}")
logger.info(f"⚙️ Config: {CONFIG_PATH}")
logger.info(f"📝 Logs: {LOG_PATH}")
if __name__ == '__main__':
try:
initialize_app()
host = config['application']['host']
port = config['application']['port']
debug = config['application']['debug']
logger.info(f"🚀 Démarrage de NeuroPulse Monitor Pro sur {host}:{port}")
logger.info(f"🌐 Accès: http://{host}:{port}/")
logger.info(f"🐛 Debug: http://{host}:{port}/debug")
logger.info(f"👤 Identifiants par défaut: admin / admin")
app.run(host=host, port=port, debug=debug, threaded=True)
except KeyboardInterrupt:
logger.info("Arrêt de NeuroPulse Monitor Pro")
except Exception as e:
logger.error(f"Erreur critique: {e}")
sys.exit(1)