141 lines
4.1 KiB
Python
141 lines
4.1 KiB
Python
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 []
|