#!/usr/bin/env python3 """ Plugin loader for dynamic command discovery and loading Handles scanning, loading, and registering command plugins """ import os import sys import importlib import importlib.util from pathlib import Path from typing import Dict, List, Any, Optional, Type import logging from .commands.base_command import BaseCommand class PluginLoader: """Handles dynamic loading and discovery of command plugins""" def __init__(self, bot, commands_dir: str = None): self.bot = bot self.logger = bot.logger self.commands_dir = commands_dir or os.path.join(os.path.dirname(__file__), 'commands') self.loaded_plugins: Dict[str, BaseCommand] = {} self.plugin_metadata: Dict[str, Dict[str, Any]] = {} self.keyword_mappings: Dict[str, str] = {} # keyword -> plugin_name def discover_plugins(self) -> List[str]: """Discover all Python files in the commands directory that could be plugins""" plugin_files = [] commands_path = Path(self.commands_dir) if not commands_path.exists(): self.logger.error(f"Commands directory does not exist: {self.commands_dir}") return plugin_files # Scan for Python files (excluding __init__.py and base_command.py) for file_path in commands_path.glob("*.py"): if file_path.name not in ["__init__.py", "base_command.py", "plugin_loader.py"]: plugin_files.append(file_path.stem) self.logger.info(f"Discovered {len(plugin_files)} potential plugin files: {plugin_files}") return plugin_files def load_plugin(self, plugin_name: str) -> Optional[BaseCommand]: """Load a single plugin by name""" try: # Construct the full module path module_path = f"modules.commands.{plugin_name}" # Check if module is already loaded if module_path in sys.modules: module = sys.modules[module_path] else: # Import the module module = importlib.import_module(module_path) # Find the command class (should be the only class that inherits from BaseCommand) command_class = None for name, obj in inspect.getmembers(module, inspect.isclass): if (issubclass(obj, BaseCommand) and obj != BaseCommand and obj.__module__ == module_path): command_class = obj break if not command_class: self.logger.warning(f"No valid command class found in {plugin_name}") return None # Instantiate the command plugin_instance = command_class(self.bot) # Validate plugin metadata metadata = plugin_instance.get_metadata() if not metadata.get('name'): # Use the class name as the plugin name if not specified metadata['name'] = command_class.__name__.lower().replace('command', '') plugin_instance.name = metadata['name'] self.logger.info(f"Successfully loaded plugin: {metadata['name']} from {plugin_name}") return plugin_instance except Exception as e: self.logger.error(f"Failed to load plugin {plugin_name}: {e}") return None def load_all_plugins(self) -> Dict[str, BaseCommand]: """Load all discovered plugins""" plugin_files = self.discover_plugins() loaded_plugins = {} for plugin_file in plugin_files: plugin_instance = self.load_plugin(plugin_file) if plugin_instance: metadata = plugin_instance.get_metadata() plugin_name = metadata['name'] loaded_plugins[plugin_name] = plugin_instance self.plugin_metadata[plugin_name] = metadata # Build keyword mappings self._build_keyword_mappings(plugin_name, metadata) self.loaded_plugins = loaded_plugins self.logger.info(f"Loaded {len(loaded_plugins)} plugins: {list(loaded_plugins.keys())}") return loaded_plugins def _build_keyword_mappings(self, plugin_name: str, metadata: Dict[str, Any]): """Build keyword to plugin name mappings""" # Map keywords to plugin name for keyword in metadata.get('keywords', []): self.keyword_mappings[keyword.lower()] = plugin_name # Map aliases to plugin name for alias in metadata.get('aliases', []): self.keyword_mappings[alias.lower()] = plugin_name def get_plugin_by_keyword(self, keyword: str) -> Optional[BaseCommand]: """Get a plugin instance by keyword""" plugin_name = self.keyword_mappings.get(keyword.lower()) if plugin_name: return self.loaded_plugins.get(plugin_name) return None def get_plugin_by_name(self, name: str) -> Optional[BaseCommand]: """Get a plugin instance by name""" return self.loaded_plugins.get(name) def get_all_plugins(self) -> Dict[str, BaseCommand]: """Get all loaded plugins""" return self.loaded_plugins.copy() def get_plugin_metadata(self, plugin_name: str = None) -> Dict[str, Any]: """Get metadata for a specific plugin or all plugins""" if plugin_name: return self.plugin_metadata.get(plugin_name, {}) return self.plugin_metadata.copy() def get_plugins_by_category(self, category: str) -> Dict[str, BaseCommand]: """Get all plugins in a specific category""" return { name: plugin for name, plugin in self.loaded_plugins.items() if plugin.category == category } def reload_plugin(self, plugin_name: str) -> bool: """Reload a specific plugin""" try: # Remove from loaded plugins if plugin_name in self.loaded_plugins: del self.loaded_plugins[plugin_name] # Remove from metadata if plugin_name in self.plugin_metadata: del self.plugin_metadata[plugin_name] # Remove keyword mappings keywords_to_remove = [] for keyword, mapped_name in self.keyword_mappings.items(): if mapped_name == plugin_name: keywords_to_remove.append(keyword) for keyword in keywords_to_remove: del self.keyword_mappings[keyword] # Reload the plugin plugin_instance = self.load_plugin(plugin_name) if plugin_instance: metadata = plugin_instance.get_metadata() self.loaded_plugins[plugin_name] = plugin_instance self.plugin_metadata[plugin_name] = metadata self._build_keyword_mappings(plugin_name, metadata) self.logger.info(f"Successfully reloaded plugin: {plugin_name}") return True else: self.logger.error(f"Failed to reload plugin: {plugin_name}") return False except Exception as e: self.logger.error(f"Error reloading plugin {plugin_name}: {e}") return False def validate_plugin(self, plugin_instance: BaseCommand) -> List[str]: """Validate a plugin instance and return any issues""" issues = [] metadata = plugin_instance.get_metadata() # Check required metadata if not metadata.get('name'): issues.append("Plugin missing 'name' metadata") if not metadata.get('description'): issues.append("Plugin missing 'description' metadata") # Check if execute method is implemented if not hasattr(plugin_instance, 'execute'): issues.append("Plugin missing 'execute' method") # Check for keyword conflicts for keyword in metadata.get('keywords', []): if keyword.lower() in self.keyword_mappings: existing_plugin = self.keyword_mappings[keyword.lower()] if existing_plugin != metadata['name']: issues.append(f"Keyword '{keyword}' conflicts with plugin '{existing_plugin}'") return issues # Import inspect at module level for the load_plugin method import inspect