""" Registry Actions VWB - Gestionnaire d'Actions VisionOnly Auteur : Dom, Alice, Kiro - 09 janvier 2026 Ce module implémente le registry des actions VisionOnly pour le Visual Workflow Builder. Il permet l'enregistrement, la recherche et la gestion des actions disponibles. Fonctionnalités : - Enregistrement automatique des actions - Recherche par catégorie et type - Thread-safety pour accès concurrent - Chargement dynamique des actions """ import threading from typing import Dict, List, Optional, Type, Any, Set from datetime import datetime from pathlib import Path import importlib import inspect from .base_action import BaseVWBAction class VWBActionRegistry: """ Registry thread-safe pour les actions VWB. Ce registry maintient un catalogue des actions disponibles et permet leur recherche et instanciation dynamique. """ def __init__(self): """Initialise le registry.""" self._actions: Dict[str, Type[BaseVWBAction]] = {} self._categories: Dict[str, Set[str]] = {} self._metadata: Dict[str, Dict[str, Any]] = {} self._lock = threading.RLock() self._initialized = False print("📋 Registry Actions VWB initialisé") def register_action(self, action_class: Type[BaseVWBAction], action_id: Optional[str] = None, category: str = "default", metadata: Optional[Dict[str, Any]] = None) -> bool: """ Enregistre une action dans le registry. Args: action_class: Classe de l'action à enregistrer action_id: Identifiant unique (auto-généré si None) category: Catégorie de l'action metadata: Métadonnées additionnelles Returns: True si l'enregistrement a réussi """ with self._lock: try: # Générer l'ID si non fourni if action_id is None: action_id = self._generate_action_id(action_class) # Vérifier que l'action hérite de BaseVWBAction if not issubclass(action_class, BaseVWBAction): print(f"⚠️ {action_class.__name__} n'hérite pas de BaseVWBAction") return False # Vérifier l'unicité de l'ID if action_id in self._actions: print(f"⚠️ Action '{action_id}' déjà enregistrée") return False # Enregistrer l'action self._actions[action_id] = action_class # Gérer les catégories if category not in self._categories: self._categories[category] = set() self._categories[category].add(action_id) # Stocker les métadonnées self._metadata[action_id] = { 'class_name': action_class.__name__, 'module': action_class.__module__, 'category': category, 'registered_at': datetime.now().isoformat(), 'metadata': metadata or {} } print(f"✅ Action '{action_id}' enregistrée (catégorie: {category})") return True except Exception as e: print(f"❌ Erreur enregistrement action '{action_id}': {e}") return False def get_action_class(self, action_id: str) -> Optional[Type[BaseVWBAction]]: """ Récupère la classe d'une action par son ID. Args: action_id: Identifiant de l'action Returns: Classe de l'action ou None si non trouvée """ with self._lock: return self._actions.get(action_id) def create_action(self, action_id: str, parameters: Optional[Dict[str, Any]] = None, **kwargs) -> Optional[BaseVWBAction]: """ Crée une instance d'action. Args: action_id: Identifiant de l'action parameters: Paramètres de l'action **kwargs: Arguments additionnels pour le constructeur Returns: Instance de l'action ou None si erreur """ with self._lock: action_class = self._actions.get(action_id) if action_class is None: print(f"⚠️ Action '{action_id}' non trouvée dans le registry") return None try: # Préparer les arguments du constructeur constructor_args = { 'action_id': f"{action_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}", 'parameters': parameters or {} } constructor_args.update(kwargs) # Créer l'instance instance = action_class(**constructor_args) print(f"✅ Instance d'action '{action_id}' créée") return instance except Exception as e: print(f"❌ Erreur création instance '{action_id}': {e}") return None def list_actions(self, category: Optional[str] = None) -> List[str]: """ Liste les actions disponibles. Args: category: Filtrer par catégorie (optionnel) Returns: Liste des IDs d'actions """ with self._lock: if category is None: return list(self._actions.keys()) else: return list(self._categories.get(category, set())) def list_categories(self) -> List[str]: """ Liste les catégories disponibles. Returns: Liste des catégories """ with self._lock: return list(self._categories.keys()) def get_action_metadata(self, action_id: str) -> Optional[Dict[str, Any]]: """ Récupère les métadonnées d'une action. Args: action_id: Identifiant de l'action Returns: Métadonnées ou None si non trouvée """ with self._lock: return self._metadata.get(action_id) def search_actions(self, query: str, category: Optional[str] = None) -> List[str]: """ Recherche des actions par nom ou description. Args: query: Terme de recherche category: Filtrer par catégorie (optionnel) Returns: Liste des IDs d'actions correspondantes """ with self._lock: query_lower = query.lower() results = [] actions_to_search = self.list_actions(category) for action_id in actions_to_search: metadata = self._metadata.get(action_id, {}) class_name = metadata.get('class_name', '').lower() # Rechercher dans l'ID et le nom de classe if (query_lower in action_id.lower() or query_lower in class_name): results.append(action_id) return results def get_registry_stats(self) -> Dict[str, Any]: """ Obtient les statistiques du registry. Returns: Dictionnaire avec les statistiques """ with self._lock: return { 'total_actions': len(self._actions), 'categories': { cat: len(actions) for cat, actions in self._categories.items() }, 'initialized': self._initialized, 'last_update': datetime.now().isoformat() } def auto_discover_actions(self, base_path: Optional[Path] = None) -> int: """ Découvre et enregistre automatiquement les actions. Args: base_path: Chemin de base pour la découverte (optionnel) Returns: Nombre d'actions découvertes """ with self._lock: if base_path is None: base_path = Path(__file__).parent discovered_count = 0 try: # Découvrir les actions dans vision_ui vision_ui_path = base_path / "vision_ui" if vision_ui_path.exists(): discovered_count += self._discover_in_directory( vision_ui_path, "vision_ui" ) # Découvrir les actions dans d'autres catégories for category_dir in base_path.iterdir(): if (category_dir.is_dir() and category_dir.name not in ["__pycache__", "vision_ui"] and not category_dir.name.startswith(".")): discovered_count += self._discover_in_directory( category_dir, category_dir.name ) self._initialized = True print(f"🔍 Découverte automatique terminée : {discovered_count} actions trouvées") return discovered_count except Exception as e: print(f"❌ Erreur découverte automatique : {e}") return discovered_count def _discover_in_directory(self, directory: Path, category: str) -> int: """ Découvre les actions dans un répertoire. Args: directory: Répertoire à explorer category: Catégorie des actions Returns: Nombre d'actions découvertes """ discovered_count = 0 for py_file in directory.glob("*.py"): if py_file.name.startswith("__"): continue try: # Construire le nom du module module_name = f"visual_workflow_builder.backend.actions.{category}.{py_file.stem}" # Importer le module module = importlib.import_module(module_name) # Chercher les classes d'actions for name, obj in inspect.getmembers(module, inspect.isclass): if (obj != BaseVWBAction and issubclass(obj, BaseVWBAction) and obj.__module__ == module.__name__): # Générer l'ID de l'action action_id = self._generate_action_id(obj) # Enregistrer l'action if self.register_action(obj, action_id, category): discovered_count += 1 except Exception as e: print(f"⚠️ Erreur import {py_file}: {e}") return discovered_count def _generate_action_id(self, action_class: Type[BaseVWBAction]) -> str: """ Génère un ID d'action à partir de la classe. Args: action_class: Classe de l'action Returns: ID généré """ class_name = action_class.__name__ # Convertir VWBClickAnchorAction -> click_anchor if class_name.startswith("VWB") and class_name.endswith("Action"): # Enlever VWB et Action core_name = class_name[3:-6] # Convertir CamelCase en snake_case import re snake_case = re.sub('([A-Z]+)', r'_\1', core_name).lower() return snake_case.lstrip('_') # Fallback : utiliser le nom de classe en minuscules return class_name.lower() def clear(self): """Vide le registry.""" with self._lock: self._actions.clear() self._categories.clear() self._metadata.clear() self._initialized = False print("🗑️ Registry vidé") # Instance globale du registry _global_registry: Optional[VWBActionRegistry] = None _registry_lock = threading.Lock() def get_global_registry() -> VWBActionRegistry: """ Obtient l'instance globale du registry (singleton thread-safe). Returns: Instance du registry """ global _global_registry if _global_registry is None: with _registry_lock: if _global_registry is None: _global_registry = VWBActionRegistry() # Auto-découverte des actions au premier accès try: _global_registry.auto_discover_actions() except Exception as e: print(f"⚠️ Erreur auto-découverte : {e}") return _global_registry def register_action(action_class: Type[BaseVWBAction], action_id: Optional[str] = None, category: str = "default", metadata: Optional[Dict[str, Any]] = None) -> bool: """ Enregistre une action dans le registry global. Args: action_class: Classe de l'action action_id: Identifiant unique (optionnel) category: Catégorie de l'action metadata: Métadonnées additionnelles Returns: True si l'enregistrement a réussi """ return get_global_registry().register_action( action_class, action_id, category, metadata ) def get_action_class(action_id: str) -> Optional[Type[BaseVWBAction]]: """ Récupère une classe d'action du registry global. Args: action_id: Identifiant de l'action Returns: Classe de l'action ou None """ return get_global_registry().get_action_class(action_id) def create_action(action_id: str, parameters: Optional[Dict[str, Any]] = None, **kwargs) -> Optional[BaseVWBAction]: """ Crée une instance d'action depuis le registry global. Args: action_id: Identifiant de l'action parameters: Paramètres de l'action **kwargs: Arguments additionnels Returns: Instance de l'action ou None """ return get_global_registry().create_action(action_id, parameters, **kwargs) def list_available_actions(category: Optional[str] = None) -> List[str]: """ Liste les actions disponibles dans le registry global. Args: category: Filtrer par catégorie (optionnel) Returns: Liste des IDs d'actions """ return get_global_registry().list_actions(category) def get_registry_info() -> Dict[str, Any]: """ Obtient les informations du registry global. Returns: Informations du registry """ registry = get_global_registry() stats = registry.get_registry_stats() return { 'stats': stats, 'actions': { action_id: registry.get_action_metadata(action_id) for action_id in registry.list_actions() }, 'categories': registry.list_categories() } # Décorateur pour l'enregistrement automatique def vwb_action(action_id: Optional[str] = None, category: str = "default", metadata: Optional[Dict[str, Any]] = None): """ Décorateur pour l'enregistrement automatique d'actions VWB. Args: action_id: Identifiant unique (optionnel) category: Catégorie de l'action metadata: Métadonnées additionnelles Returns: Décorateur """ def decorator(action_class: Type[BaseVWBAction]): register_action(action_class, action_id, category, metadata) return action_class return decorator if __name__ == "__main__": # Test du registry print("🧪 Test du Registry Actions VWB") registry = VWBActionRegistry() # Test de découverte automatique discovered = registry.auto_discover_actions() print(f"Actions découvertes : {discovered}") # Afficher les statistiques stats = registry.get_registry_stats() print(f"Statistiques : {stats}") # Lister les actions actions = registry.list_actions() print(f"Actions disponibles : {actions}") # Test de création d'action if actions: test_action_id = actions[0] instance = registry.create_action(test_action_id) if instance: print(f"✅ Instance créée pour '{test_action_id}': {type(instance).__name__}") else: print(f"❌ Échec création instance pour '{test_action_id}'")