From fc05146f247db94c93e70fb084ae8971b475927f Mon Sep 17 00:00:00 2001 From: minecraft1024a Date: Sat, 29 Nov 2025 20:06:42 +0800 Subject: [PATCH 1/3] =?UTF-8?q?feat(plugin=5Fsystem):=20=E5=AE=9E=E7=8E=B0?= =?UTF-8?q?=E6=8F=92=E4=BB=B6=E7=BB=84=E4=BB=B6=E7=9A=84=E6=9F=A5=E8=AF=A2?= =?UTF-8?q?=E4=B8=8E=E5=90=AF=E7=A6=81=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为 `/system plugin` 命令引入了完整的组件级管理功能，以提供比插件级更精细的控制粒度。 哼，总是只能开关整个插件也太粗糙了。这次更新后，主人你就可以更灵活地控制机器人的行为了，不用再因为一个小功能不想要就禁用整个插件啦。 新增功能包括： - **组件查询**: 通过 `info`, `list`, `search`, `disabled` 子命令，可以查看插件详情、列出/搜索组件以及查看全局禁用列表。 - **状态控制**: 通过 `enable` 和 `disable` 子命令，可以全局启用或禁用指定的组件。 - **帮助文本**: 更新了相关的帮助信息，让你不会迷路。 - **安全保护**: 核心类型的组件（如 Router, Prompt）被保护，防止被意外禁用，我可不想你把系统搞坏了。 --- .../built_in/system_management/plugin.py | 216 +++++++++++++++++- 1 file changed, 208 insertions(+), 8 deletions(-) diff --git a/src/plugins/built_in/system_management/plugin.py b/src/plugins/built_in/system_management/plugin.py index 247c459b7..eba45d14c 100644 --- a/src/plugins/built_in/system_management/plugin.py +++ b/src/plugins/built_in/system_management/plugin.py @@ -81,7 +81,7 @@ class SystemCommand(PlusCommand): 🔧 主要功能： • `/system help` - 显示此帮助 • `/system permission` - 权限管理 -• `/system plugin` - 插件管理 +• `/system plugin` - 插件与组件管理 • `/system schedule` - 定时任务管理 • `/system prompt` - 提示词注入管理 """ @@ -98,16 +98,18 @@ class SystemCommand(PlusCommand): """ elif target == "plugin": help_text = """🔌 插件管理命令帮助 -📋 基本操作： -• `/system plugin help` - 显示插件管理帮助 -• `/system plugin report` - 查看系统插件报告 -• `/system plugin rescan` - 重新扫描所有插件目录 ⚙️ 插件控制： +• `/system plugin rescan` - 重新扫描所有插件目录 • `/system plugin load <插件名>` - 加载指定插件 • `/system plugin reload <插件名>` - 重新加载指定插件 • `/system plugin reload_all` - 重新加载所有插件 -🎯 局部控制 (需要 `system.plugin.manage.local` 权限): + +🔧 全局组件控制 (需要 `system.plugin.manage` 权限): +• `/system plugin enable <组件名>` - 全局启用组件 +• `/system plugin disable <组件名>` - 全局禁用组件 + +🎯 局部组件控制 (需要 `system.plugin.manage.local` 权限): • `/system plugin enable_local <名称> [group <群号> | private ]` - 在指定会话局部启用组件 • `/system plugin disable_local <名称> [group <群号> | private ]` - 在指定会话局部禁用组件 """ @@ -156,6 +158,15 @@ class SystemCommand(PlusCommand): await self._show_help("plugin") elif action in ["report", "报告"]: await self._show_system_report() + elif action in ["info", "详情"] and remaining_args: + await self._show_plugin_info(remaining_args[0]) + elif action in ["list", "列表"]: + comp_type = remaining_args[0] if remaining_args else None + await self._list_components(comp_type) + elif action in ["search", "搜索"] and remaining_args: + await self._search_components(remaining_args[0]) + elif action in ["disabled", "禁用列表"]: + await self._list_disabled_components() elif action in ["rescan", "重扫"]: await self._rescan_plugin_dirs() elif action in ["load", "加载"] and len(remaining_args) > 0: @@ -164,6 +175,10 @@ class SystemCommand(PlusCommand): await self._reload_plugin(remaining_args[0]) elif action in ["reload_all", "重载全部"]: await self._reload_all_plugins() + elif action in ["enable", "启用"] and len(remaining_args) >= 1: + await self._set_global_component_state(remaining_args[0], enabled=True) + elif action in ["disable", "禁用"] and len(remaining_args) >= 1: + await self._set_global_component_state(remaining_args[0], enabled=False) elif action in ["enable_local", "局部启用"] and len(remaining_args) >= 1: await self._set_local_component_state(remaining_args, enabled=True) elif action in ["disable_local", "局部禁用"] and len(remaining_args) >= 1: @@ -430,8 +445,193 @@ class SystemCommand(PlusCommand): await self.send_text("\n".join(response_parts)) # ================================================================= - # Permission Management Section + # Permission Management Section (Plugin Stats & Info) # ================================================================= + @require_permission("plugin.manage", deny_message="❌ 你没有权限查看插件详情") + async def _show_plugin_info(self, plugin_name: str): + """显示单个插件的详细信息""" + details = plugin_info_api.get_plugin_details(plugin_name) + + if not details: + # 尝试模糊匹配 + all_plugins = plugin_info_api.list_plugins("loaded") + suggestions = [p for p in all_plugins if plugin_name.lower() in p.lower()] + if suggestions: + await self.send_text( + f"❌ 未找到插件 `{plugin_name}`\n" + f"你可能想找的是: {', '.join([f'`{s}`' for s in suggestions[:5]])}" + ) + else: + await self.send_text(f"❌ 未找到插件 `{plugin_name}`") + return + + response_parts = [ + f"🔌 **插件详情: {details['display_name']}**", + f" • 内部名称: `{details['name']}`", + f" • 版本: {details['version']}", + f" • 作者: {details['author']}", + f" • 状态: {'✅ 已启用' if details['enabled'] else '❌ 已禁用'}", + f" • 加载状态: {details['status']}", + ] + + if details.get('description'): + response_parts.append(f" • 描述: {details['description']}") + + if details.get('license'): + response_parts.append(f" • 许可证: {details['license']}") + + # 组件信息 + if details['components']: + response_parts.append(f"\n🧩 **组件列表** (共 {len(details['components'])} 个):") + for comp in details['components']: + status = "✅" if comp['enabled'] else "❌" + response_parts.append(f" {status} `{comp['name']}` ({comp['component_type']})") + if comp.get('description'): + response_parts.append(f" {comp['description'][:50]}...") + + await self._send_long_message("\n".join(response_parts)) + + @require_permission("plugin.manage", deny_message="❌ 你没有权限查看组件列表") + async def _list_components(self, comp_type_str: str | None): + """列出指定类型的组件""" + # 显示可用类型帮助 + available_types = [t.value for t in ComponentType] + + if comp_type_str: + # 尝试匹配组件类型 + comp_type = None + for t in ComponentType: + if t.value.lower() == comp_type_str.lower() or t.name.lower() == comp_type_str.lower(): + comp_type = t + break + + if not comp_type: + await self.send_text( + f"❌ 未知的组件类型: `{comp_type_str}`\n" + f"可用类型: {', '.join([f'`{t}`' for t in available_types])}" + ) + return + + components = plugin_info_api.list_components(comp_type, enabled_only=False) + title = f"🧩 **{comp_type.value} 组件列表** (共 {len(components)} 个)" + else: + # 列出所有类型的统计 + response_parts = ["🧩 **组件类型概览**", ""] + for t in ComponentType: + comps = plugin_info_api.list_components(t, enabled_only=False) + enabled = sum(1 for c in comps if c['enabled']) + if comps: + response_parts.append(f"• **{t.value}**: {enabled}/{len(comps)} 启用") + + response_parts.append(f"\n💡 使用 `/system plugin list <类型>` 查看详情") + response_parts.append(f"可用类型: {', '.join([f'`{t}`' for t in available_types])}") + await self.send_text("\n".join(response_parts)) + return + + if not components: + await self.send_text(f"📭 没有找到 {comp_type.value} 类型的组件") + return + + response_parts = [title, ""] + for comp in components: + status = "✅" if comp['enabled'] else "❌" + response_parts.append(f"{status} `{comp['name']}` (来自: `{comp['plugin_name']}`)") + + await self._send_long_message("\n".join(response_parts)) + + @require_permission("plugin.manage", deny_message="❌ 你没有权限搜索组件") + async def _search_components(self, keyword: str): + """搜索组件""" + results = plugin_info_api.search_components_by_name(keyword, case_sensitive=False) + + if not results: + await self.send_text(f"🔍 未找到包含 `{keyword}` 的组件") + return + + response_parts = [f"🔍 **搜索结果** (关键词: `{keyword}`, 共 {len(results)} 个)", ""] + + for comp in results: + status = "✅" if comp['enabled'] else "❌" + response_parts.append( + f"{status} `{comp['name']}` ({comp['component_type']})\n" + f" 来自: `{comp['plugin_name']}`" + ) + + await self._send_long_message("\n".join(response_parts)) + + @require_permission("plugin.manage", deny_message="❌ 你没有权限查看禁用组件") + async def _list_disabled_components(self): + """列出所有禁用的组件""" + disabled = component_state_api.get_disabled_components() + + if not disabled: + await self.send_text("✅ 当前没有被禁用的组件") + return + + response_parts = [f"🚫 **禁用组件列表** (共 {len(disabled)} 个)", ""] + + # 按插件分组 + by_plugin: dict[str, list] = {} + for comp in disabled: + plugin_name = comp.plugin_name + if plugin_name not in by_plugin: + by_plugin[plugin_name] = [] + by_plugin[plugin_name].append(comp) + + for plugin_name, comps in by_plugin.items(): + response_parts.append(f"🔌 **{plugin_name}**:") + for comp in comps: + response_parts.append(f" ❌ `{comp.name}` ({comp.component_type.value})") + + await self._send_long_message("\n".join(response_parts)) + + @require_permission("plugin.manage", deny_message="❌ 你没有权限管理组件状态") + async def _set_global_component_state(self, comp_name: str, enabled: bool): + """全局启用或禁用组件""" + # 搜索组件 + found_components = plugin_info_api.search_components_by_name(comp_name, exact_match=True) + + if not found_components: + # 尝试模糊搜索给出建议 + fuzzy_results = plugin_info_api.search_components_by_name(comp_name, exact_match=False) + if fuzzy_results: + suggestions = ", ".join([f"`{c['name']}`" for c in fuzzy_results[:5]]) + await self.send_text(f"❌ 未找到名为 `{comp_name}` 的组件\n你可能想找的是: {suggestions}") + else: + await self.send_text(f"❌ 未找到名为 `{comp_name}` 的组件") + return + + if len(found_components) > 1: + suggestions = "\n".join([f"- `{c['name']}` (类型: {c['component_type']})" for c in found_components]) + await self.send_text(f"❌ 发现多个名为 `{comp_name}` 的组件，操作已取消。\n找到的组件:\n{suggestions}") + return + + component_info = found_components[0] + comp_type_str = component_info["component_type"] + component_type = ComponentType(comp_type_str) + + # 禁用保护 + if not enabled: + protected_types = [ + ComponentType.INTEREST_CALCULATOR, + ComponentType.PROMPT, + ComponentType.ROUTER, + ] + if component_type in protected_types: + await self.send_text(f"❌ 无法禁用核心组件 `{comp_name}` ({comp_type_str})") + return + + # 执行操作 + success = await component_state_api.set_component_enabled(comp_name, component_type, enabled) + + action_text = "启用" if enabled else "禁用" + if success: + await self.send_text(f"✅ 已全局{action_text}组件 `{comp_name}` ({comp_type_str})") + else: + if component_type == ComponentType.CHATTER and not enabled: + await self.send_text(f"❌ 无法禁用最后一个 Chatter 组件 `{comp_name}`") + else: + await self.send_text(f"❌ {action_text}组件 `{comp_name}` 失败，请检查日志") @require_permission("plugin.manage", deny_message="❌ 你没有权限查看插件报告") async def _show_system_report(self): @@ -790,7 +990,7 @@ class SystemManagementPlugin(BasePlugin): config_file_name: str = "config.toml" # 配置文件名 config_schema: ClassVar[dict] = { "plugin": { - "enabled": ConfigField(bool, default=True, description="是否启用插件"), + "enabled": ConfigField(bool, default=False, description="是否启用插件"), } } From feb4e7155e8d3357a74b5d577dd5bd65204819fc Mon Sep 17 00:00:00 2001 From: minecraft1024a Date: Sat, 29 Nov 2025 20:07:40 +0800 Subject: [PATCH 2/3] =?UTF-8?q?chore(system=5Fmanagement):=20=E9=BB=98?= =?UTF-8?q?=E8=AE=A4=E7=A6=81=E7=94=A8=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 --- src/plugins/built_in/system_management/plugin.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/plugins/built_in/system_management/plugin.py b/src/plugins/built_in/system_management/plugin.py index eba45d14c..411caa34b 100644 --- a/src/plugins/built_in/system_management/plugin.py +++ b/src/plugins/built_in/system_management/plugin.py @@ -984,7 +984,7 @@ class SystemCommand(PlusCommand): @register_plugin class SystemManagementPlugin(BasePlugin): plugin_name: str = "system_management" - enable_plugin: bool = True + enable_plugin: bool = False dependencies: ClassVar[list[str]] = [] # 插件依赖列表 python_dependencies: ClassVar[list[str]] = [] # Python包依赖列表，现在使用内置API config_file_name: str = "config.toml" # 配置文件名 From ad0f495911f636b63340da3e9ea2a78464d5ec48 Mon Sep 17 00:00:00 2001 From: minecraft1024a Date: Sat, 29 Nov 2025 20:20:59 +0800 Subject: [PATCH 3/3] =?UTF-8?q?feat(permission=5Fapi):=20=E5=A2=9E?= =?UTF-8?q?=E5=BC=BA=E6=9D=83=E9=99=90API=E6=96=87=E6=A1=A3=EF=BC=8C?= =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E8=AF=A6=E7=BB=86=E6=B3=A8=E9=87=8A=E5=92=8C?= =?UTF-8?q?=E7=A4=BA=E4=BE=8B=20fix(system=5Fmanagement):=20=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D=E7=B3=BB=E7=BB=9F=E5=91=BD=E4=BB=A4=E6=89=A7=E8=A1=8C?= =?UTF-8?q?=E6=97=B6=E7=BC=BA=E5=A4=B1=E7=94=A8=E6=88=B7=E4=BF=A1=E6=81=AF?= =?UTF-8?q?=E7=9A=84=E9=94=99=E8=AF=AF=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/plugin_system/apis/permission_api.py | 362 ++++++++++++++++-- src/plugin_system/base/plus_command.py | 2 +- .../built_in/system_management/plugin.py | 9 +- 3 files changed, 347 insertions(+), 26 deletions(-) diff --git a/src/plugin_system/apis/permission_api.py b/src/plugin_system/apis/permission_api.py index 37394267c..82f6ac785 100644 --- a/src/plugin_system/apis/permission_api.py +++ b/src/plugin_system/apis/permission_api.py @@ -1,80 +1,291 @@ -"""纯异步权限API定义。所有外部调用方必须使用 await。""" +""" +纯异步权限API定义。 -from abc import ABC, abstractmethod -from dataclasses import dataclass -from enum import Enum -from typing import Any +这个模块提供了一套完整的权限管理系统，用于控制用户对各种功能的访问权限。 +所有外部调用方必须使用 await 关键字，因为这是异步API。 + +主要组件： +- PermissionLevel: 权限等级枚举 +- PermissionNode: 权限节点数据类，描述单个权限项 +- UserInfo: 用户信息数据类，标识用户身份 +- IPermissionManager: 权限管理器抽象接口 +- PermissionAPI: 对外暴露的权限API封装类 +""" + +from abc import ABC, abstractmethod # ABC: 抽象基类，abstractmethod: 抽象方法装饰器 +from dataclasses import dataclass # dataclass: 自动生成 __init__, __repr__ 等方法的装饰器 +from enum import Enum # Enum: 枚举类型基类 +from typing import Any # Any: 表示任意类型 from src.common.logger import get_logger -logger = get_logger(__name__) +logger = get_logger(__name__) # 获取当前模块的日志记录器 class PermissionLevel(Enum): - MASTER = "master" + """ + 权限等级枚举类。 + + 定义了系统中的权限等级，目前只有 MASTER（管理员/主人）级别。 + MASTER 用户拥有最高权限，可以执行所有操作。 + """ + MASTER = "master" # 管理员/主人权限 @dataclass class PermissionNode: - node_name: str - description: str - plugin_name: str - default_granted: bool = False + """ + 权限节点数据类。 + + 每个权限节点代表一个具体的权限项，例如"发送消息"、"管理用户"等。 + + 属性: + node_name: 权限节点名称，例如 "plugin.chat.send_message" + description: 权限描述，用于向用户展示这个权限的用途 + plugin_name: 注册这个权限的插件名称 + default_granted: 是否默认授予所有用户，False 表示需要显式授权 + """ + node_name: str # 权限节点唯一标识名 + description: str # 权限的人类可读描述 + plugin_name: str # 所属插件名称 + default_granted: bool = False # 默认是否授予（默认不授予） @dataclass class UserInfo: - platform: str - user_id: str + """ + 用户信息数据类。 + + 用于唯一标识一个用户，通过平台+用户ID的组合确定用户身份。 + + 属性: + platform: 用户所在平台，例如 "qq", "telegram", "discord" + user_id: 用户在该平台上的唯一标识ID + """ + platform: str # 平台标识，如 "qq", "telegram" + user_id: str # 用户ID def __post_init__(self): + """ + dataclass 的后初始化钩子。 + + 确保 user_id 始终是字符串类型，即使传入的是数字也会被转换。 + 这样可以避免类型不一致导致的比较问题。 + """ self.user_id = str(self.user_id) class IPermissionManager(ABC): + """ + 权限管理器抽象接口（Interface）。 + + 这是一个抽象基类，定义了权限管理器必须实现的所有方法。 + 具体的权限管理实现类需要继承此接口并实现所有抽象方法。 + + 使用抽象接口的好处： + 1. 解耦：PermissionAPI 不需要知道具体的实现细节 + 2. 可测试：可以轻松创建 Mock 实现用于测试 + 3. 可替换：可以随时更换不同的权限管理实现 + """ + @abstractmethod - async def check_permission(self, user: UserInfo, permission_node: str) -> bool: ... + async def check_permission(self, user: UserInfo, permission_node: str) -> bool: + """ + 检查用户是否拥有指定权限。 + + Args: + user: 要检查的用户信息 + permission_node: 权限节点名称 + + Returns: + bool: True 表示用户拥有该权限，False 表示没有 + """ + ... @abstractmethod - async def is_master(self, user: UserInfo) -> bool: ... # 同步快速判断 + async def is_master(self, user: UserInfo) -> bool: + """ + 检查用户是否是管理员/主人。 + + 管理员拥有最高权限，通常绕过所有权限检查。 + + Args: + user: 要检查的用户信息 + + Returns: + bool: True 表示是管理员，False 表示不是 + """ + ... @abstractmethod - async def register_permission_node(self, node: PermissionNode) -> bool: ... + async def register_permission_node(self, node: PermissionNode) -> bool: + """ + 注册一个新的权限节点。 + + 插件在加载时会调用此方法注册自己需要的权限。 + + Args: + node: 要注册的权限节点信息 + + Returns: + bool: True 表示注册成功，False 表示失败（可能是重复注册） + """ + ... @abstractmethod - async def grant_permission(self, user: UserInfo, permission_node: str) -> bool: ... + async def grant_permission(self, user: UserInfo, permission_node: str) -> bool: + """ + 授予用户指定权限。 + + Args: + user: 目标用户信息 + permission_node: 要授予的权限节点名称 + + Returns: + bool: True 表示授权成功，False 表示失败 + """ + ... @abstractmethod - async def revoke_permission(self, user: UserInfo, permission_node: str) -> bool: ... + async def revoke_permission(self, user: UserInfo, permission_node: str) -> bool: + """ + 撤销用户的指定权限。 + + Args: + user: 目标用户信息 + permission_node: 要撤销的权限节点名称 + + Returns: + bool: True 表示撤销成功，False 表示失败 + """ + ... @abstractmethod - async def get_user_permissions(self, user: UserInfo) -> list[str]: ... + async def get_user_permissions(self, user: UserInfo) -> list[str]: + """ + 获取用户拥有的所有权限列表。 + + Args: + user: 目标用户信息 + + Returns: + list[str]: 用户拥有的权限节点名称列表 + """ + ... @abstractmethod - async def get_all_permission_nodes(self) -> list[PermissionNode]: ... + async def get_all_permission_nodes(self) -> list[PermissionNode]: + """ + 获取系统中所有已注册的权限节点。 + + Returns: + list[PermissionNode]: 所有权限节点的列表 + """ + ... @abstractmethod - async def get_plugin_permission_nodes(self, plugin_name: str) -> list[PermissionNode]: ... + async def get_plugin_permission_nodes(self, plugin_name: str) -> list[PermissionNode]: + """ + 获取指定插件注册的所有权限节点。 + + Args: + plugin_name: 插件名称 + + Returns: + list[PermissionNode]: 该插件注册的权限节点列表 + """ + ... class PermissionAPI: + """ + 权限API封装类。 + + 这是对外暴露的权限操作接口，插件和其他模块通过这个类来进行权限相关操作。 + 它封装了底层的 IPermissionManager，提供更简洁的调用方式。 + + 使用方式： + from src.plugin_system.apis.permission_api import permission_api + + # 检查权限 + has_perm = await permission_api.check_permission("qq", "12345", "chat.send") + + # 检查是否是管理员 + is_admin = await permission_api.is_master("qq", "12345") + + 设计模式： + 这是一个单例模式的变体，模块级别的 permission_api 实例供全局使用。 + """ + def __init__(self): - self._permission_manager: IPermissionManager | None = None + """ + 初始化 PermissionAPI。 + + 初始时权限管理器为 None，需要在系统启动时通过 set_permission_manager 设置。 + """ + self._permission_manager: IPermissionManager | None = None # 底层权限管理器实例 def set_permission_manager(self, manager: IPermissionManager): + """ + 设置权限管理器实例。 + + 这个方法应该在系统启动时被调用，注入具体的权限管理器实现。 + + Args: + manager: 实现了 IPermissionManager 接口的权限管理器实例 + """ self._permission_manager = manager def _ensure_manager(self): + """ + 确保权限管理器已设置（内部辅助方法）。 + + 如果权限管理器未设置，抛出 RuntimeError 异常。 + 这是一个防御性编程措施，帮助开发者快速发现配置问题。 + + Raises: + RuntimeError: 当权限管理器未设置时 + """ if self._permission_manager is None: raise RuntimeError("权限管理器未设置，请先调用 set_permission_manager") async def check_permission(self, platform: str, user_id: str, permission_node: str) -> bool: + """ + 检查用户是否拥有指定权限。 + + 这是最常用的权限检查方法，在执行需要权限的操作前调用。 + + Args: + platform: 用户所在平台（如 "qq", "telegram"） + user_id: 用户ID + permission_node: 要检查的权限节点名称 + + Returns: + bool: True 表示用户拥有权限，False 表示没有 + + Example: + if await permission_api.check_permission("qq", "12345", "admin.ban_user"): + # 执行封禁操作 + pass + """ self._ensure_manager() if not self._permission_manager: return False return await self._permission_manager.check_permission(UserInfo(platform, user_id), permission_node) async def is_master(self, platform: str, user_id: str) -> bool: + """ + 检查用户是否是管理员/主人。 + + 管理员是系统的最高权限用户，通常在配置文件中指定。 + + Args: + platform: 用户所在平台 + user_id: 用户ID + + Returns: + bool: True 表示是管理员，False 表示不是 + """ self._ensure_manager() if not self._permission_manager: return False @@ -87,11 +298,35 @@ class PermissionAPI: plugin_name: str, default_granted: bool = False, *, - allow_relative: bool = True, + allow_relative: bool = True, # 仅关键字参数，目前未使用，预留给相对权限名功能 ) -> bool: + """ + 注册一个新的权限节点。 + + 插件在初始化时应调用此方法注册自己需要的权限节点。 + + Args: + node_name: 权限节点名称，建议使用 "插件名.功能.操作" 的格式 + description: 权限描述，向用户解释这个权限的作用 + plugin_name: 注册此权限的插件名称 + default_granted: 是否默认授予所有用户（默认 False，需要显式授权） + allow_relative: 预留参数，是否允许相对权限名（目前未使用） + + Returns: + bool: True 表示注册成功，False 表示失败 + + Example: + await permission_api.register_permission_node( + node_name="my_plugin.chat.send_image", + description="允许发送图片消息", + plugin_name="my_plugin", + default_granted=True # 所有用户默认都能发图片 + ) + """ self._ensure_manager() - original_name = node_name + original_name = node_name # 保存原始名称（预留给相对路径处理） + # 创建权限节点对象 node = PermissionNode(node_name, description, plugin_name, default_granted) if not self._permission_manager: return False @@ -100,28 +335,90 @@ class PermissionAPI: async def grant_permission(self, platform: str, user_id: str, permission_node: str) -> bool: + """ + 授予用户指定权限。 + + 通常由管理员调用，给某个用户赋予特定权限。 + + Args: + platform: 目标用户所在平台 + user_id: 目标用户ID + permission_node: 要授予的权限节点名称 + + Returns: + bool: True 表示授权成功，False 表示失败 + + Example: + # 授予用户管理权限 + await permission_api.grant_permission("qq", "12345", "admin.manage_users") + """ self._ensure_manager() if not self._permission_manager: return False return await self._permission_manager.grant_permission(UserInfo(platform, user_id), permission_node) async def revoke_permission(self, platform: str, user_id: str, permission_node: str) -> bool: + """ + 撤销用户的指定权限。 + + 通常由管理员调用，移除某个用户的特定权限。 + + Args: + platform: 目标用户所在平台 + user_id: 目标用户ID + permission_node: 要撤销的权限节点名称 + + Returns: + bool: True 表示撤销成功，False 表示失败 + """ self._ensure_manager() if not self._permission_manager: return False return await self._permission_manager.revoke_permission(UserInfo(platform, user_id), permission_node) async def get_user_permissions(self, platform: str, user_id: str) -> list[str]: + """ + 获取用户拥有的所有权限列表。 + + 可用于展示用户的权限信息，或进行批量权限检查。 + + Args: + platform: 目标用户所在平台 + user_id: 目标用户ID + + Returns: + list[str]: 用户拥有的所有权限节点名称列表 + + Example: + perms = await permission_api.get_user_permissions("qq", "12345") + print(f"用户拥有以下权限: {perms}") + """ self._ensure_manager() if not self._permission_manager: return [] return await self._permission_manager.get_user_permissions(UserInfo(platform, user_id)) async def get_all_permission_nodes(self) -> list[dict[str, Any]]: + """ + 获取系统中所有已注册的权限节点。 + + 返回所有插件注册的权限节点信息，可用于权限管理界面展示。 + + Returns: + list[dict]: 权限节点信息列表，每个字典包含： + - node_name: 权限节点名称 + - description: 权限描述 + - plugin_name: 所属插件名称 + - default_granted: 是否默认授予 + + Note: + 返回字典而非 PermissionNode 对象，便于序列化和API响应。 + """ self._ensure_manager() if not self._permission_manager: return [] nodes = await self._permission_manager.get_all_permission_nodes() + # 将 PermissionNode 对象转换为字典，便于序列化 return [ { "node_name": n.node_name, @@ -133,10 +430,22 @@ class PermissionAPI: ] async def get_plugin_permission_nodes(self, plugin_name: str) -> list[dict[str, Any]]: + """ + 获取指定插件注册的所有权限节点。 + + 用于查看某个特定插件定义了哪些权限。 + + Args: + plugin_name: 插件名称 + + Returns: + list[dict]: 该插件的权限节点信息列表，格式同 get_all_permission_nodes + """ self._ensure_manager() if not self._permission_manager: return [] nodes = await self._permission_manager.get_plugin_permission_nodes(plugin_name) + # 将 PermissionNode 对象转换为字典 return [ { "node_name": n.node_name, @@ -148,4 +457,9 @@ class PermissionAPI: ] +# ============================================================ +# 模块级单例实例 +# ============================================================ +# 全局权限API实例，供其他模块导入使用 +# 使用方式: from src.plugin_system.apis.permission_api import permission_api permission_api = PermissionAPI() diff --git a/src/plugin_system/base/plus_command.py b/src/plugin_system/base/plus_command.py index dc39bbee8..991d81454 100644 --- a/src/plugin_system/base/plus_command.py +++ b/src/plugin_system/base/plus_command.py @@ -68,7 +68,7 @@ class PlusCommand(ABC): self.log_prefix = "[PlusCommand]" # chat_stream 会在运行时被 bot.py 设置 - self.chat_stream: "ChatStream | None" = None + self.chat_stream: "ChatStream" # 解析命令参数 self._parse_command() diff --git a/src/plugins/built_in/system_management/plugin.py b/src/plugins/built_in/system_management/plugin.py index 411caa34b..3f14075d2 100644 --- a/src/plugins/built_in/system_management/plugin.py +++ b/src/plugins/built_in/system_management/plugin.py @@ -31,6 +31,7 @@ from src.plugin_system.base.component_types import ( 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 +from src.plugin_system.apis.permission_api import permission_api logger = get_logger("SystemManagement") @@ -48,9 +49,15 @@ class SystemCommand(PlusCommand): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) - @require_permission("access", deny_message="❌ 你没有权限使用此命令") async def execute(self, args: CommandArgs) -> tuple[bool, str | None, bool]: """执行系统管理命令""" + if not self.chat_stream.user_info: + logger.error("chat_stream缺失用户信息,请报告开发者") + return False, "chat_stream缺失用户信息,请报告开发者", True + has_permission = await permission_api.check_permission(platform=self.chat_stream.platform,user_id=self.chat_stream.user_info.user_id,permission_node="access") + if has_permission: + logger.warning("没有权限使用此命令") + return False, "没有权限使用此命令", True if args.is_empty: await self._show_help("all") return True, "显示帮助信息", True