""" Classe de base pour les plugins de clients torrent. Tous les plugins doivent hériter de cette classe. """ from abc import ABC, abstractmethod from typing import Optional, Dict, Any, List from dataclasses import dataclass @dataclass class TorrentClientConfig: """Configuration d'un client torrent""" host: str port: int = 0 # 0 = pas de port explicite (utiliser le port par défaut du protocole) username: str = "" password: str = "" use_ssl: bool = False path: str = "" # Chemin optionnel (ex: /qbittorrent) @property def base_url(self) -> str: protocol = "https" if self.use_ssl else "http" # Si le port est 0 ou vide, ne pas l'inclure dans l'URL if self.port and self.port > 0: url = f"{protocol}://{self.host}:{self.port}" else: url = f"{protocol}://{self.host}" # Ajouter le chemin s'il existe if self.path: # S'assurer que le chemin commence par / et ne finit pas par / path = self.path.strip('/') if path: url = f"{url}/{path}" return url @dataclass class TorrentInfo: """Informations sur un torrent""" hash: str name: str size: int progress: float # 0.0 à 1.0 status: str # downloading, seeding, paused, error, etc. download_speed: int # bytes/s upload_speed: int # bytes/s seeds: int peers: int save_path: str class TorrentClientPlugin(ABC): """ Classe abstraite pour les plugins de clients torrent. Pour créer un nouveau plugin : 1. Créer un fichier dans plugins/torrent_clients/ 2. Hériter de TorrentClientPlugin 3. Implémenter toutes les méthodes abstraites 4. Définir PLUGIN_NAME, PLUGIN_DESCRIPTION, PLUGIN_VERSION """ # Métadonnées du plugin (à surcharger) PLUGIN_NAME: str = "Base Plugin" PLUGIN_DESCRIPTION: str = "Plugin de base" PLUGIN_VERSION: str = "1.0.0" PLUGIN_AUTHOR: str = "Unknown" # Capacités du plugin SUPPORTS_TORRENT_FILES: bool = True # Supporte les URLs .torrent en plus des magnets def __init__(self, config: TorrentClientConfig): self.config = config self._connected = False @abstractmethod def connect(self) -> bool: """ Établit la connexion avec le client torrent. Retourne True si la connexion réussit. """ pass @abstractmethod def disconnect(self) -> None: """Ferme la connexion avec le client torrent.""" pass @abstractmethod def is_connected(self) -> bool: """Vérifie si la connexion est active.""" pass @abstractmethod def add_torrent_url(self, url: str, save_path: Optional[str] = None, category: Optional[str] = None, paused: bool = False) -> bool: """ Ajoute un torrent via son URL (magnet ou .torrent). Args: url: URL du torrent (magnet:// ou http://.torrent) save_path: Chemin de sauvegarde (optionnel) category: Catégorie/Label (optionnel) paused: Démarrer en pause (défaut: False) Returns: True si l'ajout réussit """ pass @abstractmethod def add_torrent_file(self, file_content: bytes, filename: str, save_path: Optional[str] = None, category: Optional[str] = None, paused: bool = False) -> bool: """ Ajoute un torrent via son contenu fichier. Args: file_content: Contenu binaire du fichier .torrent filename: Nom du fichier save_path: Chemin de sauvegarde (optionnel) category: Catégorie/Label (optionnel) paused: Démarrer en pause (défaut: False) Returns: True si l'ajout réussit """ pass @abstractmethod def get_torrents(self) -> List[TorrentInfo]: """ Récupère la liste de tous les torrents. Returns: Liste des torrents """ pass @abstractmethod def get_torrent(self, torrent_hash: str) -> Optional[TorrentInfo]: """ Récupère les informations d'un torrent spécifique. Args: torrent_hash: Hash du torrent Returns: Informations du torrent ou None si non trouvé """ pass @abstractmethod def pause_torrent(self, torrent_hash: str) -> bool: """Met un torrent en pause.""" pass @abstractmethod def resume_torrent(self, torrent_hash: str) -> bool: """Reprend un torrent en pause.""" pass @abstractmethod def delete_torrent(self, torrent_hash: str, delete_files: bool = False) -> bool: """ Supprime un torrent. Args: torrent_hash: Hash du torrent delete_files: Supprimer aussi les fichiers téléchargés Returns: True si la suppression réussit """ pass @abstractmethod def get_categories(self) -> List[str]: """Récupère la liste des catégories/labels disponibles.""" pass def get_info(self) -> Dict[str, Any]: """Retourne les informations du plugin.""" return { "name": self.PLUGIN_NAME, "description": self.PLUGIN_DESCRIPTION, "version": self.PLUGIN_VERSION, "author": self.PLUGIN_AUTHOR, "connected": self.is_connected() } def test_connection(self) -> Dict[str, Any]: """ Teste la connexion au client. Returns: {"success": bool, "message": str, "version": str (si connecté)} """ try: if self.connect(): return { "success": True, "message": "Connexion réussie", "client": self.PLUGIN_NAME } else: return { "success": False, "message": "Échec de la connexion" } except Exception as e: return { "success": False, "message": str(e) }