markitect-main/markitect/plugins/manager.py

"""
Plugin manager for discovering, loading, and managing plugins.

This module provides the main interface for plugin management in MarkiTect.
"""

import importlib
import importlib.util
import sys
from pathlib import Path
from typing import Dict, List, Optional, Any, Type
import yaml
import json

from .base import BasePlugin, PluginType, get_plugin_class_from_module
from .registry import plugin_registry


class PluginManager:
    """Main plugin manager for MarkiTect."""

    def __init__(self, config_path: Optional[str] = None):
        """
        Initialize plugin manager.

        Args:
            config_path: Optional path to plugin configuration file
        """
        self.config = self._load_config(config_path)
        self.plugin_directories = self._get_plugin_directories()
        self._discovered_plugins: Dict[str, Dict[str, Any]] = {}

    def discover_plugins(self, refresh: bool = False) -> Dict[str, Dict[str, Any]]:
        """
        Discover all available plugins.

        Args:
            refresh: Force refresh of plugin discovery

        Returns:
            Dictionary of discovered plugins with metadata
        """
        if self._discovered_plugins and not refresh:
            return self._discovered_plugins

        self._discovered_plugins = {}

        # Discover built-in plugins
        self._discover_builtin_plugins()

        # Discover plugins in configured directories
        for directory in self.plugin_directories:
            self._discover_plugins_in_directory(directory)

        # Discover plugins from installed packages
        self._discover_installed_plugins()

        return self._discovered_plugins

    def load_plugin(self, name: str, config: Dict[str, Any] = None) -> Optional[BasePlugin]:
        """
        Load a specific plugin by name.

        Args:
            name: Plugin name
            config: Optional plugin configuration

        Returns:
            Loaded plugin instance or None if failed
        """
        try:
            # Check if already loaded in registry
            if plugin_registry.is_loaded(name):
                return plugin_registry.get_plugin(name, config)

            # Check if plugin is already registered but not loaded
            if name in plugin_registry._plugins:
                return plugin_registry.get_plugin(name, config)

            # Try to discover and load
            discovered = self.discover_plugins()
            if name not in discovered:
                return None
        except Exception:
            # Add debugging info but don't break - continue to discovery logic
            pass

        plugin_info = discovered[name]
        try:
            # Load plugin module
            if 'module_path' in plugin_info:
                module = self._load_module_from_path(plugin_info['module_path'])
            elif 'module_name' in plugin_info:
                module = importlib.import_module(plugin_info['module_name'])
            else:
                return None

            # Find plugin class
            plugin_classes = get_plugin_class_from_module(module)
            if not plugin_classes:
                return None

            # Use first plugin class found (or specific one if specified)
            plugin_class = plugin_classes[0]
            if 'class_name' in plugin_info:
                for cls in plugin_classes:
                    if cls.__name__ == plugin_info['class_name']:
                        plugin_class = cls
                        break

            # Register and load
            plugin_registry.register(plugin_class, name)
            return plugin_registry.get_plugin(name, config)

        except Exception:
            return None

    def load_enabled_plugins(self) -> Dict[str, BasePlugin]:
        """
        Load all plugins marked as enabled in configuration.

        Returns:
            Dictionary of loaded plugins
        """
        enabled_plugins = {}
        plugin_configs = self.config.get('plugins', {})

        for plugin_name, plugin_config in plugin_configs.items():
            if plugin_config.get('enabled', False):
                plugin = self.load_plugin(plugin_name, plugin_config.get('config', {}))
                if plugin:
                    enabled_plugins[plugin_name] = plugin

        return enabled_plugins

    def get_plugins_by_type(self, plugin_type: PluginType) -> List[BasePlugin]:
        """
        Get all loaded plugins of a specific type.

        Args:
            plugin_type: Type of plugins to retrieve

        Returns:
            List of loaded plugin instances
        """
        plugin_names = plugin_registry.get_plugins_by_type(plugin_type)
        plugins = []

        for name in plugin_names:
            plugin = plugin_registry.get_plugin(name)
            if plugin:
                plugins.append(plugin)

        return plugins

    def unload_plugin(self, name: str) -> bool:
        """
        Unload a plugin.

        Args:
            name: Plugin name to unload

        Returns:
            True if unloaded successfully
        """
        return plugin_registry.unregister(name)

    def reload_plugin(self, name: str, config: Dict[str, Any] = None) -> Optional[BasePlugin]:
        """
        Reload a plugin with new configuration.

        Args:
            name: Plugin name
            config: New configuration

        Returns:
            Reloaded plugin instance or None if failed
        """
        if plugin_registry.reload_plugin(name, config):
            return plugin_registry.get_plugin(name)
        return None

    def list_plugins(self, plugin_type: Optional[PluginType] = None) -> Dict[str, Dict[str, Any]]:
        """
        List all available plugins.

        Args:
            plugin_type: Optional type filter

        Returns:
            Dictionary of plugins with metadata
        """
        all_plugins = plugin_registry.list_plugins()

        if plugin_type:
            filtered_plugins = {}
            for name, info in all_plugins.items():
                if info.get('type') == plugin_type.value:
                    filtered_plugins[name] = info
            return filtered_plugins

        return all_plugins

    def validate_plugin_dependencies(self, plugin_name: str) -> List[str]:
        """
        Validate plugin dependencies.

        Args:
            plugin_name: Plugin to validate

        Returns:
            List of missing dependencies
        """
        # This is a simplified implementation
        # In a full implementation, you'd check MarkiTect version,
        # required packages, etc.
        return []

    def _load_config(self, config_path: Optional[str] = None) -> Dict[str, Any]:
        """Load plugin configuration."""
        if config_path is None:
            # Try multiple default locations
            possible_paths = [
                Path('.markitect/plugins.yml'),
                Path('.markitect/plugins.yaml'),
                Path('.markitect/plugins.json'),
                Path('markitect_plugins.yml'),
                Path('markitect_plugins.yaml'),
                Path('markitect_plugins.json')
            ]
        else:
            possible_paths = [Path(config_path)]

        for path in possible_paths:
            if path.exists():
                try:
                    with open(path, 'r') as f:
                        if path.suffix.lower() in ['.yml', '.yaml']:
                            return yaml.safe_load(f) or {}
                        elif path.suffix.lower() == '.json':
                            return json.load(f)
                except Exception:
                    continue

        # Return default configuration
        return {
            'plugin_directories': [
                'plugins',
                '.markitect/plugins',
                str(Path.home() / '.markitect' / 'plugins')
            ],
            'auto_discover': True,
            'auto_load_enabled': True,
            'plugins': {}
        }

    def _get_plugin_directories(self) -> List[Path]:
        """Get list of directories to search for plugins."""
        directories = []
        for dir_path in self.config.get('plugin_directories', []):
            path = Path(dir_path)
            if path.exists() and path.is_dir():
                directories.append(path)
        return directories

    def _discover_builtin_plugins(self) -> None:
        """Discover built-in plugins in the markitect package."""
        try:
            # Import built-in plugin modules
            builtin_modules = [
                'markitect.plugins.builtin.processors',
                'markitect.plugins.builtin.formatters',
                'markitect.plugins.builtin.validators',
                'markitect.plugins.builtin.exporters'
            ]

            for module_name in builtin_modules:
                try:
                    # Import the module, which will trigger @register_plugin decorators
                    module = importlib.import_module(module_name)
                    self._register_plugins_from_module(module, f"builtin.{module_name.split('.')[-1]}")
                except ImportError:
                    continue
        except Exception:
            pass

    def _discover_plugins_in_directory(self, directory: Path) -> None:
        """Discover plugins in a specific directory."""
        for plugin_path in directory.glob('*.py'):
            if plugin_path.name.startswith('__'):
                continue

            try:
                module = self._load_module_from_path(plugin_path)
                self._register_plugins_from_module(module, plugin_path.stem)
            except Exception:
                continue

    def _discover_installed_plugins(self) -> None:
        """Discover plugins from installed Python packages."""
        # This would use entry points to discover plugins
        # For now, this is a placeholder
        pass

    def _load_module_from_path(self, module_path: Path):
        """Load a Python module from file path."""
        spec = importlib.util.spec_from_file_location(module_path.stem, module_path)
        if spec and spec.loader:
            module = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(module)
            return module
        return None

    def _register_plugins_from_module(self, module, module_name: str) -> None:
        """Register all plugins found in a module."""
        plugin_classes = get_plugin_class_from_module(module)

        for plugin_class in plugin_classes:
            # For built-in plugins, importing the module will auto-register them
            # via the @register_plugin decorator, so we just need to track them
            plugin_name = f"{module_name}.{plugin_class.__name__}"
            self._discovered_plugins[plugin_name] = {
                'class_name': plugin_class.__name__,
                'module_name': module.__name__ if hasattr(module, '__name__') else module_name,
                'discovered_from': 'module'
            }

            # Also add by registered name if it exists in registry
            # This happens after the @register_plugin decorator runs
            try:
                instance = plugin_class()
                registered_name = instance.metadata.name
                if registered_name and registered_name in plugin_registry._plugins:
                    # Add an alias for the registered name
                    self._discovered_plugins[registered_name] = {
                        'class_name': plugin_class.__name__,
                        'module_name': module.__name__ if hasattr(module, '__name__') else module_name,
                        'discovered_from': 'module',
                        'alias_for': plugin_name
                    }
            except Exception:
                pass