refactor(core): Mise en place de l'architecture multi-moteur (Strategy Pattern)

Author: Deniz09OK

__pycache__/moteur_base.cpython-311.pyc

@@ -1,12 +1,11 @@
 #!/usr/bin/env python3
-import subprocess
 import argparse
 import json
 import os
 import threading
 import logging
-from datetime import datetime
 from flask import Flask, render_template, jsonify
+from moteur_base import choisir_moteur
 
 dossier_script = os.path.dirname(os.path.realpath(__file__))
 chemin_changelog = os.path.join(dossier_script, "CHANGELOG.md")
@@ -28,64 +27,8 @@
 parser.add_argument("--serve", action="store_true", help="Lance directement le tableau de bord web sur le port 8080")
 args = parser.parse_args()
 
-
-def collecter_donnees():
-    with open("/etc/hostname") as f:
-        hostname = f.read().strip()
-
-    ram_info = ""
-    with open("/proc/meminfo") as f:
-        for ligne in f:
-            if "MemTotal" in ligne:
-                ram_info = ligne.strip()
-
-    cpu_info = "Inconnu"
-    with open("/proc/cpuinfo", "r") as f:
-        for ligne in f:
-            if "model name" in ligne:
-                cpu_info = ligne.split(":")[1].strip()
-                break
-
-    cmd_ps = subprocess.run(["ps", "aux"], capture_output=True, text=True)
-    processus_actifs = cmd_ps.stdout
-
-    cmd_tree = subprocess.run(["tree", "-L", "2", "/home/vagrant"], capture_output=True, text=True)
-    arborescence = cmd_tree.stdout
-
-    cmd_ports = subprocess.run(["ss", "-tuln"], capture_output=True, text=True)
-    ports_ouverts = cmd_ports.stdout
-
-    cmd_disk = subprocess.run(["df", "-h"], capture_output=True, text=True)
-    espace_disque = cmd_disk.stdout
-
-    cmd_uptime = subprocess.run(["uptime"], capture_output=True, text=True)
-    uptime_raw = cmd_uptime.stdout.strip()
-    parties = uptime_raw.split("load average:")
-    uptime_info = parties[0].strip().rstrip(",") if len(parties) > 0 else "Inconnu"
-    load_average = parties[1].strip() if len(parties) > 1 else "Inconnu"
-
-    utilisateurs_sudo = "Aucun"
-    with open("/etc/group", "r") as f:
-        for ligne in f:
-            if ligne.startswith("sudo:"):
-                utilisateurs_sudo = ligne.strip()
-                break
-
-    date_audit = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-
-    return {
-        "date": date_audit,
-        "machine": hostname,
-        "ram": ram_info,
-        "cpu": cpu_info,
-        "processus": processus_actifs,
-        "arborescence": arborescence,
-        "securite_ports": ports_ouverts,
-        "securite_sudoers": utilisateurs_sudo,
-        "stockage": espace_disque,
-        "uptime": uptime_info,
-        "load_average": load_average
-    }
+# Auto-détection de l'OS et initialisation du moteur approprié
+moteur_actif = choisir_moteur()
 
 
 def afficher_menu():
@@ -144,7 +87,7 @@ def mode_serve() -> None:
 
     @app.route('/')
     def dashboard():
