# Onglet Utilisateurs Amadea — Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** Ajouter un onglet "Utilisateurs" affichant en temps réel les utilisateurs connectés à Amadea Web 8 x64 (statut, dernière action, graphique d'activité horaire), avec configuration du chemin logs et des seuils de statut. **Architecture:** Un module `user_monitor.py` tourne en background thread (miroir de `SystemMonitor`) — il parse les fichiers `awevents_YY-MM-DD_*.{txt,log}` et `isoft_YY-MM-DD_*.{txt,log}` du jour, calcule les statuts utilisateurs et expose un cache thread-safe. L'API `/api/users` retourne ce cache directement. Un second endpoint `/api/users/activity/weekly` lit les fichiers des 7 derniers jours à la demande. **Tech Stack:** Python 3.10+, Flask 3.1, Bootstrap 5.3, Bootstrap Icons, barres CSS pures (aucune librairie graphique), pytest --- ## Fichiers créés / modifiés | Fichier | Action | Rôle | |---|---|---| | `user_monitor.py` | Créer | Classe `UserMonitor` : thread, parsing, cache | | `tests/__init__.py` | Créer | Package tests | | `tests/test_user_monitor.py` | Créer | Tests unitaires du parsing et du calcul de statut | | `config_manager.py` | Modifier | Ajout des clés `amadea_log_path` + `user_status_thresholds` | | `app.py` | Modifier | Instanciation `UserMonitor`, 4 nouvelles routes | | `templates/base.html` | Modifier | Lien "Utilisateurs" dans la navbar | | `templates/settings.html` | Modifier | 2 nouveaux blocs de configuration | | `templates/users.html` | Créer | Tableau utilisateurs + graphique CSS + JS auto-refresh | | `requirements.txt` | Modifier | Ajout de `pytest` | --- ## Task 1 : Étendre ConfigManager avec les nouvelles clés **Files:** - Modify: `config_manager.py` - Modify: `requirements.txt` - [ ] **Étape 1 : Ajouter les nouvelles clés dans `get_default_config()`** Dans `config_manager.py`, dans la fonction `get_default_config()`, ajouter après la clé `"smtp"` : ```python "amadea_log_path": r"C:\ProgramData\ISoft\Amadea Web 8 x64\data\logs", "user_status_thresholds": { "active_minutes": 5, "inactive_minutes": 30, }, ``` La méthode `_load()` existante fusionne automatiquement les nouvelles clés avec une config déjà sauvegardée — aucune migration manuelle n'est nécessaire. - [ ] **Étape 2 : Ajouter pytest à requirements.txt** Ajouter à la fin de `requirements.txt` : ``` pytest==8.3.* ``` - [ ] **Étape 3 : Commit** ```bash git add config_manager.py requirements.txt git commit -m "feat: add amadea_log_path and user_status_thresholds config keys" ``` --- ## Task 2 : Créer `user_monitor.py` **Files:** - Create: `user_monitor.py` - Create: `tests/__init__.py` - Create: `tests/test_user_monitor.py` - [ ] **Étape 1 : Créer `user_monitor.py`** ```python """Suivi des utilisateurs connectes a Amadea via parsing des logs.""" import glob import os import re import threading import time from datetime import datetime, timedelta # Regex compilees au niveau module (performance) _AWEVENTS_RE = re.compile( r'^(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\.\d+;[^;]*;;;;"login=([^,]+),action=([^,]+),Label=(.+?)"?\s*$' ) _ISOFT_LOGIN_RE = re.compile( r'^(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}).*method=OpenUserSession.*login=([A-Za-z0-9_]+)' ) def _log_files_for_date(log_path, prefix, date_str): """Retourne les fichiers de logs pour un prefixe et une date donnes, tries par index.""" pattern = os.path.join(log_path, f"{prefix}_{date_str}_*") files = [f for f in glob.glob(pattern) if not f.endswith('.zip')] return sorted(files, key=lambda f: int(re.search(r'_(\d+)\.[^.]+$', f).group(1))) class UserMonitor: def __init__(self, config_manager): self.config = config_manager self._cache = {"users": {}, "hourly": [], "error": None, "no_files": False} self._lock = threading.Lock() self._running = False self._thread = None @property def data(self): with self._lock: return dict(self._cache) def start(self): if self._running: return self._running = True self._thread = threading.Thread(target=self._loop, daemon=True) self._thread.start() def stop(self): self._running = False def _loop(self): last_parse = 0 while self._running: interval = self.config.get("check_interval_minutes", 1) * 60 if time.time() - last_parse >= interval: try: self.parse_logs() except Exception as e: print(f"[UserMonitor] Erreur: {e}") last_parse = time.time() time.sleep(5) def parse_logs(self): log_path = self.config.get( "amadea_log_path", r"C:\ProgramData\ISoft\Amadea Web 8 x64\data\logs" ) thresholds = self.config.get( "user_status_thresholds", {"active_minutes": 5, "inactive_minutes": 30} ) if not os.path.isdir(log_path): with self._lock: self._cache = { "error": f"Dossier de logs introuvable : {log_path}", "users": {}, "hourly": [], "no_files": False, } return date_str = datetime.now().strftime("%y-%m-%d") awevents_files = _log_files_for_date(log_path, "awevents", date_str) if not awevents_files: with self._lock: self._cache = {"no_files": True, "error": None, "users": {}, "hourly": []} return now = datetime.now() cutoff_24h = now - timedelta(hours=24) users = {} hourly = {h: set() for h in range(24)} for filepath in awevents_files: try: with open(filepath, "r", encoding="utf-8", errors="ignore") as f: for line in f: self._parse_awevents_line(line, users, cutoff_24h, hourly) except (PermissionError, OSError): continue isoft_files = _log_files_for_date(log_path, "isoft", date_str) for filepath in isoft_files: try: with open(filepath, "r", encoding="utf-8", errors="ignore") as f: for line in f: self._parse_isoft_line(line, users) except (PermissionError, OSError): continue self._compute_statuses(users, thresholds, now) status_order = {"actif": 0, "inactif": 1, "deconnecte": 2} sorted_users = dict( sorted(users.items(), key=lambda x: status_order.get(x[1]["status"], 3)) ) hourly_data = [{"hour": h, "count": len(logins)} for h, logins in sorted(hourly.items())] with self._lock: self._cache = { "error": None, "no_files": False, "users": sorted_users, "hourly": hourly_data, } def _parse_awevents_line(self, line, users, cutoff_24h, hourly): m = _AWEVENTS_RE.match(line) if not m: return ts_str, login, action, label = m.group(1), m.group(2), m.group(3), m.group(4) try: ts = datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S") except ValueError: return login = login.strip() if not login: return is_logout = "se deconnecter" in label.lower() if login not in users: users[login] = { "login": login, "last_action_time": ts, "last_action_label": label[:60], "action_count_24h": 0, "status": "deconnecte", "explicit_logout": is_logout, "logout_time": ts if is_logout else None, "connected_since": ts, } else: user = users[login] if ts > user["last_action_time"]: user["last_action_time"] = ts user["last_action_label"] = label[:60] if is_logout: user["explicit_logout"] = True user["logout_time"] = ts elif user["explicit_logout"] and user.get("logout_time") and ts > user["logout_time"]: # Activite apres deconnexion explicite = reconnexion user["explicit_logout"] = False user["logout_time"] = None if ts >= cutoff_24h: users[login]["action_count_24h"] += 1 hourly[ts.hour].add(login) def _parse_isoft_line(self, line, users): m = _ISOFT_LOGIN_RE.match(line) if not m: return ts_str, login = m.group(1), m.group(2) try: ts = datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S") except ValueError: return if login in users and users[login]["connected_since"] is None: users[login]["connected_since"] = ts def _compute_statuses(self, users, thresholds, now): active_min = thresholds.get("active_minutes", 5) inactive_min = thresholds.get("inactive_minutes", 30) for user in users.values(): delta = (now - user["last_action_time"]).total_seconds() / 60 if user.get("explicit_logout"): user["status"] = "deconnecte" elif delta > inactive_min: user["status"] = "deconnecte" elif delta > active_min: user["status"] = "inactif" else: user["status"] = "actif" def get_weekly_activity(self): """Retourne le nombre max d'utilisateurs actifs simultanes par jour (7 derniers jours).""" log_path = self.config.get( "amadea_log_path", r"C:\ProgramData\ISoft\Amadea Web 8 x64\data\logs" ) if not os.path.isdir(log_path): return [] result = [] today = datetime.now().date() for delta in range(6, -1, -1): day = today - timedelta(days=delta) date_str = day.strftime("%y-%m-%d") files = _log_files_for_date(log_path, "awevents", date_str) if not files: result.append({"date": day.isoformat(), "count": None}) continue hourly = {h: set() for h in range(24)} for filepath in files: try: with open(filepath, "r", encoding="utf-8", errors="ignore") as f: for line in f: m = re.match( r'^(\d{4}-\d{2}-\d{2} (\d{2}):\d{2}:\d{2}).*login=([^,]+),', line ) if m: hour = int(m.group(2)) login = m.group(3).strip() if login: hourly[hour].add(login) except (PermissionError, OSError): continue max_concurrent = max((len(v) for v in hourly.values()), default=0) result.append({"date": day.isoformat(), "count": max_concurrent}) return result ``` - [ ] **Étape 2 : Créer `tests/__init__.py`** (fichier vide) ```bash mkdir -p tests && touch tests/__init__.py ``` - [ ] **Étape 3 : Créer `tests/test_user_monitor.py`** ```python """Tests unitaires pour user_monitor.py""" from datetime import datetime, timedelta import pytest from user_monitor import UserMonitor class FakeConfig: def get(self, key, default=None): return { "amadea_log_path": "/nonexistent", "user_status_thresholds": {"active_minutes": 5, "inactive_minutes": 30}, "check_interval_minutes": 1, }.get(key, default) def make_monitor(): return UserMonitor(FakeConfig()) # --- Parsing awevents --- def test_parse_awevents_line_basic(): monitor = make_monitor() users, hourly = {}, {h: set() for h in range(24)} cutoff = datetime(2026, 3, 30, 0, 0, 0) line = '2026-03-30 10:34:24.034;INFO ;;;;"login=JENKINS,action=SelectionChange,Label=BAO_Main/MenuPrincipal"\n' monitor._parse_awevents_line(line, users, cutoff, hourly) assert "JENKINS" in users assert users["JENKINS"]["last_action_time"] == datetime(2026, 3, 30, 10, 34, 24) assert users["JENKINS"]["action_count_24h"] == 1 assert users["JENKINS"]["explicit_logout"] is False assert hourly[10] == {"JENKINS"} def test_parse_awevents_line_explicit_logout(): monitor = make_monitor() users, hourly = {}, {h: set() for h in range(24)} cutoff = datetime(2026, 3, 30, 0, 0, 0) line = '2026-03-30 11:34:00.500;INFO ;;;;"login=MB,action=Action,Label=Main/se deconnecter"\n' monitor._parse_awevents_line(line, users, cutoff, hourly) assert users["MB"]["explicit_logout"] is True assert users["MB"]["logout_time"] == datetime(2026, 3, 30, 11, 34, 0) def test_parse_awevents_line_reconnect_after_logout(): monitor = make_monitor() users, hourly = {}, {h: set() for h in range(24)} cutoff = datetime(2026, 3, 30, 0, 0, 0) logout_line = '2026-03-30 11:34:00.500;INFO ;;;;"login=MB,action=Action,Label=Main/se deconnecter"\n' reconnect_line = '2026-03-30 11:34:19.594;INFO ;;;;"login=MB,action=Action,Label=Main/OuvrirCTRL 1/Table"\n' monitor._parse_awevents_line(logout_line, users, cutoff, hourly) assert users["MB"]["explicit_logout"] is True monitor._parse_awevents_line(reconnect_line, users, cutoff, hourly) assert users["MB"]["explicit_logout"] is False def test_parse_awevents_line_invalid_ignored(): monitor = make_monitor() users, hourly = {}, {h: set() for h in range(24)} cutoff = datetime(2026, 3, 30, 0, 0, 0) monitor._parse_awevents_line("ligne invalide sans format attendu\n", users, cutoff, hourly) assert users == {} def test_parse_awevents_action_count_outside_24h(): monitor = make_monitor() users, hourly = {}, {h: set() for h in range(24)} cutoff = datetime(2026, 3, 30, 12, 0, 0) old_line = '2026-03-30 08:00:00.000;INFO ;;;;"login=JENKINS,action=Click,Label=Main/Page"\n' monitor._parse_awevents_line(old_line, users, cutoff, hourly) assert users["JENKINS"]["action_count_24h"] == 0 # --- Calcul de statut --- def test_compute_statuses_actif(): monitor = make_monitor() now = datetime(2026, 3, 30, 12, 0, 0) thresholds = {"active_minutes": 5, "inactive_minutes": 30} users = {"JENKINS": {"last_action_time": now - timedelta(minutes=2), "explicit_logout": False}} monitor._compute_statuses(users, thresholds, now) assert users["JENKINS"]["status"] == "actif" def test_compute_statuses_inactif(): monitor = make_monitor() now = datetime(2026, 3, 30, 12, 0, 0) thresholds = {"active_minutes": 5, "inactive_minutes": 30} users = {"MB": {"last_action_time": now - timedelta(minutes=15), "explicit_logout": False}} monitor._compute_statuses(users, thresholds, now) assert users["MB"]["status"] == "inactif" def test_compute_statuses_deconnecte_timeout(): monitor = make_monitor() now = datetime(2026, 3, 30, 12, 0, 0) thresholds = {"active_minutes": 5, "inactive_minutes": 30} users = {"KO": {"last_action_time": now - timedelta(minutes=45), "explicit_logout": False}} monitor._compute_statuses(users, thresholds, now) assert users["KO"]["status"] == "deconnecte" def test_compute_statuses_deconnecte_explicit(): monitor = make_monitor() now = datetime(2026, 3, 30, 12, 0, 0) thresholds = {"active_minutes": 5, "inactive_minutes": 30} users = {"MB": {"last_action_time": now - timedelta(minutes=2), "explicit_logout": True}} monitor._compute_statuses(users, thresholds, now) assert users["MB"]["status"] == "deconnecte" ``` - [ ] **Étape 4 : Lancer les tests** ```bash .venv/bin/pip install pytest -q .venv/bin/pytest tests/test_user_monitor.py -v ``` Résultat attendu : 9 tests PASSED. - [ ] **Étape 5 : Commit** ```bash git add user_monitor.py tests/__init__.py tests/test_user_monitor.py requirements.txt git commit -m "feat: add UserMonitor with Amadea log parsing" ``` --- ## Task 3 : Étendre `app.py` — routes et instanciation **Files:** - Modify: `app.py` - [ ] **Étape 1 : Importer et instancier UserMonitor** Dans `app.py`, après `from alerter import EmailAlerter`, ajouter : ```python from user_monitor import UserMonitor ``` Après `monitor = SystemMonitor(config, alerter)`, ajouter : ```python user_monitor = UserMonitor(config) ``` - [ ] **Étape 2 : Démarrer UserMonitor dans `main()`** Dans `main()`, après `monitor.start()`, ajouter : ```python user_monitor.start() user_monitor.parse_logs() ``` - [ ] **Étape 3 : Ajouter les 4 nouvelles routes** Ajouter avant la fonction `check_port_available` : ```python @app.route("/users") @login_required def users(): return render_template("users.html") @app.route("/api/users") @login_required def api_users(): cache = user_monitor.data if cache.get("error"): return jsonify({"error": cache["error"]}) if cache.get("no_files"): return jsonify({"no_files": True}) users_list = [ { "login": u["login"], "status": u["status"], "last_action_time": u["last_action_time"].strftime("%H:%M:%S") if u.get("last_action_time") else None, "last_action_label": u.get("last_action_label", ""), "action_count_24h": u.get("action_count_24h", 0), "connected_since": u["connected_since"].strftime("%H:%M") if u.get("connected_since") else None, "explicit_logout": u.get("explicit_logout", False), } for u in cache.get("users", {}).values() ] return jsonify({"users": users_list, "hourly": cache.get("hourly", [])}) @app.route("/api/users/activity/weekly") @login_required def api_users_weekly(): return jsonify({"weekly": user_monitor.get_weekly_activity()}) @app.route("/settings/amadea-log-path", methods=["POST"]) @login_required def update_amadea_log_path(): path = request.form.get("amadea_log_path", "").strip() if not path: flash("Le chemin ne peut pas etre vide.", "danger") return redirect(url_for("settings")) config.set("amadea_log_path", path) flash("Chemin des logs Amadea mis a jour.", "success") return redirect(url_for("settings")) @app.route("/settings/user-thresholds", methods=["POST"]) @login_required def update_user_thresholds(): try: active = int(request.form["active_minutes"]) inactive = int(request.form["inactive_minutes"]) if active < 1 or inactive < 1: flash("Les seuils doivent etre d'au moins 1 minute.", "danger") return redirect(url_for("settings")) if active >= inactive: flash("Le seuil 'actif' doit etre inferieur au seuil 'inactif'.", "danger") return redirect(url_for("settings")) config.set("user_status_thresholds", {"active_minutes": active, "inactive_minutes": inactive}) flash("Seuils utilisateurs mis a jour.", "success") except (ValueError, KeyError) as e: flash(f"Erreur: {e}", "danger") return redirect(url_for("settings")) ``` - [ ] **Étape 4 : Vérifier le démarrage** ```bash .venv/bin/python app.py ``` Résultat attendu : `[Supervision] Demarrage sur le port 5000` sans traceback. Ctrl+C pour arrêter. - [ ] **Étape 5 : Commit** ```bash git add app.py git commit -m "feat: wire UserMonitor into app, add /users and /api/users routes" ``` --- ## Task 4 : Mettre à jour la navigation dans `base.html` **Files:** - Modify: `templates/base.html` - [ ] **Étape 1 : Ajouter le lien "Utilisateurs"** Dans `templates/base.html`, après le `
  • ` pour "Alertes" (après la balise `
    • ` qui ferme le lien Alertes), ajouter : ```html
  • Utilisateurs
    • ``` - [ ] **Étape 2 : Vérifier visuellement** Ouvrir http://localhost:5000 — la navbar doit afficher : Tableau de bord | Configuration | Alertes | Utilisateurs. - [ ] **Étape 3 : Commit** ```bash git add templates/base.html git commit -m "feat: add Utilisateurs link to navbar" ``` --- ## Task 5 : Mettre à jour `settings.html` — 2 nouveaux blocs **Files:** - Modify: `templates/settings.html` - [ ] **Étape 1 : Ajouter les 2 blocs avant `{% endblock %}`** Dans `templates/settings.html`, juste avant la balise `{% endblock %}` finale, ajouter : ```html
    Chemin des logs Amadea
    Dossier contenant les fichiers isoft_*.txt et awevents_*.txt.
    Seuils statut utilisateurs
    Au-dela du seuil inactif, le statut passe a Deconnecte.
    ``` - [ ] **Étape 2 : Vérifier visuellement** Ouvrir http://localhost:5000/settings. Les deux nouveaux blocs doivent apparaître en bas avec les valeurs par défaut pré-remplies. Tester la sauvegarde de chaque formulaire. - [ ] **Étape 3 : Commit** ```bash git add templates/settings.html git commit -m "feat: add Amadea log path and user status thresholds to settings" ``` --- ## Task 6 : Créer `templates/users.html` **Files:** - Create: `templates/users.html` - [ ] **Étape 1 : Créer le template** Le JavaScript utilise exclusivement `textContent` et `createElement` pour insérer des données dans le DOM — aucune interpolation de données dans des chaînes HTML. ```html {% extends "base.html" %} {% block title %}Supervision - Utilisateurs{% endblock %} {% block content %}

    Utilisateurs Amadea

    Utilisateurs actifs par heure
    Les donnees historiques des jours precedents ne sont pas disponibles (fichiers archives).
    Utilisateurs connectes aujourd'hui
    Utilisateur Statut Derniere action Actions (24h) Depuis
    {% endblock %} {% block scripts %} {% endblock %} ``` - [ ] **Étape 2 : Vérifier visuellement** Lancer l'app et ouvrir http://localhost:5000/users. - Sur macOS (dossier logs Windows inexistant) : alerte orange "Dossier de logs introuvable" - Sur Windows avec logs présents : tableau + graphique barres bleues - Tester le sélecteur "7 derniers jours" - Vérifier auto-refresh via la console réseau (requête toutes les 30s) - [ ] **Étape 3 : Commit final** ```bash git add templates/users.html git commit -m "feat: add users.html template with table and activity chart" ``` --- ## Vérification finale - [ ] `pytest tests/test_user_monitor.py -v` → tous verts - [ ] Routes fonctionnelles : `/users`, `/api/users`, `/api/users/activity/weekly` - [ ] Configuration → sauvegarder un chemin de logs → vérifier `data/config.json` - [ ] Configuration → seuils invalides (actif >= inactif) → message d'erreur affiché - [ ] Onglets existants (Tableau de bord, Configuration, Alertes) → comportement inchangé ```bash git log --oneline -6 ``` Résultat attendu : ``` feat: add users.html template with table and activity chart feat: add Amadea log path and user status thresholds to settings feat: add Utilisateurs link to navbar feat: wire UserMonitor into app, add /users and /api/users routes feat: add UserMonitor with Amadea log parsing feat: add amadea_log_path and user_status_thresholds config keys ```