""" 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 version: Optional[str] = None 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_all_dependencies(self, plugin_class) -> Tuple[List[DependencyCheck], List[DependencyCheck]]: """Check all dependencies for a plugin. Returns: Tuple of (installed_deps, missing_deps) """ deps = getattr(plugin_class, 'dependencies', {}) pip_deps = deps.get('pip', []) installed = [] missing = [] for dep in pip_deps: check = self.check_dependency(dep) if check.installed: installed.append(check) else: missing.append(check) return installed, 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 dependencies for a plugin. Returns: (all_successful, messages) """ _, missing = self.check_all_dependencies(plugin_class) if not missing: return True, ["All dependencies already installed"] messages = [] all_success = True for dep in 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 missing dependencies.""" _, missing = self.check_all_dependencies(plugin_class) if not missing: return "All dependencies are installed." lines = [f"Missing dependencies ({len(missing)}):"] for dep in missing: lines.append(f" • {dep.name}") return "\n".join(lines) def has_dependencies(self, plugin_class) -> bool: """Check if a plugin has declared dependencies.""" deps = getattr(plugin_class, 'dependencies', {}) return bool(deps.get('pip', [])) # 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