-        donnees = collecter_donnees()
+        donnees = moteur_actif.collecter_donnees()
         return render_template('dashboard.html',
                                date_audit=donnees["date"],
                                hostname=donnees["machine"],
@@ -161,7 +104,7 @@ def dashboard():
 
     @app.route('/api/donnees')
     def api_donnees():
-        donnees = collecter_donnees()
+        donnees = moteur_actif.collecter_donnees()
         donnees["version_lsit"] = version_lsit
         return jsonify(donnees)
 
@@ -189,11 +132,11 @@ def main():
             break
         elif choix == "1":
             print("\nCollecte des données en cours...")
-            donnees = collecter_donnees()
+            donnees = moteur_actif.collecter_donnees()
             mode_txt(donnees)
         elif choix == "2":
             print("\nCollecte des données en cours...")
-            donnees = collecter_donnees()
+            donnees = moteur_actif.collecter_donnees()
             mode_json(donnees)
         elif choix == "3":
             print("\nDémarrage du serveur web...")
@@ -206,7 +149,7 @@ def main():
     if args.serve:
         mode_serve()
     else:
-        donnees = collecter_donnees()
+        donnees = moteur_actif.collecter_donnees()
         if args.format == "json":
             mode_json(donnees)
         else:

__pycache__/moteur_linux.cpython-311.pyc

@@ -0,0 +1,112 @@
+#!/usr/bin/env python3
+
+import platform
+from abc import ABC, abstractmethod
+from datetime import datetime
+from typing import Dict
+
+
+class MoteurBase(ABC):
+    """
+    Classe abstraite définissant le contrat que tout moteur OS doit respecter.
+    Impossible à instancier directement - les enfants (Linux, FreeBSD) 
+    DOIVENT implémenter toutes les méthodes @abstractmethod.
+    """
+
+    @abstractmethod
+    def obtenir_hostname(self) -> str:
+        """Retourne le nom de la machine."""
+        pass
+
+    @abstractmethod
+    def obtenir_ram(self) -> str:
+        """Retourne les informations sur la mémoire totale."""
+        pass
+
+    @abstractmethod
+    def obtenir_cpu(self) -> str:
+        """Retourne le modèle du processeur."""
+        pass
+
+    @abstractmethod
+    def obtenir_processus(self) -> str:
+        """Retourne la liste des processus actifs."""
+        pass
+
+    @abstractmethod
+    def obtenir_arborescence(self, chemin: str, profondeur: int = 2) -> str:
+        """Retourne l'arborescence d'un dossier."""
+        pass
+
+    @abstractmethod
+    def obtenir_ports_ouverts(self) -> str:
+        """Retourne la liste des ports ouverts."""
+        pass
+
+    @abstractmethod
+    def obtenir_sudoers(self) -> str:
+        """Retourne les utilisateurs ayant des droits sudo/wheel."""
+        pass
+
+    @abstractmethod
+    def obtenir_stockage(self) -> str:
+        """Retourne l'espace disque."""
+        pass
+
+    @abstractmethod
+    def obtenir_uptime(self) -> str:
+        """Retourne le temps de fonctionnement."""
+        pass
+
+    @abstractmethod
+    def obtenir_load_average(self) -> str:
+        """Retourne la charge système."""
+        pass
+
+    # Méthode NON abstraite (commune à tous les OS)
+    def obtenir_date_audit(self) -> str:
+        """Retourne la date actuelle - identique sur tous les OS."""
+        return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+
+    def collecter_donnees(self) -> Dict[str, str]:
+        """Collecte toutes les données et retourne un dictionnaire."""
+        return {
+            "date": self.obtenir_date_audit(),
+            "machine": self.obtenir_hostname(),
+            "ram": self.obtenir_ram(),
+            "cpu": self.obtenir_cpu(),
+            "processus": self.obtenir_processus(),
+            "arborescence": self.obtenir_arborescence("/home/vagrant"),
+            "securite_ports": self.obtenir_ports_ouverts(),
+            "securite_sudoers": self.obtenir_sudoers(),
+            "stockage": self.obtenir_stockage(),
+            "uptime": self.obtenir_uptime(),
+            "load_average": self.obtenir_load_average()
+        }
+
+
+def choisir_moteur() -> MoteurBase:
+    """
+    Routeur intelligent qui détecte l'OS et retourne le moteur approprié.
+    Utilise platform.system() qui renvoie "Linux", "FreeBSD", "Windows", etc.
+    """
+    systeme = platform.system()
+
+    if systeme == "Linux":
+        from moteur_linux import MoteurLinux
+        return MoteurLinux()
+
+    elif systeme == "FreeBSD":
+        # TODO: À implémenter dans l'étape suivante
+        # from moteur_freebsd import MoteurFreeBSD
+        # return MoteurFreeBSD()
+        raise NotImplementedError(
+            f"Le moteur FreeBSD n'est pas encore implémenté. "
+            f"OS détecté : {systeme}"
+        )
+
+    else:
+        raise NotImplementedError(
+            f"Système d'exploitation non supporté : {systeme}. "
+            f"Seuls Linux et FreeBSD sont pris en charge."
+        )

lsit.py

@@ -0,0 +1,78 @@
+#!/usr/bin/env python3
+
+import subprocess
+from moteur_base import MoteurBase
+
+
+class MoteurLinux(MoteurBase):
+    """
+    Moteur de collecte de données pour les systèmes Linux (Debian, Ubuntu, etc.).
+    Utilise /proc, /etc et les commandes GNU/Linux.
+    """
+
+    def obtenir_hostname(self) -> str:
+        """Lit le hostname depuis /etc/hostname (spécifique Linux)."""
+        with open("/etc/hostname") as f:
+            return f.read().strip()
+
+    def obtenir_ram(self) -> str:
+        """Lit la RAM depuis /proc/meminfo (spécifique Linux)."""
+        with open("/proc/meminfo") as f:
+            for ligne in f:
+                if "MemTotal" in ligne:
+                    return ligne.strip()
+        return "Inconnu"
+
+    def obtenir_cpu(self) -> str:
+        """Lit le modèle CPU depuis /proc/cpuinfo (spécifique Linux)."""
+        with open("/proc/cpuinfo", "r") as f:
+            for ligne in f:
+                if "model name" in ligne:
+                    return ligne.split(":")[1].strip()
+        return "Inconnu"
+
+    def obtenir_processus(self) -> str:
+        """Liste les processus avec ps aux (commande POSIX, fonctionne sur Linux)."""
+        cmd = subprocess.run(["ps", "aux"], capture_output=True, text=True)
+        return cmd.stdout
+
+    def obtenir_arborescence(self, chemin: str, profondeur: int = 2) -> str:
+        """Affiche l'arborescence avec tree (paquet à installer sur Linux)."""
+        cmd = subprocess.run(
+            ["tree", "-L", str(profondeur), chemin],
+            capture_output=True,
+            text=True
+        )
+        return cmd.stdout
+
+    def obtenir_ports_ouverts(self) -> str:
+        """Liste les ports ouverts avec ss (remplaçant de netstat sur Linux)."""
+        cmd = subprocess.run(["ss", "-tuln"], capture_output=True, text=True)
+        return cmd.stdout
+
+    def obtenir_sudoers(self) -> str:
+        """Lit le groupe sudo depuis /etc/group (spécifique Linux)."""
+        with open("/etc/group", "r") as f:
+            for ligne in f:
+                if ligne.startswith("sudo:"):
+                    return ligne.strip()
+        return "Aucun"
+
+    def obtenir_stockage(self) -> str:
+        """Affiche l'espace disque avec df -h (commande POSIX)."""
+        cmd = subprocess.run(["df", "-h"], capture_output=True, text=True)
+        return cmd.stdout
+
+    def obtenir_uptime(self) -> str:
+        """Récupère l'uptime via la commande uptime."""
+        cmd = subprocess.run(["uptime"], capture_output=True, text=True)
+        uptime_raw = cmd.stdout.strip()
+        parties = uptime_raw.split("load average:")
+        return parties[0].strip().rstrip(",") if len(parties) > 0 else "Inconnu"
+
+    def obtenir_load_average(self) -> str:
+        """Récupère le load average via la commande uptime."""
+        cmd = subprocess.run(["uptime"], capture_output=True, text=True)
+        uptime_raw = cmd.stdout.strip()
+        parties = uptime_raw.split("load average:")
+        return parties[1].strip() if len(parties) > 1 else "Inconnu"