263 lines
9.0 KiB
Python
263 lines
9.0 KiB
Python
"""
|
|
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
|