""" EU-Utility - Plugin Dependency Manager Manages plugin dependencies and auto-installs them when needed. """ import sys import subprocess import importlib from typing import Dict, List, Tuple, Optional, Any from dataclasses import dataclass from pathlib import Path import json @dataclass class DependencyCheck: """Result of a dependency check.""" name: str required: bool installed: bool dep_type: str = 'pip' # 'pip' or 'plugin' version: Optional[str] = None error: Optional[str] = None @dataclass class PluginDependencyCheck: """Result of a plugin dependency check.""" plugin_id: str installed: bool enabled: bool = False error: Optional[str] = None class PluginDependencyManager: """Manages plugin dependencies with auto-install capability.""" def __init__(self, plugin_manager=None): self.plugin_manager = plugin_manager self.install_status_file = Path("config/dependency_installs.json") self._install_status = self._load_install_status() def _load_install_status(self) -> Dict: """Load status of previously installed dependencies.""" if self.install_status_file.exists(): try: with open(self.install_status_file, 'r') as f: return json.load(f) except: pass return {} def _save_install_status(self): """Save status of installed dependencies.""" try: self.install_status_file.parent.mkdir(parents=True, exist_ok=True) with open(self.install_status_file, 'w') as f: json.dump(self._install_status, f, indent=2) except Exception as e: print(f"[DependencyManager] Failed to save status: {e}") def check_dependency(self, package_name: str) -> DependencyCheck: """Check if a pip package is installed.""" # Handle version specifiers (e.g., "package>=1.0") base_name = package_name.split('[')[0].split('>')[0].split('<')[0].split('=')[0].strip() try: module = importlib.import_module(base_name.replace('-', '_')) version = getattr(module, '__version__', 'unknown') return DependencyCheck( name=package_name, required=True, installed=True, version=version ) except ImportError: return DependencyCheck( name=package_name, required=True, installed=False ) except Exception as e: return DependencyCheck( name=package_name, required=True, installed=False, error=str(e) ) def check_plugin_dependency(self, plugin_id: str) -> PluginDependencyCheck: """Check if a required plugin is available and enabled.""" if not self.plugin_manager: return PluginDependencyCheck( plugin_id=plugin_id, installed=False, error="Plugin manager not available" ) # Check if plugin is discovered (available) all_plugins = self.plugin_manager.get_all_discovered_plugins() if plugin_id not in all_plugins: return PluginDependencyCheck( plugin_id=plugin_id, installed=False, error=f"Plugin '{plugin_id}' not found" ) # Check if plugin is enabled is_enabled = self.plugin_manager.is_plugin_enabled(plugin_id) return PluginDependencyCheck( plugin_id=plugin_id, installed=True, enabled=is_enabled ) def check_all_dependencies(self, plugin_class) -> Tuple[List[DependencyCheck], List[DependencyCheck], List[PluginDependencyCheck], List[PluginDependencyCheck]]: """Check all dependencies for a plugin (both pip and plugin dependencies). Returns: Tuple of (installed_pip, missing_pip, installed_plugins, missing_plugins) """ deps = getattr(plugin_class, 'dependencies', {}) pip_deps = deps.get('pip', []) plugin_deps = deps.get('plugins', []) # Check pip dependencies pip_installed = [] pip_missing = [] for dep in pip_deps: check = self.check_dependency(dep) if check.installed: pip_installed.append(check) else: pip_missing.append(check) # Check plugin dependencies plugin_installed = [] plugin_missing = [] for dep in plugin_deps: check = self.check_plugin_dependency(dep) if check.installed and check.enabled: plugin_installed.append(check) else: plugin_missing.append(check) return pip_installed, pip_missing, plugin_installed, plugin_missing def install_dependency(self, package_name: str) -> Tuple[bool, str]: """Install a pip package. Returns: (success, message) """ try: print(f"[DependencyManager] Installing {package_name}...") result = subprocess.run( [sys.executable, '-m', 'pip', 'install', package_name], capture_output=True, text=True, timeout=300 ) if result.returncode == 0: self._install_status[package_name] = { 'installed': True, 'output': result.stdout[-500:] if len(result.stdout) > 500 else result.stdout } self._save_install_status() return True, f"Successfully installed {package_name}" else: return False, f"Failed to install {package_name}: {result.stderr}" except subprocess.TimeoutExpired: return False, f"Installation of {package_name} timed out" except Exception as e: return False, f"Error installing {package_name}: {e}" def install_all_missing(self, plugin_class, progress_callback=None) -> Tuple[bool, List[str]]: """Install all missing pip dependencies for a plugin. Returns: (all_successful, messages) """ _, pip_missing, _, _ = self.check_all_dependencies(plugin_class) if not pip_missing: return True, ["All pip dependencies already installed"] messages = [] all_success = True for dep in pip_missing: if progress_callback: progress_callback(f"Installing {dep.name}...") success, msg = self.install_dependency(dep.name) messages.append(msg) if not success: all_success = False return all_success, messages def get_missing_dependencies_text(self, plugin_class) -> str: """Get a formatted text of all missing dependencies (pip and plugin).""" _, pip_missing, _, plugin_missing = self.check_all_dependencies(plugin_class) lines = [] if pip_missing: lines.append(f"Missing Python packages ({len(pip_missing)}):") for dep in pip_missing: lines.append(f" • {dep.name}") if plugin_missing: lines.append(f"\nRequired plugins ({len(plugin_missing)}) - must be enabled:") for dep in plugin_missing: status = "not found" if not dep.installed else "disabled" lines.append(f" • {dep.plugin_id} ({status})") if not lines: return "All dependencies are installed." return "\n".join(lines) def get_plugin_dependencies_text(self, plugin_class) -> str: """Get formatted text of plugin dependencies specifically.""" _, _, _, plugin_missing = self.check_all_dependencies(plugin_class) if not plugin_missing: return "" lines = [f"This plugin requires the following plugins to be enabled:"] for dep in plugin_missing: lines.append(f" • {dep.plugin_id}") return "\n".join(lines) def has_dependencies(self, plugin_class) -> bool: """Check if a plugin has declared dependencies (pip or plugin).""" deps = getattr(plugin_class, 'dependencies', {}) return bool(deps.get('pip', []) or deps.get('plugins', [])) def has_plugin_dependencies(self, plugin_class) -> bool: """Check if a plugin depends on other plugins.""" deps = getattr(plugin_class, 'dependencies', {}) return bool(deps.get('plugins', [])) # Singleton instance _dep_manager = None def get_dependency_manager(plugin_manager=None) -> PluginDependencyManager: """Get the plugin dependency manager singleton.""" global _dep_manager if _dep_manager is None: _dep_manager = PluginDependencyManager(plugin_manager) return _dep_manager