import importlib.util import os import sys # Added this line from typing import Dict, List, Optional from .plugin_base import PluginBase, PluginConfig from .logger import belief_scope # [DEF:PluginLoader:Class] # @TIER: STANDARD # @SEMANTICS: plugin, loader, dynamic, import # @PURPOSE: Scans a specified directory for Python modules, dynamically loads them, and registers any classes that are valid implementations of the PluginBase interface. # @LAYER: Core # @RELATION: Depends on PluginBase. It is used by the main application to discover and manage available plugins. class PluginLoader: """ Scans a directory for Python modules, loads them, and identifies classes that inherit from PluginBase. """ # [DEF:__init__:Function] # @PURPOSE: Initializes the PluginLoader with a directory to scan. # @PRE: plugin_dir is a valid directory path. # @POST: Plugins are loaded and registered. # @PARAM: plugin_dir (str) - The directory containing plugin modules. def __init__(self, plugin_dir: str): with belief_scope("__init__"): self.plugin_dir = plugin_dir self._plugins: Dict[str, PluginBase] = {} self._plugin_configs: Dict[str, PluginConfig] = {} self._load_plugins() # [/DEF:__init__:Function] # [DEF:_load_plugins:Function] # @PURPOSE: Scans the plugin directory and loads all valid plugins. # @PRE: plugin_dir exists or can be created. # @POST: _load_module is called for each .py file. def _load_plugins(self): with belief_scope("_load_plugins"): """ Scans the plugin directory, imports modules, and registers valid plugins. """ if not os.path.exists(self.plugin_dir): os.makedirs(self.plugin_dir) # Add the plugin directory's parent to sys.path to enable relative imports within plugins # This assumes plugin_dir is something like 'backend/src/plugins' # and we want 'backend/src' to be on the path for 'from ..core...' imports plugin_parent_dir = os.path.abspath(os.path.join(self.plugin_dir, os.pardir)) if plugin_parent_dir not in sys.path: sys.path.insert(0, plugin_parent_dir) for filename in os.listdir(self.plugin_dir): file_path = os.path.join(self.plugin_dir, filename) # Handle directory-based plugins (packages) if os.path.isdir(file_path): init_file = os.path.join(file_path, "__init__.py") if os.path.exists(init_file): self._load_module(filename, init_file) continue # Handle single-file plugins if filename.endswith(".py") and filename != "__init__.py": module_name = filename[:-3] self._load_module(module_name, file_path) # [/DEF:_load_plugins:Function] # [DEF:_load_module:Function] # @PURPOSE: Loads a single Python module and discovers PluginBase implementations. # @PRE: module_name and file_path are valid. # @POST: Plugin classes are instantiated and registered. # @PARAM: module_name (str) - The name of the module. # @PARAM: file_path (str) - The path to the module file. def _load_module(self, module_name: str, file_path: str): with belief_scope("_load_module"): """ Loads a single Python module and extracts PluginBase subclasses. """ # Try to determine the correct package prefix based on how the app is running # For standalone execution, we need to handle the import differently if __name__ == "__main__" or "test" in __name__: # When running as standalone or in tests, use relative import package_name = f"plugins.{module_name}" elif "backend.src" in __name__: package_prefix = "backend.src.plugins" package_name = f"{package_prefix}.{module_name}" else: package_prefix = "src.plugins" package_name = f"{package_prefix}.{module_name}" # print(f"DEBUG: Loading plugin {module_name} as {package_name}") spec = importlib.util.spec_from_file_location(package_name, file_path) if spec is None or spec.loader is None: print(f"Could not load module spec for {package_name}") # Replace with proper logging return module = importlib.util.module_from_spec(spec) try: spec.loader.exec_module(module) except Exception as e: print(f"Error loading plugin module {module_name}: {e}") # Replace with proper logging return for attribute_name in dir(module): attribute = getattr(module, attribute_name) if ( isinstance(attribute, type) and issubclass(attribute, PluginBase) and attribute is not PluginBase ): try: plugin_instance = attribute() self._register_plugin(plugin_instance) except Exception as e: print(f"Error instantiating plugin {attribute_name} in {module_name}: {e}") # Replace with proper logging # [/DEF:_load_module:Function] # [DEF:_register_plugin:Function] # @PURPOSE: Registers a PluginBase instance and its configuration. # @PRE: plugin_instance is a valid implementation of PluginBase. # @POST: Plugin is added to _plugins and _plugin_configs. # @PARAM: plugin_instance (PluginBase) - The plugin instance to register. def _register_plugin(self, plugin_instance: PluginBase): with belief_scope("_register_plugin"): """ Registers a valid plugin instance. """ plugin_id = plugin_instance.id if plugin_id in self._plugins: print(f"Warning: Duplicate plugin ID '{plugin_id}' found. Skipping.") # Replace with proper logging return try: schema = plugin_instance.get_schema() # Basic validation to ensure it's a dictionary if not isinstance(schema, dict): raise TypeError("get_schema() must return a dictionary.") plugin_config = PluginConfig( id=plugin_instance.id, name=plugin_instance.name, description=plugin_instance.description, version=plugin_instance.version, ui_route=plugin_instance.ui_route, schema=schema, ) # The following line is commented out because it requires a schema to be passed to validate against. # The schema provided by the plugin is the one being validated, not the data. # validate(instance={}, schema=schema) self._plugins[plugin_id] = plugin_instance self._plugin_configs[plugin_id] = plugin_config from ..core.logger import logger logger.info(f"Plugin '{plugin_instance.name}' (ID: {plugin_id}) loaded successfully.") except Exception as e: from ..core.logger import logger logger.error(f"Error validating plugin '{plugin_instance.name}' (ID: {plugin_id}): {e}") # [/DEF:_register_plugin:Function] # [DEF:get_plugin:Function] # @PURPOSE: Retrieves a loaded plugin instance by its ID. # @PRE: plugin_id is a string. # @POST: Returns plugin instance or None. # @PARAM: plugin_id (str) - The unique identifier of the plugin. # @RETURN: Optional[PluginBase] - The plugin instance if found, otherwise None. def get_plugin(self, plugin_id: str) -> Optional[PluginBase]: with belief_scope("get_plugin"): """ Returns a loaded plugin instance by its ID. """ return self._plugins.get(plugin_id) # [/DEF:get_plugin:Function] # [DEF:get_all_plugin_configs:Function] # @PURPOSE: Returns a list of all registered plugin configurations. # @PRE: None. # @POST: Returns list of all PluginConfig objects. # @RETURN: List[PluginConfig] - A list of plugin configurations. def get_all_plugin_configs(self) -> List[PluginConfig]: with belief_scope("get_all_plugin_configs"): """ Returns a list of all loaded plugin configurations. """ return list(self._plugin_configs.values()) # [/DEF:get_all_plugin_configs:Function] # [DEF:has_plugin:Function] # @PURPOSE: Checks if a plugin with the given ID is registered. # @PRE: plugin_id is a string. # @POST: Returns True if plugin exists. # @PARAM: plugin_id (str) - The unique identifier of the plugin. # @RETURN: bool - True if the plugin is registered, False otherwise. def has_plugin(self, plugin_id: str) -> bool: with belief_scope("has_plugin"): """ Checks if a plugin with the given ID is loaded. """ return plugin_id in self._plugins # [/DEF:has_plugin:Function] # [/DEF:PluginLoader:Class]