From 28ec9139de874398b6f19d3c380f2c5769ecb4ab Mon Sep 17 00:00:00 2001 From: minecraft1024a Date: Sun, 2 Nov 2025 12:48:40 +0800 Subject: [PATCH] =?UTF-8?q?refactor(plugins):=20=E6=95=B4=E5=90=88?= =?UTF-8?q?=E6=9D=83=E9=99=90=E5=92=8C=E6=8F=92=E4=BB=B6=E7=AE=A1=E7=90=86?= =?UTF-8?q?=E5=8A=9F=E8=83=BD=E8=87=B3=E7=B3=BB=E7=BB=9F=E7=AE=A1=E7=90=86?= =?UTF-8?q?=E6=8F=92=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将原有的 `permission_management` 和 `plugin_management` 插件的功能合并到 `system_management` 插件中,以简化内置插件结构并统一管理入口。 - 删除了独立的 `permission_management` 和 `plugin_management` 插件目录。 - 将权限管理和插件管理的命令逻辑迁移至 `system_management` 插件的 `SystemCommand` 类中。 - 通过这种方式,核心的管理功能被集中到一个统一的 `/system` 命令下,提高了内聚性并减少了维护开销。 --- .../permission_management/__init__.py | 16 - .../built_in/permission_management/plugin.py | 398 -------------- .../built_in/plugin_management/__init__.py | 17 - .../built_in/plugin_management/plugin.py | 518 ------------------ .../built_in/system_management/plugin.py | 2 +- 5 files changed, 1 insertion(+), 950 deletions(-) delete mode 100644 src/plugins/built_in/permission_management/__init__.py delete mode 100644 src/plugins/built_in/permission_management/plugin.py delete mode 100644 src/plugins/built_in/plugin_management/__init__.py delete mode 100644 src/plugins/built_in/plugin_management/plugin.py diff --git a/src/plugins/built_in/permission_management/__init__.py b/src/plugins/built_in/permission_management/__init__.py deleted file mode 100644 index 1e58619ea..000000000 --- a/src/plugins/built_in/permission_management/__init__.py +++ /dev/null @@ -1,16 +0,0 @@ -from src.plugin_system.base.plugin_metadata import PluginMetadata - -__plugin_meta__ = PluginMetadata( - name="权限管理插件(Permission Management)", - description="通过系统API管理权限", - usage="该插件提供 `permission_management` command。", - version="1.0.0", - author="MoFox-Studio", - license="GPL-v3.0-or-later", - repository_url="https://github.com/MoFox-Studio", - keywords=["plugins", "permission", "management", "built-in"], - extra={ - "is_built_in": True, - "plugin_type": "permission", - }, -) diff --git a/src/plugins/built_in/permission_management/plugin.py b/src/plugins/built_in/permission_management/plugin.py deleted file mode 100644 index 066ee220f..000000000 --- a/src/plugins/built_in/permission_management/plugin.py +++ /dev/null @@ -1,398 +0,0 @@ -""" -权限管理插件 - -提供权限系统的管理命令,包括权限授权、撤销、查询等功能。 -使用新的PlusCommand系统重构。 -""" - -import re -from typing import ClassVar - -from src.plugin_system.apis.logging_api import get_logger -from src.plugin_system.apis.permission_api import permission_api -from src.plugin_system.apis.plugin_register_api import register_plugin -from src.plugin_system.base.base_plugin import BasePlugin -from src.plugin_system.base.command_args import CommandArgs -from src.plugin_system.base.component_types import ( - ChatType, - PermissionNodeField, - PlusCommandInfo, -) -from src.plugin_system.base.config_types import ConfigField -from src.plugin_system.base.plus_command import PlusCommand -from src.plugin_system.utils.permission_decorators import require_permission - -logger = get_logger("Permission") - - -class PermissionCommand(PlusCommand): - """权限管理命令 - 使用PlusCommand系统""" - - command_name = "permission" - command_description = "权限管理命令,支持授权、撤销、查询等功能" - command_aliases: ClassVar[list[str]] = ["perm", "权限"] - priority = 10 - chat_type_allow = ChatType.ALL - intercept_message = True - - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - - async def execute(self, args: CommandArgs) -> tuple[bool, str | None, bool]: - """执行权限管理命令""" - if args.is_empty: - await self._show_help() - return True, "显示帮助信息", True - - subcommand = args.get_first.lower() - remaining_args = args.get_args()[1:] # 获取除第一个参数外的所有参数 - chat_stream = self.message.chat_info.stream_id - - if subcommand in ["grant", "授权", "give"]: - await self._grant_permission(chat_stream, remaining_args) - return True, "执行授权命令", True - - elif subcommand in ["revoke", "撤销", "remove"]: - await self._revoke_permission(chat_stream, remaining_args) - return True, "执行撤销命令", True - - elif subcommand in ["list", "列表", "ls"]: - await self._list_permissions(chat_stream, remaining_args) - return True, "执行列表命令", True - - elif subcommand in ["check", "检查"]: - await self._check_permission(chat_stream, remaining_args) - return True, "执行检查命令", True - - elif subcommand in ["nodes", "节点"]: - await self._list_nodes(chat_stream, remaining_args) - return True, "执行节点命令", True - - elif subcommand in ["allnodes", "全部节点", "all"]: - await self._list_all_nodes_with_description(chat_stream) - return True, "执行全部节点命令", True - - elif subcommand in ["help", "帮助"]: - await self._show_help() - return True, "显示帮助信息", True - - else: - await self.send_text(f"❌ 未知的子命令: {subcommand}\n使用 /permission help 查看帮助") - return True, "未知子命令", True - - async def _show_help(self): - """显示帮助信息""" - help_text = """📋 权限管理命令帮助 - -🔐 管理命令(需要管理权限): -• /permission grant <@用户|QQ号> <权限节点> - 授权用户权限 -• /permission revoke <@用户|QQ号> <权限节点> - 撤销用户权限 - -👀 查看命令(需要查看权限): -• /permission list [用户] - 查看用户权限列表 -• /permission check <@用户|QQ号> <权限节点> - 检查用户是否拥有权限 -• /permission nodes [插件名] - 查看权限节点列表 -• /permission allnodes - 查看所有插件的权限节点详情 - -❓ 其他: -• /permission help - 显示此帮助 - -📝 示例: -• /permission grant @张三 plugin.example.command -• /permission list 123456789 -• /permission nodes example_plugin -• /permission allnodes - -🔄 别名:可以使用 /perm 或 /权限 代替 /permission""" - - await self.send_text(help_text) - - @staticmethod - def _parse_user_mention(mention: str) -> str | None: - """解析用户提及,提取QQ号 - - 支持的格式: - - @<用户名:QQ号> 格式 - - [CQ:at,qq=QQ号] 格式 - - 直接的QQ号 - """ - # 匹配 @<用户名:QQ号> 格式,提取QQ号 - at_match = re.search(r"@<[^:]+:(\d+)>", mention) - if at_match: - return at_match.group(1) - - # 直接是数字 - if mention.isdigit(): - return mention - - return None - - @staticmethod - def parse_user_from_args(args: CommandArgs, index: int = 0) -> str | None: - """从CommandArgs中解析用户ID - - Args: - args: 命令参数对象 - index: 参数索引,默认为0(第一个参数) - - Returns: - Optional[str]: 解析出的用户ID,如果解析失败返回None - """ - if index >= args.count(): - return None - - mention = args.get_arg(index) - - # 匹配 @<用户名:QQ号> 格式,提取QQ号 - at_match = re.search(r"@<[^:]+:(\d+)>", mention) - if at_match: - return at_match.group(1) - - # 匹配传统的 [CQ:at,qq=数字] 格式 - cq_match = re.search(r"\[CQ:at,qq=(\d+)\]", mention) - if cq_match: - return cq_match.group(1) - - # 直接是数字 - if mention.isdigit(): - return mention - - return None - - @require_permission("plugin.permission.manage", "❌ 你没有权限管理的权限") - async def _grant_permission(self, chat_stream, args: list[str]): - """授权用户权限""" - if len(args) < 2: - await self.send_text("❌ 用法: /permission grant <@用户|QQ号> <权限节点>") - return - - # 解析用户ID - 使用新的解析方法 - user_id = self._parse_user_mention(args[0]) - if not user_id: - await self.send_text("❌ 无效的用户格式,请使用 @<用户名:QQ号> 或直接输入QQ号") - return - - permission_node = args[1] - - # 执行授权 - success = await permission_api.grant_permission(chat_stream.platform, user_id, permission_node) - - if success: - await self.send_text(f"✅ 已授权用户 {user_id} 权限节点 `{permission_node}`") - else: - await self.send_text("❌ 授权失败,请检查权限节点是否存在") - - @require_permission("plugin.permission.manage", "❌ 你没有权限管理的权限") - async def _revoke_permission(self, chat_stream, args: list[str]): - """撤销用户权限""" - if len(args) < 2: - await self.send_text("❌ 用法: /permission revoke <@用户|QQ号> <权限节点>") - return - - # 解析用户ID - 使用新的解析方法 - user_id = self._parse_user_mention(args[0]) - if not user_id: - await self.send_text("❌ 无效的用户格式,请使用 @<用户名:QQ号> 或直接输入QQ号") - return - - permission_node = args[1] - - # 执行撤销 - success = await permission_api.revoke_permission(chat_stream.platform, user_id, permission_node) - - if success: - await self.send_text(f"✅ 已撤销用户 {user_id} 权限节点 `{permission_node}`") - else: - await self.send_text("❌ 撤销失败,请检查权限节点是否存在") - - @require_permission("plugin.permission.view", "❌ 你没有查看权限的权限") - async def _list_permissions(self, chat_stream, args: list[str]): - """列出用户权限""" - target_user_id = None - - if args: - # 指定了用户 - 使用新的解析方法 - target_user_id = self._parse_user_mention(args[0]) - if not target_user_id: - await self.send_text("❌ 无效的用户格式,请使用 @<用户名:QQ号> 或直接输入QQ号") - return - else: - # 查看自己的权限 - target_user_id = chat_stream.user_info.user_id - - # 检查是否为Master用户 - is_master = await permission_api.is_master(chat_stream.platform, target_user_id) - - # 获取用户权限 - permissions = await permission_api.get_user_permissions(chat_stream.platform, target_user_id) - - if is_master: - response = f"👑 用户 `{target_user_id}` 是Master用户,拥有所有权限" - else: - if permissions: - perm_list = "\n".join([f"• `{perm}`" for perm in permissions]) - response = f"📋 用户 `{target_user_id}` 拥有的权限:\n{perm_list}" - else: - response = f"📋 用户 `{target_user_id}` 没有任何权限" - - await self.send_text(response) - - @require_permission("plugin.permission.view", "❌ 你没有查看权限的权限") - async def _check_permission(self, chat_stream, args: list[str]): - """检查用户权限""" - if len(args) < 2: - await self.send_text("❌ 用法: /permission check <@用户|QQ号> <权限节点>") - return - - # 解析用户ID - 使用新的解析方法 - user_id = self._parse_user_mention(args[0]) - if not user_id: - await self.send_text("❌ 无效的用户格式,请使用 @<用户名:QQ号> 或直接输入QQ号") - return - - permission_node = args[1] - - # 检查权限 - has_permission = await permission_api.check_permission(chat_stream.platform, user_id, permission_node) - is_master = permission_api.is_master(chat_stream.platform, user_id) - - if has_permission: - if is_master: - response = f"✅ 用户 `{user_id}` 拥有权限 `{permission_node}`(Master用户)" - else: - response = f"✅ 用户 `{user_id}` 拥有权限 `{permission_node}`" - else: - response = f"❌ 用户 `{user_id}` 没有权限 `{permission_node}`" - - await self.send_text(response) - - @require_permission("plugin.permission.view", "❌ 你没有查看权限的权限") - async def _list_nodes(self, chat_stream, args: list[str]): - """列出权限节点""" - plugin_name = args[0] if args else None - - if plugin_name: - # 获取指定插件的权限节点 - nodes = await permission_api.get_plugin_permission_nodes(plugin_name) - title = f"📋 插件 {plugin_name} 的权限节点:" - else: - # 获取所有权限节点 - nodes = await permission_api.get_all_permission_nodes() - title = "📋 所有权限节点:" - - if not nodes: - if plugin_name: - response = f"📋 插件 {plugin_name} 没有注册任何权限节点" - else: - response = "📋 系统中没有任何权限节点" - else: - node_list = [] - for node in nodes: - default_text = "(默认授权)" if node["default_granted"] else "(默认拒绝)" - node_list.append(f"• {node['node_name']} {default_text}") - node_list.append(f" 📄 {node['description']}") - if not plugin_name: - node_list.append(f" 🔌 插件: {node['plugin_name']}") - node_list.append("") # 空行分隔 - - response = title + "\n" + "\n".join(node_list) - - await self.send_text(response) - - @require_permission("plugin.permission.view", "❌ 你没有查看权限的权限") - async def _list_all_nodes_with_description(self, chat_stream): - """列出所有插件的权限节点(带详细描述)""" - # 获取所有权限节点 - all_nodes = await permission_api.get_all_permission_nodes() - - if not all_nodes: - response = "📋 系统中没有任何权限节点" - await self.send_text(response) - return - - # 按插件名分组节点 - plugins_dict = {} - for node in all_nodes: - plugin_name = node["plugin_name"] - if plugin_name not in plugins_dict: - plugins_dict[plugin_name] = [] - plugins_dict[plugin_name].append(node) - - # 构建响应消息 - response_parts = ["📋 所有插件权限节点详情:\n"] - - for plugin_name in sorted(plugins_dict.keys()): - nodes = plugins_dict[plugin_name] - response_parts.append(f"🔌 **{plugin_name}** ({len(nodes)}个节点):") - - for node in nodes: - default_text = "✅默认授权" if node["default_granted"] else "❌默认拒绝" - response_parts.append(f" • `{node['node_name']}` - {default_text}") - response_parts.append(f" 📄 {node['description']}") - - response_parts.append("") # 插件间空行分隔 - - # 添加统计信息 - total_nodes = len(all_nodes) - total_plugins = len(plugins_dict) - response_parts.append(f"📊 统计:共 {total_plugins} 个插件,{total_nodes} 个权限节点") - - response = "\n".join(response_parts) - - # 如果消息太长,分段发送 - if len(response) > 4000: # 预留一些空间避免超出限制 - await self._send_long_message(response) - else: - await self.send_text(response) - - async def _send_long_message(self, message: str): - """发送长消息,自动分段""" - lines = message.split("\n") - current_chunk = [] - current_length = 0 - - for line in lines: - line_length = len(line) + 1 # +1 for newline - - # 如果添加这一行会超出限制,先发送当前块 - if current_length + line_length > 3500 and current_chunk: - await self.send_text("\n".join(current_chunk)) - current_chunk = [] - current_length = 0 - - current_chunk.append(line) - current_length += line_length - - # 发送最后一块 - if current_chunk: - await self.send_text("\n".join(current_chunk)) - - -@register_plugin -class PermissionManagerPlugin(BasePlugin): - plugin_name: str = "permission_manager_plugin" - enable_plugin: bool = False - dependencies: ClassVar[list[str]] = [] - python_dependencies: ClassVar[list[str]] = [] - config_file_name: str = "config.toml" - config_schema: ClassVar[dict] = { - "plugin": { - "enabled": ConfigField(bool, default=True, description="是否启用插件"), - "config_version": ConfigField(type=str, default="1.1.0", description="配置文件版本"), - } - } - - def get_plugin_components(self) -> list[tuple[PlusCommandInfo, type[PlusCommand]]]: - """返回插件的PlusCommand组件""" - return [(PermissionCommand.get_plus_command_info(), PermissionCommand)] - - permission_nodes: ClassVar[list[PermissionNodeField]] = [ - PermissionNodeField( - node_name="manage", - description="权限管理:可以授权和撤销其他用户的权限", - ), - PermissionNodeField( - node_name="view", - description="权限查看:可以查看权限节点和用户权限信息", - ), - ] \ No newline at end of file diff --git a/src/plugins/built_in/plugin_management/__init__.py b/src/plugins/built_in/plugin_management/__init__.py deleted file mode 100644 index 6010720ac..000000000 --- a/src/plugins/built_in/plugin_management/__init__.py +++ /dev/null @@ -1,17 +0,0 @@ -from src.plugin_system.base.plugin_metadata import PluginMetadata - -__plugin_meta__ = PluginMetadata( - name="插件和组件管理 (Plugin and Component Management)", - description="通过系统API管理插件和组件的生命周期,包括加载、卸载、启用和禁用等操作。", - usage="该插件提供 `plugin_management` command。", - version="1.0.0", - author="MoFox-Bot团队", - license="GPL-v3.0-or-later", - repository_url="https://github.com/MaiM-with-u/maibot", - keywords=["plugins", "components", "management", "built-in"], - categories=["Core System", "Plugin Management"], - extra={ - "is_built_in": True, - "plugin_type": "plugin_management", - }, -) diff --git a/src/plugins/built_in/plugin_management/plugin.py b/src/plugins/built_in/plugin_management/plugin.py deleted file mode 100644 index e9fbaaea1..000000000 --- a/src/plugins/built_in/plugin_management/plugin.py +++ /dev/null @@ -1,518 +0,0 @@ -import asyncio -from typing import ClassVar - -from src.plugin_system import ( - BasePlugin, - ComponentInfo, - ComponentType, - ConfigField, - component_manage_api, - plugin_manage_api, - register_plugin, -) -from src.plugin_system.apis.permission_api import permission_api -from src.plugin_system.base.command_args import CommandArgs -from src.plugin_system.base.component_types import ChatType, PlusCommandInfo -from src.plugin_system.base.plus_command import PlusCommand -from src.plugin_system.utils.permission_decorators import require_permission - - -class ManagementCommand(PlusCommand): - """插件管理命令 - 使用PlusCommand系统""" - - command_name = "pm" - command_description = "插件管理命令,支持插件和组件的管理操作" - command_aliases: ClassVar[list[str]] = ["pluginmanage", "插件管理"] - priority = 10 - chat_type_allow = ChatType.ALL - intercept_message = True - - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - - @require_permission("plugin.management.admin", "❌ 你没有插件管理的权限") - async def execute(self, args: CommandArgs) -> tuple[bool, str, bool]: - """执行插件管理命令""" - if args.is_empty: - await self._show_help("all") - return True, "显示帮助信息", True - - subcommand = args.get_first.lower() - remaining_args = args.get_args()[1:] # 获取除第一个参数外的所有参数 - - if subcommand in ["plugin", "插件"]: - return await self._handle_plugin_commands(remaining_args) - elif subcommand in ["component", "组件", "comp"]: - return await self._handle_component_commands(remaining_args) - elif subcommand in ["help", "帮助"]: - await self._show_help("all") - return True, "显示帮助信息", True - else: - await self.send_text(f"❌ 未知的子命令: {subcommand}\n使用 /pm help 查看帮助") - return True, "未知子命令", True - - async def _handle_plugin_commands(self, args: list[str]) -> tuple[bool, str, bool]: - """处理插件相关命令""" - if not args: - await self._show_help("plugin") - return True, "显示插件帮助", True - - action = args[0].lower() - - if action in ["help", "帮助"]: - await self._show_help("plugin") - elif action in ["list", "列表"]: - await self._list_registered_plugins() - elif action in ["list_enabled", "已启用"]: - await self._list_loaded_plugins() - elif action in ["rescan", "重扫"]: - await self._rescan_plugin_dirs() - elif action in ["load", "加载"] and len(args) > 1: - await self._load_plugin(args[1]) - elif action in ["unload", "卸载"] and len(args) > 1: - await self._unload_plugin(args[1]) - elif action in ["reload", "重载"] and len(args) > 1: - await self._reload_plugin(args[1]) - elif action in ["force_reload", "强制重载"] and len(args) > 1: - await self._force_reload_plugin(args[1]) - elif action in ["add_dir", "添加目录"] and len(args) > 1: - await self._add_dir(args[1]) - else: - await self.send_text("❌ 插件管理命令不合法\n使用 /pm plugin help 查看帮助") - return False, "命令不合法", True - - return True, "插件命令执行完成", True - - async def _handle_component_commands(self, args: list[str]) -> tuple[bool, str, bool]: - """处理组件相关命令""" - if not args: - await self._show_help("component") - return True, "显示组件帮助", True - - action = args[0].lower() - - if action in ["help", "帮助"]: - await self._show_help("component") - elif action in ["list", "列表"]: - if len(args) == 1: - await self._list_all_registered_components() - elif len(args) == 2: - if args[1] in ["enabled", "启用"]: - await self._list_enabled_components() - elif args[1] in ["disabled", "禁用"]: - await self._list_disabled_components() - else: - await self.send_text("❌ 组件列表命令不合法") - return False, "命令不合法", True - elif len(args) == 3: - if args[1] in ["enabled", "启用"]: - await self._list_enabled_components(target_type=args[2]) - elif args[1] in ["disabled", "禁用"]: - await self._list_disabled_components(target_type=args[2]) - elif args[1] in ["type", "类型"]: - await self._list_registered_components_by_type(args[2]) - else: - await self.send_text("❌ 组件列表命令不合法") - return False, "命令不合法", True - elif action in ["enable", "启用"] and len(args) >= 4: - scope = args[1].lower() - component_name = args[2] - component_type = args[3] - if scope in ["global", "全局"]: - await self._globally_enable_component(component_name, component_type) - elif scope in ["local", "本地"]: - await self._locally_enable_component(component_name, component_type) - else: - await self.send_text("❌ 组件启用命令不合法,范围应为 global 或 local") - return False, "命令不合法", True - elif action in ["disable", "禁用"] and len(args) >= 4: - scope = args[1].lower() - component_name = args[2] - component_type = args[3] - if scope in ["global", "全局"]: - await self._globally_disable_component(component_name, component_type) - elif scope in ["local", "本地"]: - await self._locally_disable_component(component_name, component_type) - else: - await self.send_text("❌ 组件禁用命令不合法,范围应为 global 或 local") - return False, "命令不合法", True - else: - await self.send_text("❌ 组件管理命令不合法\n使用 /pm component help 查看帮助") - return False, "命令不合法", True - - return True, "组件命令执行完成", True - - async def _show_help(self, target: str): - """显示帮助信息""" - help_msg = "" - if target == "all": - help_msg = """📋 插件管理命令帮助 - -🔧 主要功能: -• `/pm help` - 显示此帮助 -• `/pm plugin` - 插件管理命令 -• `/pm component` - 组件管理命令 - -📝 使用示例: -• `/pm plugin help` - 查看插件管理帮助 -• `/pm component help` - 查看组件管理帮助 - -🔄 别名:可以使用 `/pluginmanage` 或 `/插件管理` 代替 `/pm`""" - elif target == "plugin": - help_msg = """🔌 插件管理命令帮助 - -📋 基本操作: -• `/pm plugin help` - 显示插件管理帮助 -• `/pm plugin list` - 列出所有注册的插件 -• `/pm plugin list_enabled` - 列出所有加载(启用)的插件 -• `/pm plugin rescan` - 重新扫描所有插件目录 - -⚙️ 插件控制: -• `/pm plugin load <插件名>` - 加载指定插件 -• `/pm plugin unload <插件名>` - 卸载指定插件 -• `/pm plugin reload <插件名>` - 重新加载指定插件 -• `/pm plugin force_reload <插件名>` - 强制重载指定插件(深度清理) -• `/pm plugin add_dir <目录路径>` - 添加插件目录 - -�📝 示例: -• `/pm plugin load echo_example` -• `/pm plugin force_reload permission_manager_plugin`""" - elif target == "component": - help_msg = """🧩 组件管理命令帮助 - -📋 基本查看: -• `/pm component help` - 显示组件管理帮助 -• `/pm component list` - 列出所有注册的组件 -• `/pm component list enabled [类型]` - 列出启用的组件 -• `/pm component list disabled [类型]` - 列出禁用的组件 -• `/pm component list type <组件类型>` - 列出指定类型的组件 - -⚙️ 组件控制: -• `/pm component enable global <组件名> <类型>` - 全局启用组件 -• `/pm component enable local <组件名> <类型>` - 本聊天启用组件 -• `/pm component disable global <组件名> <类型>` - 全局禁用组件 -• `/pm component disable local <组件名> <类型>` - 本聊天禁用组件 - -📝 组件类型: -• `action` - 动作组件 -• `command` - 命令组件 -• `event_handler` - 事件处理组件 -• `plus_command` - 增强命令组件 - -💡 示例: -• `/pm component list type plus_command` -• `/pm component enable global echo_command command`""" - - await self.send_text(help_msg) - - async def _list_loaded_plugins(self): - """列出已加载的插件""" - plugins = plugin_manage_api.list_loaded_plugins() - await self.send_text(f"📦 已加载的插件: {', '.join(plugins) if plugins else '无'}") - - async def _list_registered_plugins(self): - """列出已注册的插件""" - plugins = plugin_manage_api.list_registered_plugins() - await self.send_text(f"📋 已注册的插件: {', '.join(plugins) if plugins else '无'}") - - async def _rescan_plugin_dirs(self): - """重新扫描插件目录""" - plugin_manage_api.rescan_plugin_directory() - await self.send_text("🔄 插件目录重新扫描已启动") - - async def _load_plugin(self, plugin_name: str): - """加载指定插件""" - success, count = plugin_manage_api.load_plugin(plugin_name) - if success: - await self.send_text(f"✅ 插件加载成功: `{plugin_name}`") - else: - if count == 0: - await self.send_text(f"⚠️ 插件 `{plugin_name}` 为禁用状态") - else: - await self.send_text(f"❌ 插件加载失败: `{plugin_name}`") - - async def _unload_plugin(self, plugin_name: str): - """卸载指定插件""" - 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): - """重新加载指定插件""" - success = await plugin_manage_api.reload_plugin(plugin_name) - if success: - await self.send_text(f"✅ 插件重新加载成功: `{plugin_name}`") - else: - await self.send_text(f"❌ 插件重新加载失败: `{plugin_name}`") - - async def _force_reload_plugin(self, plugin_name: str): - """强制重载指定插件(深度清理)""" - await self.send_text(f"🔄 开始强制重载插件: `{plugin_name}`...") - - try: - success = plugin_manage_api.force_reload_plugin(plugin_name) - if success: - await self.send_text(f"✅ 插件强制重载成功: `{plugin_name}`") - else: - await self.send_text(f"❌ 插件强制重载失败: `{plugin_name}`") - except Exception as e: - await self.send_text(f"❌ 强制重载过程中发生错误: {e!s}") - - 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: - await self.send_text(f"❌ 插件目录添加失败: `{dir_path}`") - - @staticmethod - def _fetch_all_registered_components() -> list[ComponentInfo]: - all_plugin_info = component_manage_api.get_all_plugin_info() - if not all_plugin_info: - - return [] - - components_info: list[ComponentInfo] = [] - for plugin_info in all_plugin_info.values(): - components_info.extend(plugin_info.components) - return components_info - - def _fetch_locally_disabled_components(self) -> list[str]: - """获取本地禁用的组件列表""" - stream_id = self.message.chat_stream.stream_id - locally_disabled_components_actions = component_manage_api.get_locally_disabled_components( - stream_id, ComponentType.ACTION - ) - locally_disabled_components_commands = component_manage_api.get_locally_disabled_components( - stream_id, ComponentType.COMMAND - ) - locally_disabled_components_plus_commands = component_manage_api.get_locally_disabled_components( - stream_id, ComponentType.PLUS_COMMAND - ) - locally_disabled_components_event_handlers = component_manage_api.get_locally_disabled_components( - stream_id, ComponentType.EVENT_HANDLER - ) - return ( - locally_disabled_components_actions - + locally_disabled_components_commands - + locally_disabled_components_plus_commands - + locally_disabled_components_event_handlers - ) - - async def _list_all_registered_components(self): - """列出所有已注册的组件""" - components_info = self._fetch_all_registered_components() - if not components_info: - await self.send_text("📋 没有注册的组件") - return - - all_components_str = ", ".join( - f"`{component.name}` ({component.component_type})" for component in components_info - ) - await self.send_text(f"📋 已注册的组件:\n{all_components_str}") - - async def _list_enabled_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": - enabled_components = [component for component in components_info if 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"✅ 满足条件的已启用全局组件:\n{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"✅ 本聊天满足条件的已启用组件:\n{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"❌ 满足条件的已禁用全局组件:\n{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"❌ 本聊天满足条件的已禁用组件:\n{disabled_components_str}") - - async def _list_registered_components_by_type(self, target_type: str): - """按类型列出已注册的组件""" - type_mapping = { - "action": ComponentType.ACTION, - "command": ComponentType.COMMAND, - "event_handler": ComponentType.EVENT_HANDLER, - "plus_command": ComponentType.PLUS_COMMAND, - } - - component_type = type_mapping.get(target_type.lower()) - if not component_type: - await self.send_text( - f"❌ 未知组件类型: `{target_type}`\n支持的类型: action, command, event_handler, plus_command" - ) - 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}` 组件:\n{components_str}") - - async def _globally_enable_component(self, component_name: str, component_type: str): - """全局启用组件""" - type_mapping = { - "action": ComponentType.ACTION, - "command": ComponentType.COMMAND, - "event_handler": ComponentType.EVENT_HANDLER, - "plus_command": ComponentType.PLUS_COMMAND, - } - - target_component_type = type_mapping.get(component_type.lower()) - if not target_component_type: - 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): - """全局禁用组件""" - type_mapping = { - "action": ComponentType.ACTION, - "command": ComponentType.COMMAND, - "event_handler": ComponentType.EVENT_HANDLER, - "plus_command": ComponentType.PLUS_COMMAND, - } - - target_component_type = type_mapping.get(component_type.lower()) - if not target_component_type: - 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): - """本地启用组件""" - type_mapping = { - "action": ComponentType.ACTION, - "command": ComponentType.COMMAND, - "event_handler": ComponentType.EVENT_HANDLER, - "plus_command": ComponentType.PLUS_COMMAND, - } - - target_component_type = type_mapping.get(component_type.lower()) - if not target_component_type: - await self.send_text(f"❌ 未知组件类型: `{component_type}`") - return - - stream_id = self.message.chat_stream.stream_id - if component_manage_api.locally_enable_component(component_name, target_component_type, 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): - """本地禁用组件""" - type_mapping = { - "action": ComponentType.ACTION, - "command": ComponentType.COMMAND, - "event_handler": ComponentType.EVENT_HANDLER, - "plus_command": ComponentType.PLUS_COMMAND, - } - - target_component_type = type_mapping.get(component_type.lower()) - if not target_component_type: - await self.send_text(f"❌ 未知组件类型: `{component_type}`") - return - - stream_id = self.message.chat_stream.stream_id - if component_manage_api.locally_disable_component(component_name, target_component_type, stream_id): - await self.send_text(f"✅ 本地禁用组件成功: `{component_name}`") - else: - await self.send_text(f"❌ 本地禁用组件失败: `{component_name}`") - - -@register_plugin -class PluginManagementPlugin(BasePlugin): - plugin_name: str = "plugin_management_plugin" - enable_plugin: bool = False - dependencies: ClassVar[list[str]] = [] - python_dependencies: ClassVar[list[str]] = [] - config_file_name: str = "config.toml" - config_schema: ClassVar[dict] = { - "plugin": { - "enabled": ConfigField(bool, default=False, description="是否启用插件"), - "config_version": ConfigField(type=str, default="1.1.0", description="配置文件版本"), - }, - } - - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - # 注册权限节点 - - async def on_plugin_loaded(self): - await permission_api.register_permission_node( - "plugin.management.admin", - "插件管理:可以管理插件和组件的加载、卸载、启用、禁用等操作", - "plugin_management", - False, - ) - - def get_plugin_components(self) -> list[tuple[PlusCommandInfo, type[PlusCommand]]]: - """返回插件的PlusCommand组件""" - components = [] - if self.get_config("plugin.enabled", True): - components.append((ManagementCommand.get_plus_command_info(), ManagementCommand)) - return components diff --git a/src/plugins/built_in/system_management/plugin.py b/src/plugins/built_in/system_management/plugin.py index 1431921f4..57645f10d 100644 --- a/src/plugins/built_in/system_management/plugin.py +++ b/src/plugins/built_in/system_management/plugin.py @@ -468,7 +468,7 @@ class SystemCommand(PlusCommand): total_plugins = len(plugins_dict) response_parts.append(f"📊 统计:共 {total_plugins} 个插件,{total_nodes} 个权限节点") response = "\n".join(response_parts) - + if len(response) > 4000: await self._send_long_message(response) else: