371 lines
13 KiB
Python
371 lines
13 KiB
Python
"""
|
|
Auto Dependency Plugin - 依赖自动安装插件
|
|
|
|
该插件允许其他插件在声明文件 (manifest.json) 中声明所需的系统依赖，
|
|
然后扫描所有插件的声明文件，检查并安装缺失的系统依赖。
|
|
|
|
通过插件加载器的 /PL 注入能力接口进行对接。
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from typing import List, Dict, Any, Optional
|
|
from pathlib import Path
|
|
|
|
from oss.plugin.base import BasePlugin
|
|
from oss.core.context import Context
|
|
|
|
|
|
class AutoDependencyPlugin(BasePlugin):
|
|
"""依赖自动安装插件"""
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.name = "auto_dependency"
|
|
self.version = "1.0.0"
|
|
self.description = "自动扫描并安装插件声明的系统依赖"
|
|
self.plugins_dir: Optional[Path] = None
|
|
self.manifest_filename = "manifest.json"
|
|
self.logger = None
|
|
|
|
def init(self, deps: Optional[Dict[str, Any]] = None):
|
|
"""初始化插件"""
|
|
# 获取插件目录路径
|
|
self.plugins_dir = Path(__file__).parent
|
|
if deps and 'logger' in deps:
|
|
self.logger = deps['logger']
|
|
else:
|
|
import logging
|
|
self.logger = logging.getLogger(self.name)
|
|
|
|
self.logger.info(f"AutoDependencyPlugin 初始化完成，插件目录：{self.plugins_dir}")
|
|
|
|
def start(self):
|
|
"""启动插件"""
|
|
self.logger.info("AutoDependencyPlugin 启动")
|
|
|
|
def stop(self):
|
|
"""停止插件"""
|
|
self.logger.info("AutoDependencyPlugin 停止")
|
|
|
|
def scan(self) -> List[Dict[str, Any]]:
|
|
"""
|
|
扫描所有插件的声明文件，收集系统依赖信息
|
|
|
|
Returns:
|
|
List[Dict]: 包含所有插件依赖信息的列表
|
|
每个元素格式: {
|
|
"plugin": str, # 插件名称
|
|
"dependencies": List, # 依赖列表
|
|
"package_manager": str # 包管理器类型
|
|
}
|
|
"""
|
|
all_dependencies = []
|
|
|
|
if not self.plugins_dir.exists():
|
|
self.logger.warning(f"插件目录不存在: {self.plugins_dir}")
|
|
return all_dependencies
|
|
|
|
# 遍历所有插件文件
|
|
for plugin_file in self.plugins_dir.glob("*.py"):
|
|
plugin_name = plugin_file.stem
|
|
|
|
# 跳过自身和__init__等文件
|
|
if plugin_name.startswith("_") or plugin_name == self.name:
|
|
continue
|
|
|
|
# 查找对应的 manifest 文件
|
|
manifest_path = self._find_manifest_for_plugin(plugin_name)
|
|
|
|
if manifest_path and manifest_path.exists():
|
|
try:
|
|
with open(manifest_path, 'r', encoding='utf-8') as f:
|
|
manifest = json.load(f)
|
|
|
|
# 提取系统依赖信息
|
|
system_deps = manifest.get("system_dependencies", [])
|
|
package_manager = manifest.get("package_manager", "apt-get")
|
|
|
|
if system_deps:
|
|
all_dependencies.append({
|
|
"plugin": plugin_name,
|
|
"dependencies": system_deps,
|
|
"package_manager": package_manager,
|
|
"manifest_path": str(manifest_path)
|
|
})
|
|
|
|
self.logger.info(
|
|
f"插件 {plugin_name} 声明了 {len(system_deps)} 个系统依赖"
|
|
)
|
|
|
|
except json.JSONDecodeError as e:
|
|
self.logger.error(f"解析 {manifest_path} 失败: {e}")
|
|
except Exception as e:
|
|
self.logger.error(f"处理插件 {plugin_name} 时出错: {e}")
|
|
|
|
return all_dependencies
|
|
|
|
def check(self, dependencies: List[Dict[str, Any]]) -> Dict[str, Any]:
|
|
"""
|
|
检查指定的系统依赖是否已安装
|
|
|
|
Args:
|
|
dependencies: 依赖信息列表，格式同 scan() 返回值
|
|
|
|
Returns:
|
|
Dict: 检查结果
|
|
{
|
|
"total": int, # 总依赖数
|
|
"installed": int, # 已安装数
|
|
"missing": List[Dict], # 缺失的依赖详情
|
|
"all_installed": bool # 是否全部已安装
|
|
}
|
|
"""
|
|
result = {
|
|
"total": 0,
|
|
"installed": 0,
|
|
"missing": [],
|
|
"all_installed": True
|
|
}
|
|
|
|
for dep_info in dependencies:
|
|
plugin_name = dep_info["plugin"]
|
|
package_manager = dep_info["package_manager"]
|
|
|
|
for package in dep_info["dependencies"]:
|
|
result["total"] += 1
|
|
|
|
if self._is_package_installed(package, package_manager):
|
|
result["installed"] += 1
|
|
self.logger.debug(f"包 {package} 已安装 (插件: {plugin_name})")
|
|
else:
|
|
result["missing"].append({
|
|
"package": package,
|
|
"plugin": plugin_name,
|
|
"package_manager": package_manager
|
|
})
|
|
result["all_installed"] = False
|
|
self.logger.warning(f"包 {package} 未安装 (插件: {plugin_name})")
|
|
|
|
return result
|
|
|
|
def install(self, missing: List[Dict[str, str]],
|
|
auto_confirm: bool = True) -> Dict[str, Any]:
|
|
"""
|
|
安装缺失的系统依赖
|
|
|
|
Args:
|
|
missing: 缺失的依赖列表，格式为 [{"package": str, "package_manager": str}]
|
|
auto_confirm: 是否自动确认安装
|
|
|
|
Returns:
|
|
Dict: 安装结果
|
|
{
|
|
"success": List[str], # 成功安装的包
|
|
"failed": List[Dict], # 安装失败的包及原因
|
|
"total": int, # 尝试安装的总数
|
|
}
|
|
"""
|
|
result = {
|
|
"success": [],
|
|
"failed": [],
|
|
"total": len(missing)
|
|
}
|
|
|
|
if not missing:
|
|
self.logger.info("没有需要安装的依赖")
|
|
return result
|
|
|
|
# 按包管理器分组
|
|
packages_by_pm: Dict[str, List[str]] = {}
|
|
for item in missing:
|
|
pm = item.get("package_manager", "apt-get")
|
|
pkg = item["package"]
|
|
|
|
if pm not in packages_by_pm:
|
|
packages_by_pm[pm] = []
|
|
packages_by_pm[pm].append(pkg)
|
|
|
|
# 执行安装
|
|
for pm, packages in packages_by_pm.items():
|
|
self.logger.info(f"使用 {pm} 安装包: {', '.join(packages)}")
|
|
|
|
success, failed = self._install_packages(packages, pm, auto_confirm)
|
|
|
|
result["success"].extend(success)
|
|
for fail_pkg, reason in failed:
|
|
result["failed"].append({
|
|
"package": fail_pkg,
|
|
"reason": reason
|
|
})
|
|
|
|
return result
|
|
|
|
def info(self) -> Dict[str, Any]:
|
|
"""
|
|
获取插件信息
|
|
|
|
Returns:
|
|
Dict: 插件详细信息
|
|
"""
|
|
return {
|
|
"name": self.name,
|
|
"version": self.version,
|
|
"description": self.description,
|
|
"supported_package_managers": [
|
|
"apt-get", "yum", "dnf", "pacman", "brew", "apk"
|
|
],
|
|
"api_methods": ["scan", "check", "install", "info"]
|
|
}
|
|
|
|
def _find_manifest_for_plugin(self, plugin_name: str) -> Optional[Path]:
|
|
"""查找插件对应的 manifest 文件"""
|
|
# 可能的 manifest 文件位置
|
|
possible_paths = [
|
|
self.plugins_dir / f"{plugin_name}.json",
|
|
self.plugins_dir / plugin_name / "manifest.json",
|
|
self.plugins_dir / f"{plugin_name}" / f"{plugin_name}.json",
|
|
]
|
|
|
|
for path in possible_paths:
|
|
if path.exists():
|
|
return path
|
|
|
|
# 也检查插件文件同目录下的同名 json 文件
|
|
plugin_file = self.plugins_dir / f"{plugin_name}.py"
|
|
if plugin_file.exists():
|
|
json_file = self.plugins_dir / f"{plugin_name}.json"
|
|
if json_file.exists():
|
|
return json_file
|
|
|
|
return None
|
|
|
|
def _is_package_installed(self, package: str, package_manager: str) -> bool:
|
|
"""检查包是否已安装"""
|
|
try:
|
|
if package_manager in ["apt-get", "apt"]:
|
|
cmd = ["dpkg", "-l", package]
|
|
elif package_manager in ["yum", "dnf"]:
|
|
cmd = ["rpm", "-q", package]
|
|
elif package_manager == "pacman":
|
|
cmd = ["pacman", "-Q", package]
|
|
elif package_manager == "brew":
|
|
cmd = ["brew", "list", "--versions", package]
|
|
elif package_manager == "apk":
|
|
cmd = ["apk", "info", "-e", package]
|
|
else:
|
|
# 默认使用 which/whereis 检查可执行文件
|
|
cmd = ["which", package]
|
|
|
|
result = subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30
|
|
)
|
|
|
|
return result.returncode == 0
|
|
|
|
except subprocess.TimeoutExpired:
|
|
self.logger.warning(f"检查包 {package} 超时")
|
|
return False
|
|
except Exception as e:
|
|
self.logger.error(f"检查包 {package} 时出错: {e}")
|
|
return False
|
|
|
|
def _install_packages(self, packages: List[str],
|
|
package_manager: str,
|
|
auto_confirm: bool = True) -> tuple:
|
|
"""
|
|
安装包
|
|
|
|
Returns:
|
|
tuple: (success_list, failed_list)
|
|
success_list: 成功安装的包名列表
|
|
failed_list: [(包名, 失败原因), ...]
|
|
"""
|
|
success = []
|
|
failed = []
|
|
|
|
try:
|
|
if package_manager in ["apt-get", "apt"]:
|
|
cmd_prefix = ["apt-get", "install", "-y"] if auto_confirm else ["apt-get", "install"]
|
|
elif package_manager == "yum":
|
|
cmd_prefix = ["yum", "install", "-y"] if auto_confirm else ["yum", "install"]
|
|
elif package_manager == "dnf":
|
|
cmd_prefix = ["dnf", "install", "-y"] if auto_confirm else ["dnf", "install"]
|
|
elif package_manager == "pacman":
|
|
cmd_prefix = ["pacman", "-S", "--noconfirm"] if auto_confirm else ["pacman", "-S"]
|
|
elif package_manager == "brew":
|
|
cmd_prefix = ["brew", "install"]
|
|
elif package_manager == "apk":
|
|
cmd_prefix = ["apk", "add"]
|
|
else:
|
|
self.logger.error(f"不支持的包管理器: {package_manager}")
|
|
for pkg in packages:
|
|
failed.append((pkg, f"不支持的包管理器: {package_manager}"))
|
|
return success, failed
|
|
|
|
# 合并命令
|
|
cmd = cmd_prefix + packages
|
|
|
|
self.logger.info(f"执行安装命令: {' '.join(cmd)}")
|
|
|
|
result = subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=300 # 5 分钟超时
|
|
)
|
|
|
|
if result.returncode == 0:
|
|
success.extend(packages)
|
|
self.logger.info(f"成功安装包: {', '.join(packages)}")
|
|
else:
|
|
error_msg = result.stderr.strip() or result.stdout.strip()
|
|
for pkg in packages:
|
|
failed.append((pkg, error_msg))
|
|
self.logger.error(f"安装包失败: {error_msg}")
|
|
|
|
except subprocess.TimeoutExpired:
|
|
for pkg in packages:
|
|
failed.append((pkg, "安装超时"))
|
|
self.logger.error("安装包超时")
|
|
except PermissionError:
|
|
for pkg in packages:
|
|
failed.append((pkg, "权限不足，需要 root 权限"))
|
|
self.logger.error("安装包需要 root 权限")
|
|
except Exception as e:
|
|
for pkg in packages:
|
|
failed.append((pkg, str(e)))
|
|
self.logger.error(f"安装包时发生异常: {e}")
|
|
|
|
return success, failed
|
|
|
|
def execute(self, action: str, **kwargs) -> Any:
|
|
"""
|
|
执行插件动作 (供插件加载器调用)
|
|
|
|
Args:
|
|
action: 动作名称 (scan, check, install, info)
|
|
**kwargs: 动作参数
|
|
|
|
Returns:
|
|
动作执行结果
|
|
"""
|
|
if action == "scan":
|
|
return self.scan()
|
|
elif action == "check":
|
|
dependencies = kwargs.get("dependencies", self.scan())
|
|
return self.check(dependencies)
|
|
elif action == "install":
|
|
missing = kwargs.get("missing", [])
|
|
auto_confirm = kwargs.get("auto_confirm", True)
|
|
return self.install(missing, auto_confirm)
|
|
elif action == "info":
|
|
return self.info()
|
|
else:
|
|
raise ValueError(f"未知的动作: {action}")
|