from abc import ABC, abstractmethod class AbstractFS(ABC): """ Interface abstraite pour tous les plugins de système de fichiers. Pour créer un nouveau plugin : 1. Créer plugins/monplugin.py 2. Hériter de AbstractFS 3. Implémenter toutes les méthodes abstraites 4. Définir PLUGIN_NAME et PLUGIN_LABEL Le plugin sera automatiquement découvert au démarrage. """ PLUGIN_NAME = None # identifiant interne ex: "sftp" PLUGIN_LABEL = None # label affiché ex: "SFTP" # ─── Cycle de vie ──────────────────────────────────────────── @abstractmethod def connect(self, config: dict): """Établir la connexion avec la config fournie.""" pass @abstractmethod def disconnect(self): """Fermer proprement la connexion.""" pass @abstractmethod def is_connected(self) -> bool: """Retourner True si la connexion est active.""" pass # ─── Navigation ────────────────────────────────────────────── @abstractmethod def list(self, path: str) -> list: """ Lister le contenu d'un dossier. Retourne une liste de dicts : { name, path, is_dir, size, mtime } Triés : dossiers d'abord, puis fichiers, alphabétique. """ pass @abstractmethod def isdir(self, path: str) -> bool: pass @abstractmethod def exists(self, path: str) -> bool: pass @abstractmethod def getsize(self, path: str) -> int: pass @abstractmethod def join(self, *parts) -> str: """Équivalent os.path.join pour ce FS.""" pass @abstractmethod def basename(self, path: str) -> str: pass @abstractmethod def dirname(self, path: str) -> str: pass @abstractmethod def relpath(self, path: str, base: str) -> str: pass # ─── Opérations ────────────────────────────────────────────── @abstractmethod def mkdir(self, path: str): """Créer un dossier (et les parents si nécessaire).""" pass @abstractmethod def rename(self, old_path: str, new_path: str): pass @abstractmethod def remove(self, path: str): pass @abstractmethod def walk(self, path: str): """ Générateur identique à os.walk : yield (root, dirs, files) """ pass # ─── Transfert ─────────────────────────────────────────────── @abstractmethod def read_chunks(self, path: str, chunk_size: int = 4 * 1024 * 1024): """ Générateur qui yield des bytes chunk par chunk. Utilisé par le moteur de copie. """ pass @abstractmethod def write_chunks(self, path: str, chunks): """ Écrire un fichier à partir d'un générateur de chunks bytes. Utilisé par le moteur de copie. """ pass def get_total_size(self, path: str) -> int: """Taille totale d'un fichier ou dossier récursif.""" if not self.isdir(path): return self.getsize(path) total = 0 for root, dirs, files in self.walk(path): for f in files: try: total += self.getsize(self.join(root, f)) except Exception: pass return total # ─── Métadonnées du plugin ──────────────────────────────────── @classmethod def get_config_fields(cls) -> list: """ Retourne la liste des champs de config nécessaires pour ce plugin. Chaque champ : { name, label, type, required, default } type : "text" | "password" | "number" | "file" Surcharger dans chaque plugin. """ return []