gardel/Mofox-Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'dev' of https://github.com/MaiM-with-u/MaiBot into dev

Browse Source
This commit is contained in:
SengokuCola
2025-06-16 00:04:42 +08:00
parent 5fcedd1531 59f6bcb19b
commit 7e2e67f691
6 changed files with 95 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/plugin_system/base/base_action.py
View File

@@ -366,7 +366,7 @@ class BaseAction(ABC):
@classmethod
def get_action_info(cls) -> "ActionInfo":
"""从类属性生成ActionInfo
所有信息都从类属性中读取，确保一致性和完整性。
Action类必须定义所有必要的类属性。
@@ -376,7 +376,7 @@ class BaseAction(ABC):
# 从类属性读取名称，如果没有定义则使用类名自动生成
name = getattr(cls, "action_name", cls.__name__.lower().replace("action", ""))
# 从类属性读取描述，如果没有定义则使用文档字符串的第一行
description = getattr(cls, "action_description", None)
if description is None:

10
src/plugin_system/base/component_types.py
View File

@@ -36,17 +36,17 @@ class ChatMode(Enum):
@dataclass
class PythonDependency:
"""Python包依赖信息"""
package_name: str # 包名称
version: str = "" # 版本要求，例如: ">=1.0.0", "==2.1.3", ""表示任意版本
optional: bool = False # 是否为可选依赖
description: str = "" # 依赖描述
install_name: str = "" # 安装时的包名（如果与import名不同）
def __post_init__(self):
if not self.install_name:
self.install_name = self.package_name
def get_pip_requirement(self) -> str:
"""获取pip安装格式的依赖字符串"""
if self.version:
@@ -141,7 +141,7 @@ class PluginInfo:
self.python_dependencies = []
if self.metadata is None:
self.metadata = {}
def get_missing_packages(self) -> List[PythonDependency]:
"""检查缺失的Python包"""
missing = []
@@ -152,7 +152,7 @@ class PluginInfo:
if not dep.optional:
missing.append(dep)
return missing
def get_pip_requirements(self) -> List[str]:
"""获取所有pip安装格式的依赖"""
return [dep.get_pip_requirement() for dep in self.python_dependencies]

78
src/plugin_system/core/dependency_manager.py
View File

@@ -7,8 +7,7 @@
import subprocess
import sys
import importlib
from typing import List, Dict, Tuple, Optional
from pathlib import Path
from typing import List, Dict, Tuple
from src.common.logger import get_logger
from src.plugin_system.base.component_types import PythonDependency
@@ -22,19 +21,21 @@ class DependencyManager:
def __init__(self):
self.install_log: List[str] = []
self.failed_installs: Dict[str, str] = {}
def check_dependencies(self, dependencies: List[PythonDependency]) -> Tuple[List[PythonDependency], List[PythonDependency]]:
def check_dependencies(
self, dependencies: List[PythonDependency]
) -> Tuple[List[PythonDependency], List[PythonDependency]]:
"""检查依赖包状态
Args:
dependencies: 依赖包列表
Returns:
Tuple[List[PythonDependency], List[PythonDependency]]: (缺失的依赖, 可选缺失的依赖)
"""
missing_required = []
missing_optional = []
for dep in dependencies:
if not self._is_package_available(dep.package_name):
if dep.optional:
@@ -45,9 +46,9 @@ class DependencyManager:
logger.error(f"必需依赖包缺失: {dep.package_name} - {dep.description}")
else:
logger.debug(f"依赖包已存在: {dep.package_name}")
return missing_required, missing_optional
def _is_package_available(self, package_name: str) -> bool:
"""检查包是否可用"""
try:
@@ -55,34 +56,34 @@ class DependencyManager:
return True
except ImportError:
return False
def install_dependencies(self, dependencies: List[PythonDependency], auto_install: bool = False) -> bool:
"""安装依赖包
Args:
dependencies: 需要安装的依赖包列表
auto_install: 是否自动安装（True时不询问用户）
Returns:
bool: 安装是否成功
"""
if not dependencies:
return True
logger.info(f"需要安装 {len(dependencies)} 个依赖包")
# 显示将要安装的包
for dep in dependencies:
install_cmd = dep.get_pip_requirement()
logger.info(f" - {install_cmd} {'(可选)' if dep.optional else '(必需)'}")
if dep.description:
logger.info(f" 说明: {dep.description}")
if not auto_install:
# 这里可以添加用户确认逻辑
logger.warning("手动安装模式：请手动运行 pip install 命令安装依赖包")
return False
# 执行安装
success_count = 0
for dep in dependencies:
@@ -90,26 +91,26 @@ class DependencyManager:
success_count += 1
else:
self.failed_installs[dep.package_name] = f"安装失败: {dep.get_pip_requirement()}"
logger.info(f"依赖安装完成: {success_count}/{len(dependencies)} 个成功")
return success_count == len(dependencies)
def _install_single_package(self, dependency: PythonDependency) -> bool:
"""安装单个包"""
pip_requirement = dependency.get_pip_requirement()
try:
logger.info(f"正在安装: {pip_requirement}")
# 使用subprocess安装包
cmd = [sys.executable, "-m", "pip", "install", pip_requirement]
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=300 # 5分钟超时
timeout=300, # 5分钟超时
)
if result.returncode == 0:
logger.info(f"✅ 成功安装: {pip_requirement}")
self.install_log.append(f"成功安装: {pip_requirement}")
@@ -119,28 +120,29 @@ class DependencyManager:
logger.error(f"错误输出: {result.stderr}")
self.install_log.append(f"安装失败: {pip_requirement} - {result.stderr}")
return False
except subprocess.TimeoutExpired:
logger.error(f"❌ 安装超时: {pip_requirement}")
return False
except Exception as e:
logger.error(f"❌ 安装异常: {pip_requirement} - {str(e)}")
return False
def generate_requirements_file(self, plugins_dependencies: List[List[PythonDependency]],
output_path: str = "plugin_requirements.txt") -> bool:
def generate_requirements_file(
self, plugins_dependencies: List[List[PythonDependency]], output_path: str = "plugin_requirements.txt"
) -> bool:
"""生成插件依赖的requirements文件
Args:
plugins_dependencies: 所有插件的依赖列表
output_path: 输出文件路径
Returns:
bool: 生成是否成功
"""
try:
all_deps = {}
# 合并所有插件的依赖
for plugin_deps in plugins_dependencies:
for dep in plugin_deps:
@@ -152,39 +154,39 @@ class DependencyManager:
logger.warning(f"依赖版本冲突: {key} ({existing.version} vs {dep.version})")
else:
all_deps[key] = dep
# 写入requirements文件
with open(output_path, "w", encoding="utf-8") as f:
f.write("# 插件依赖包自动生成\n")
f.write("# Auto-generated plugin dependencies\n\n")
# 按包名排序
sorted_deps = sorted(all_deps.values(), key=lambda x: x.install_name)
for dep in sorted_deps:
requirement = dep.get_pip_requirement()
if dep.description:
f.write(f"# {dep.description}\n")
if dep.optional:
f.write(f"# Optional dependency\n")
f.write("# Optional dependency\n")
f.write(f"{requirement}\n\n")
logger.info(f"已生成插件依赖文件: {output_path} ({len(all_deps)} 个包)")
return True
except Exception as e:
logger.error(f"生成requirements文件失败: {str(e)}")
return False
def get_install_summary(self) -> Dict[str, any]:
"""获取安装摘要"""
return {
"install_log": self.install_log.copy(),
"failed_installs": self.failed_installs.copy(),
"total_attempts": len(self.install_log),
"failed_count": len(self.failed_installs)
"failed_count": len(self.failed_installs),
}
# 全局依赖管理器实例
dependency_manager = DependencyManager()
dependency_manager = DependencyManager()

71
src/plugin_system/core/plugin_manager.py
View File

@@ -10,7 +10,7 @@ if TYPE_CHECKING:
from src.common.logger import get_logger
from src.plugin_system.core.component_registry import component_registry
from src.plugin_system.core.dependency_manager import dependency_manager
from src.plugin_system.base.component_types import ComponentType, PluginInfo, PythonDependency
from src.plugin_system.base.component_types import ComponentType, PluginInfo
logger = get_logger("plugin_manager")
@@ -347,57 +347,55 @@ class PluginManager:
def check_all_dependencies(self, auto_install: bool = False) -> Dict[str, any]:
"""检查所有插件的Python依赖包
Args:
auto_install: 是否自动安装缺失的依赖包
Returns:
Dict[str, any]: 检查结果摘要
"""
logger.info("开始检查所有插件的Python依赖包...")
all_required_missing = []
all_optional_missing = []
plugin_status = {}
for plugin_name, plugin_instance in self.loaded_plugins.items():
for plugin_name, _plugin_instance in self.loaded_plugins.items():
plugin_info = component_registry.get_plugin_info(plugin_name)
if not plugin_info or not plugin_info.python_dependencies:
plugin_status[plugin_name] = {"status": "no_dependencies", "missing": []}
continue
logger.info(f"检查插件 {plugin_name} 的依赖...")
missing_required, missing_optional = dependency_manager.check_dependencies(
plugin_info.python_dependencies
)
missing_required, missing_optional = dependency_manager.check_dependencies(plugin_info.python_dependencies)
if missing_required:
all_required_missing.extend(missing_required)
plugin_status[plugin_name] = {
"status": "missing_required",
"status": "missing_required",
"missing": [dep.package_name for dep in missing_required],
"optional_missing": [dep.package_name for dep in missing_optional]
"optional_missing": [dep.package_name for dep in missing_optional],
}
logger.error(f"插件 {plugin_name} 缺少必需依赖: {[dep.package_name for dep in missing_required]}")
elif missing_optional:
all_optional_missing.extend(missing_optional)
plugin_status[plugin_name] = {
"status": "missing_optional",
"status": "missing_optional",
"missing": [],
"optional_missing": [dep.package_name for dep in missing_optional]
"optional_missing": [dep.package_name for dep in missing_optional],
}
logger.warning(f"插件 {plugin_name} 缺少可选依赖: {[dep.package_name for dep in missing_optional]}")
else:
plugin_status[plugin_name] = {"status": "ok", "missing": []}
logger.info(f"插件 {plugin_name} 依赖检查通过")
# 汇总结果
total_missing = len(set(dep.package_name for dep in all_required_missing))
total_optional_missing = len(set(dep.package_name for dep in all_optional_missing))
logger.info(f"依赖检查完成 - 缺少必需包: {total_missing}个, 缺少可选包: {total_optional_missing}个")
# 如果需要自动安装
install_success = True
if auto_install and all_required_missing:
@@ -405,47 +403,48 @@ class PluginManager:
unique_required = {}
for dep in all_required_missing:
unique_required[dep.package_name] = dep
logger.info(f"开始自动安装 {len(unique_required)} 个必需依赖包...")
install_success = dependency_manager.install_dependencies(
list(unique_required.values()),
auto_install=True
)
install_success = dependency_manager.install_dependencies(list(unique_required.values()), auto_install=True)
return {
"total_plugins_checked": len(plugin_status),
"plugins_with_missing_required": len([p for p in plugin_status.values() if p["status"] == "missing_required"]),
"plugins_with_missing_optional": len([p for p in plugin_status.values() if p["status"] == "missing_optional"]),
"plugins_with_missing_required": len(
[p for p in plugin_status.values() if p["status"] == "missing_required"]
),
"plugins_with_missing_optional": len(
[p for p in plugin_status.values() if p["status"] == "missing_optional"]
),
"total_missing_required": total_missing,
"total_missing_optional": total_optional_missing,
"plugin_status": plugin_status,
"auto_install_attempted": auto_install and bool(all_required_missing),
"auto_install_success": install_success,
"install_summary": dependency_manager.get_install_summary()
"install_summary": dependency_manager.get_install_summary(),
}
def generate_plugin_requirements(self, output_path: str = "plugin_requirements.txt") -> bool:
"""生成所有插件依赖的requirements文件
Args:
output_path: 输出文件路径
Returns:
bool: 生成是否成功
"""
logger.info("开始生成插件依赖requirements文件...")
all_dependencies = []
for plugin_name, plugin_instance in self.loaded_plugins.items():
for plugin_name, _plugin_instance in self.loaded_plugins.items():
plugin_info = component_registry.get_plugin_info(plugin_name)
if plugin_info and plugin_info.python_dependencies:
all_dependencies.append(plugin_info.python_dependencies)
if not all_dependencies:
logger.info("没有找到任何插件依赖")
return False
return dependency_manager.generate_requirements_file(all_dependencies, output_path)