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

插件管理API测试与修改通过

Browse Source
This commit is contained in:
UnCLAS-Prommer
2025-07-24 00:31:53 +08:00
parent ae675faaa0
commit d6284b6b4c
6 changed files with 302 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

270
src/plugins/built_in/plugin_management/plugin.py
View File

@@ -1,3 +1,5 @@
import asyncio
from typing import List, Tuple, Type
from src.plugin_system import (
BasePlugin,
@@ -10,18 +12,16 @@ from src.plugin_system import (
ComponentInfo,
ComponentType,
)
from src.plugin_system.base.base_action import BaseAction
from src.plugin_system.base.base_events_handler import BaseEventHandler
from src.plugin_system.base.component_types import ActionInfo, EventHandlerInfo
class ManagementCommand(BaseCommand):
command_name: str = "management"
description: str = "管理命令"
command_pattern: str = r"(?P<manage_command>^/p_m(\s[a-zA-Z0-9_]+)*\s*$)"
command_pattern: str = r"(?P<manage_command>^/pm(\s[a-zA-Z0-9_]+)*\s*$)"
intercept_message: bool = True
async def execute(self) -> Tuple[bool, str]:
# sourcery skip: merge-duplicate-blocks
command_list = self.matched_groups["manage_command"].strip().split(" ")
if len(command_list) == 1:
await self.show_help("all")
@@ -35,6 +35,7 @@ class ManagementCommand(BaseCommand):
case "help":
await self.show_help("all")
case _:
await self.send_text("插件管理命令不合法")
return False, "命令不合法"
if len(command_list) == 3:
if command_list[1] == "plugin":
@@ -48,17 +49,18 @@ class ManagementCommand(BaseCommand):
case "rescan":
await self._rescan_plugin_dirs()
case _:
await self.send_text("插件管理命令不合法")
return False, "命令不合法"
elif command_list[1] == "component":
match command_list[2]:
case "help":
await self.show_help("component")
return True, "帮助已发送"
case "list":
pass
case _:
return False, "命令不合法"
if command_list[2] == "list":
await self._list_all_registered_components()
elif command_list[2] == "help":
await self.show_help("component")
else:
await self.send_text("插件管理命令不合法")
return False, "命令不合法"
else:
await self.send_text("插件管理命令不合法")
return False, "命令不合法"
if len(command_list) == 4:
if command_list[1] == "plugin":
@@ -72,15 +74,61 @@ class ManagementCommand(BaseCommand):
case "add_dir":
await self._add_dir(command_list[3])
case _:
await self.send_text("插件管理命令不合法")
return False, "命令不合法"
elif command_list[1] == "component":
pass
if command_list[2] != "list":
await self.send_text("插件管理命令不合法")
return False, "命令不合法"
if command_list[3] == "enabled":
await self._list_enabled_components()
elif command_list[3] == "disabled":
await self._list_disabled_components()
else:
await self.send_text("插件管理命令不合法")
return False, "命令不合法"
else:
await self.send_text("插件管理命令不合法")
return False, "命令不合法"
if len(command_list) == 5:
pass
if command_list[1] != "component":
await self.send_text("插件管理命令不合法")
return False, "命令不合法"
if command_list[2] != "list":
await self.send_text("插件管理命令不合法")
return False, "命令不合法"
if command_list[3] == "enabled":
await self._list_enabled_components(target_type=command_list[4])
elif command_list[3] == "disabled":
await self._list_disabled_components(target_type=command_list[4])
elif command_list[3] == "type":
await self._list_registered_components_by_type(command_list[4])
else:
await self.send_text("插件管理命令不合法")
return False, "命令不合法"
if len(command_list) == 6:
pass
if command_list[1] != "component":
await self.send_text("插件管理命令不合法")
return False, "命令不合法"
if command_list[2] == "enable":
if command_list[3] == "global":
await self._globally_enable_component(command_list[4], command_list[5])
elif command_list[3] == "local":
await self._locally_enable_component(command_list[4], command_list[5])
else:
await self.send_text("插件管理命令不合法")
return False, "命令不合法"
elif command_list[2] == "disable":
if command_list[3] == "global":
await self._globally_disable_component(command_list[4], command_list[5])
elif command_list[3] == "local":
await self._locally_disable_component(command_list[4], command_list[5])
else:
await self.send_text("插件管理命令不合法")
return False, "命令不合法"
else:
await self.send_text("插件管理命令不合法")
return False, "命令不合法"
return True, "命令执行完成"
@@ -90,37 +138,37 @@ class ManagementCommand(BaseCommand):
case "all":
help_msg = (
"管理命令帮助\n"
"/p_m help 管理命令提示\n"
"/p_m plugin 插件管理命令\n"
"/p_m component 组件管理命令\n"
"使用 /p_m plugin help 或 /p_m component help 获取具体帮助"
"/pm help 管理命令提示\n"
"/pm plugin 插件管理命令\n"
"/pm component 组件管理命令\n"
"使用 /pm plugin help 或 /pm component help 获取具体帮助"
)
case "plugin":
help_msg = (
"插件管理命令帮助\n"
"/p_m plugin help 插件管理命令提示\n"
"/p_m plugin list 列出所有注册的插件\n"
"/p_m plugin list_enabled 列出所有加载（启用）的插件\n"
"/p_m plugin rescan 重新扫描所有目录\n"
"/p_m plugin load <plugin_name> 加载指定插件\n"
"/p_m plugin unload <plugin_name> 卸载指定插件\n"
"/p_m plugin reload <plugin_name> 重新加载指定插件\n"
"/p_m plugin add_dir <directory_path> 添加插件目录\n"
"/pm plugin help 插件管理命令提示\n"
"/pm plugin list 列出所有注册的插件\n"
"/pm plugin list_enabled 列出所有加载（启用）的插件\n"
"/pm plugin rescan 重新扫描所有目录\n"
"/pm plugin load <plugin_name> 加载指定插件\n"
"/pm plugin unload <plugin_name> 卸载指定插件\n"
"/pm plugin reload <plugin_name> 重新加载指定插件\n"
"/pm plugin add_dir <directory_path> 添加插件目录\n"
)
case "component":
help_msg = (
"组件管理命令帮助\n"
"/p_m component help 组件管理命令提示\n"
"/p_m component list 列出所有注册的组件\n"
"/p_m component list enabled <可选: type> 列出所有启用的组件\n"
"/p_m component list disabled <可选: type> 列出所有禁用的组件\n"
"/pm component help 组件管理命令提示\n"
"/pm component list 列出所有注册的组件\n"
"/pm component list enabled <可选: type> 列出所有启用的组件\n"
"/pm component list disabled <可选: type> 列出所有禁用的组件\n"
" - <type> 可选项: local，代表当前聊天中的；global，代表全局的\n"
" - <type> 不填时为 global\n"
"/p_m component list type <component_type> 列出指定类型的组件\n"
"/p_m component global enable <component_name> <可选: component_type> 全局启用组件\n"
"/p_m component global disable <component_name> <可选: component_type> 全局禁用组件\n"
"/p_m component local enable <component_name> <可选: component_type> 本聊天启用组件\n"
"/p_m component local disable <component_name> <可选: component_type> 本聊天禁用组件\n"
"/pm component list type <component_type> 列出已经注册的指定类型的组件\n"
"/pm component enable global <component_name> <component_type> 全局启用组件\n"
"/pm component enable local <component_name> <component_type> 本聊天启用组件\n"
"/pm component disable global <component_name> <component_type> 全局禁用组件\n"
"/pm component disable local <component_name> <component_type> 本聊天禁用组件\n"
" - <component_type> 可选项: action, command, event_handler\n"
)
case _:
@@ -140,7 +188,6 @@ class ManagementCommand(BaseCommand):
await self.send_text("插件目录重新扫描执行中")
async def _load_plugin(self, plugin_name: str):
await self.send_text(f"正在加载插件: {plugin_name}")
success, count = plugin_manage_api.load_plugin(plugin_name)
if success:
await self.send_text(f"插件加载成功: {plugin_name}")
@@ -150,16 +197,14 @@ class ManagementCommand(BaseCommand):
await self.send_text(f"插件加载失败: {plugin_name}")
async def _unload_plugin(self, plugin_name: str):
await self.send_text(f"正在卸载插件: {plugin_name}")
success = plugin_manage_api.remove_plugin(plugin_name)
success = await plugin_manage_api.remove_plugin(plugin_name)
if success:
await self.send_text(f"插件卸载成功: {plugin_name}")
else:
await self.send_text(f"插件卸载失败: {plugin_name}")
async def _reload_plugin(self, plugin_name: str):
await self.send_text(f"正在重新加载插件: {plugin_name}")
success = plugin_manage_api.reload_plugin(plugin_name)
success = await plugin_manage_api.reload_plugin(plugin_name)
if success:
await self.send_text(f"插件重新加载成功: {plugin_name}")
else:
@@ -168,6 +213,7 @@ class ManagementCommand(BaseCommand):
async def _add_dir(self, dir_path: str):
await self.send_text(f"正在添加插件目录: {dir_path}")
success = plugin_manage_api.add_plugin_directory(dir_path)
await asyncio.sleep(0.5) # 防止乱序发送
if success:
await self.send_text(f"插件目录添加成功: {dir_path}")
else:
@@ -219,15 +265,151 @@ class ManagementCommand(BaseCommand):
if target_type == "global":
enabled_components = [component for component in components_info if component.enabled]
if not enabled_components:
await self.send_text("没有启用的全局组件")
await self.send_text("没有满足条件的已启用全局组件")
return
enabled_components_str = ", ".join(
f"{component.name} ({component.component_type})" for component in enabled_components
)
await self.send_text(f"启用的全局组件: {enabled_components_str}")
await self.send_text(f"满足条件的已启用全局组件: {enabled_components_str}")
elif target_type == "local":
locally_disabled_components = self._fetch_locally_disabled_components()
enabled_components = [
component
for component in components_info
if (component.name not in locally_disabled_components and component.enabled)
]
if not enabled_components:
await self.send_text("本聊天没有满足条件的已启用组件")
return
enabled_components_str = ", ".join(
f"{component.name} ({component.component_type})" for component in enabled_components
)
await self.send_text(f"本聊天满足条件的已启用组件: {enabled_components_str}")
async def _list_disabled_components(self, target_type: str = "global"):
components_info = self._fetch_all_registered_components()
if not components_info:
await self.send_text("没有注册的组件")
return
if target_type == "global":
disabled_components = [component for component in components_info if not component.enabled]
if not disabled_components:
await self.send_text("没有满足条件的已禁用全局组件")
return
disabled_components_str = ", ".join(
f"{component.name} ({component.component_type})" for component in disabled_components
)
await self.send_text(f"满足条件的已禁用全局组件: {disabled_components_str}")
elif target_type == "local":
locally_disabled_components = self._fetch_locally_disabled_components()
disabled_components = [
component
for component in components_info
if (component.name in locally_disabled_components or not component.enabled)
]
if not disabled_components:
await self.send_text("本聊天没有满足条件的已禁用组件")
return
disabled_components_str = ", ".join(
f"{component.name} ({component.component_type})" for component in disabled_components
)
await self.send_text(f"本聊天满足条件的已禁用组件: {disabled_components_str}")
async def _list_registered_components_by_type(self, target_type: str):
match target_type:
case "action":
component_type = ComponentType.ACTION
case "command":
component_type = ComponentType.COMMAND
case "event_handler":
component_type = ComponentType.EVENT_HANDLER
case _:
await self.send_text(f"未知组件类型: {target_type}")
return
components_info = component_manage_api.get_components_info_by_type(component_type)
if not components_info:
await self.send_text(f"没有注册的 {target_type} 组件")
return
components_str = ", ".join(
f"{name} ({component.component_type})" for name, component in components_info.items()
)
await self.send_text(f"注册的 {target_type} 组件: {components_str}")
async def _globally_enable_component(self, component_name: str, component_type: str):
match component_type:
case "action":
target_component_type = ComponentType.ACTION
case "command":
target_component_type = ComponentType.COMMAND
case "event_handler":
target_component_type = ComponentType.EVENT_HANDLER
case _:
await self.send_text(f"未知组件类型: {component_type}")
return
if component_manage_api.globally_enable_component(component_name, target_component_type):
await self.send_text(f"全局启用组件成功: {component_name}")
else:
await self.send_text(f"全局启用组件失败: {component_name}")
async def _globally_disable_component(self, component_name: str, component_type: str):
match component_type:
case "action":
target_component_type = ComponentType.ACTION
case "command":
target_component_type = ComponentType.COMMAND
case "event_handler":
target_component_type = ComponentType.EVENT_HANDLER
case _:
await self.send_text(f"未知组件类型: {component_type}")
return
success = await component_manage_api.globally_disable_component(component_name, target_component_type)
if success:
await self.send_text(f"全局禁用组件成功: {component_name}")
else:
await self.send_text(f"全局禁用组件失败: {component_name}")
async def _locally_enable_component(self, component_name: str, component_type: str):
match component_type:
case "action":
target_component_type = ComponentType.ACTION
case "command":
target_component_type = ComponentType.COMMAND
case "event_handler":
target_component_type = ComponentType.EVENT_HANDLER
case _:
await self.send_text(f"未知组件类型: {component_type}")
return
if component_manage_api.locally_enable_component(
component_name,
target_component_type,
self.message.chat_stream.stream_id,
):
await self.send_text(f"本地启用组件成功: {component_name}")
else:
await self.send_text(f"本地启用组件失败: {component_name}")
async def _locally_disable_component(self, component_name: str, component_type: str):
match component_type:
case "action":
target_component_type = ComponentType.ACTION
case "command":
target_component_type = ComponentType.COMMAND
case "event_handler":
target_component_type = ComponentType.EVENT_HANDLER
case _:
await self.send_text(f"未知组件类型: {component_type}")
return
if component_manage_api.locally_disable_component(
component_name,
target_component_type,
self.message.chat_stream.stream_id,
):
await self.send_text(f"本地禁用组件成功: {component_name}")
else:
await self.send_text(f"本地禁用组件失败: {component_name}")
@register_plugin