Merge branch 'dev' of https://github.com/MoFox-Studio/MoFox-Core into dev
This commit is contained in:
@@ -1,80 +1,291 @@
|
|||||||
"""纯异步权限API定义。所有外部调用方必须使用 await。"""
|
"""
|
||||||
|
纯异步权限API定义。
|
||||||
|
|
||||||
from abc import ABC, abstractmethod
|
这个模块提供了一套完整的权限管理系统,用于控制用户对各种功能的访问权限。
|
||||||
from dataclasses import dataclass
|
所有外部调用方必须使用 await 关键字,因为这是异步API。
|
||||||
from enum import Enum
|
|
||||||
from typing import Any
|
主要组件:
|
||||||
|
- 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
|
from src.common.logger import get_logger
|
||||||
|
|
||||||
logger = get_logger(__name__)
|
logger = get_logger(__name__) # 获取当前模块的日志记录器
|
||||||
|
|
||||||
|
|
||||||
class PermissionLevel(Enum):
|
class PermissionLevel(Enum):
|
||||||
MASTER = "master"
|
"""
|
||||||
|
权限等级枚举类。
|
||||||
|
|
||||||
|
定义了系统中的权限等级,目前只有 MASTER(管理员/主人)级别。
|
||||||
|
MASTER 用户拥有最高权限,可以执行所有操作。
|
||||||
|
"""
|
||||||
|
MASTER = "master" # 管理员/主人权限
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class PermissionNode:
|
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
|
@dataclass
|
||||||
class UserInfo:
|
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):
|
def __post_init__(self):
|
||||||
|
"""
|
||||||
|
dataclass 的后初始化钩子。
|
||||||
|
|
||||||
|
确保 user_id 始终是字符串类型,即使传入的是数字也会被转换。
|
||||||
|
这样可以避免类型不一致导致的比较问题。
|
||||||
|
"""
|
||||||
self.user_id = str(self.user_id)
|
self.user_id = str(self.user_id)
|
||||||
|
|
||||||
|
|
||||||
class IPermissionManager(ABC):
|
class IPermissionManager(ABC):
|
||||||
|
"""
|
||||||
|
权限管理器抽象接口(Interface)。
|
||||||
|
|
||||||
|
这是一个抽象基类,定义了权限管理器必须实现的所有方法。
|
||||||
|
具体的权限管理实现类需要继承此接口并实现所有抽象方法。
|
||||||
|
|
||||||
|
使用抽象接口的好处:
|
||||||
|
1. 解耦:PermissionAPI 不需要知道具体的实现细节
|
||||||
|
2. 可测试:可以轻松创建 Mock 实现用于测试
|
||||||
|
3. 可替换:可以随时更换不同的权限管理实现
|
||||||
|
"""
|
||||||
|
|
||||||
@abstractmethod
|
@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
|
@abstractmethod
|
||||||
async def is_master(self, user: UserInfo) -> bool: ... # 同步快速判断
|
async def is_master(self, user: UserInfo) -> bool:
|
||||||
|
"""
|
||||||
|
检查用户是否是管理员/主人。
|
||||||
|
|
||||||
|
管理员拥有最高权限,通常绕过所有权限检查。
|
||||||
|
|
||||||
|
Args:
|
||||||
|
user: 要检查的用户信息
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
bool: True 表示是管理员,False 表示不是
|
||||||
|
"""
|
||||||
|
...
|
||||||
|
|
||||||
@abstractmethod
|
@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
|
@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
|
@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
|
@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
|
@abstractmethod
|
||||||
async def get_all_permission_nodes(self) -> list[PermissionNode]: ...
|
async def get_all_permission_nodes(self) -> list[PermissionNode]:
|
||||||
|
"""
|
||||||
|
获取系统中所有已注册的权限节点。
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
list[PermissionNode]: 所有权限节点的列表
|
||||||
|
"""
|
||||||
|
...
|
||||||
|
|
||||||
@abstractmethod
|
@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:
|
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):
|
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):
|
def set_permission_manager(self, manager: IPermissionManager):
|
||||||
|
"""
|
||||||
|
设置权限管理器实例。
|
||||||
|
|
||||||
|
这个方法应该在系统启动时被调用,注入具体的权限管理器实现。
|
||||||
|
|
||||||
|
Args:
|
||||||
|
manager: 实现了 IPermissionManager 接口的权限管理器实例
|
||||||
|
"""
|
||||||
self._permission_manager = manager
|
self._permission_manager = manager
|
||||||
|
|
||||||
def _ensure_manager(self):
|
def _ensure_manager(self):
|
||||||
|
"""
|
||||||
|
确保权限管理器已设置(内部辅助方法)。
|
||||||
|
|
||||||
|
如果权限管理器未设置,抛出 RuntimeError 异常。
|
||||||
|
这是一个防御性编程措施,帮助开发者快速发现配置问题。
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
RuntimeError: 当权限管理器未设置时
|
||||||
|
"""
|
||||||
if self._permission_manager is None:
|
if self._permission_manager is None:
|
||||||
raise RuntimeError("权限管理器未设置,请先调用 set_permission_manager")
|
raise RuntimeError("权限管理器未设置,请先调用 set_permission_manager")
|
||||||
|
|
||||||
async def check_permission(self, platform: str, user_id: str, permission_node: str) -> bool:
|
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()
|
self._ensure_manager()
|
||||||
if not self._permission_manager:
|
if not self._permission_manager:
|
||||||
return False
|
return False
|
||||||
return await self._permission_manager.check_permission(UserInfo(platform, user_id), permission_node)
|
return await self._permission_manager.check_permission(UserInfo(platform, user_id), permission_node)
|
||||||
|
|
||||||
async def is_master(self, platform: str, user_id: str) -> bool:
|
async def is_master(self, platform: str, user_id: str) -> bool:
|
||||||
|
"""
|
||||||
|
检查用户是否是管理员/主人。
|
||||||
|
|
||||||
|
管理员是系统的最高权限用户,通常在配置文件中指定。
|
||||||
|
|
||||||
|
Args:
|
||||||
|
platform: 用户所在平台
|
||||||
|
user_id: 用户ID
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
bool: True 表示是管理员,False 表示不是
|
||||||
|
"""
|
||||||
self._ensure_manager()
|
self._ensure_manager()
|
||||||
if not self._permission_manager:
|
if not self._permission_manager:
|
||||||
return False
|
return False
|
||||||
@@ -87,11 +298,35 @@ class PermissionAPI:
|
|||||||
plugin_name: str,
|
plugin_name: str,
|
||||||
default_granted: bool = False,
|
default_granted: bool = False,
|
||||||
*,
|
*,
|
||||||
allow_relative: bool = True,
|
allow_relative: bool = True, # 仅关键字参数,目前未使用,预留给相对权限名功能
|
||||||
) -> bool:
|
) -> 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()
|
self._ensure_manager()
|
||||||
original_name = node_name
|
original_name = node_name # 保存原始名称(预留给相对路径处理)
|
||||||
|
|
||||||
|
# 创建权限节点对象
|
||||||
node = PermissionNode(node_name, description, plugin_name, default_granted)
|
node = PermissionNode(node_name, description, plugin_name, default_granted)
|
||||||
if not self._permission_manager:
|
if not self._permission_manager:
|
||||||
return False
|
return False
|
||||||
@@ -100,28 +335,90 @@ class PermissionAPI:
|
|||||||
|
|
||||||
|
|
||||||
async def grant_permission(self, platform: str, user_id: str, permission_node: str) -> bool:
|
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()
|
self._ensure_manager()
|
||||||
if not self._permission_manager:
|
if not self._permission_manager:
|
||||||
return False
|
return False
|
||||||
return await self._permission_manager.grant_permission(UserInfo(platform, user_id), permission_node)
|
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:
|
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()
|
self._ensure_manager()
|
||||||
if not self._permission_manager:
|
if not self._permission_manager:
|
||||||
return False
|
return False
|
||||||
return await self._permission_manager.revoke_permission(UserInfo(platform, user_id), permission_node)
|
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]:
|
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()
|
self._ensure_manager()
|
||||||
if not self._permission_manager:
|
if not self._permission_manager:
|
||||||
return []
|
return []
|
||||||
return await self._permission_manager.get_user_permissions(UserInfo(platform, user_id))
|
return await self._permission_manager.get_user_permissions(UserInfo(platform, user_id))
|
||||||
|
|
||||||
async def get_all_permission_nodes(self) -> list[dict[str, Any]]:
|
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()
|
self._ensure_manager()
|
||||||
if not self._permission_manager:
|
if not self._permission_manager:
|
||||||
return []
|
return []
|
||||||
nodes = await self._permission_manager.get_all_permission_nodes()
|
nodes = await self._permission_manager.get_all_permission_nodes()
|
||||||
|
# 将 PermissionNode 对象转换为字典,便于序列化
|
||||||
return [
|
return [
|
||||||
{
|
{
|
||||||
"node_name": n.node_name,
|
"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]]:
|
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()
|
self._ensure_manager()
|
||||||
if not self._permission_manager:
|
if not self._permission_manager:
|
||||||
return []
|
return []
|
||||||
nodes = await self._permission_manager.get_plugin_permission_nodes(plugin_name)
|
nodes = await self._permission_manager.get_plugin_permission_nodes(plugin_name)
|
||||||
|
# 将 PermissionNode 对象转换为字典
|
||||||
return [
|
return [
|
||||||
{
|
{
|
||||||
"node_name": n.node_name,
|
"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()
|
permission_api = PermissionAPI()
|
||||||
|
|||||||
@@ -68,7 +68,7 @@ class PlusCommand(ABC):
|
|||||||
self.log_prefix = "[PlusCommand]"
|
self.log_prefix = "[PlusCommand]"
|
||||||
|
|
||||||
# chat_stream 会在运行时被 bot.py 设置
|
# chat_stream 会在运行时被 bot.py 设置
|
||||||
self.chat_stream: "ChatStream | None" = None
|
self.chat_stream: "ChatStream"
|
||||||
|
|
||||||
# 解析命令参数
|
# 解析命令参数
|
||||||
self._parse_command()
|
self._parse_command()
|
||||||
|
|||||||
@@ -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.config_types import ConfigField
|
||||||
from src.plugin_system.base.plus_command import PlusCommand
|
from src.plugin_system.base.plus_command import PlusCommand
|
||||||
from src.plugin_system.utils.permission_decorators import require_permission
|
from src.plugin_system.utils.permission_decorators import require_permission
|
||||||
|
from src.plugin_system.apis.permission_api import permission_api
|
||||||
|
|
||||||
logger = get_logger("SystemManagement")
|
logger = get_logger("SystemManagement")
|
||||||
|
|
||||||
@@ -48,9 +49,15 @@ class SystemCommand(PlusCommand):
|
|||||||
def __init__(self, *args, **kwargs):
|
def __init__(self, *args, **kwargs):
|
||||||
super().__init__(*args, **kwargs)
|
super().__init__(*args, **kwargs)
|
||||||
|
|
||||||
@require_permission("access", deny_message="❌ 你没有权限使用此命令")
|
|
||||||
async def execute(self, args: CommandArgs) -> tuple[bool, str | None, bool]:
|
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:
|
if args.is_empty:
|
||||||
await self._show_help("all")
|
await self._show_help("all")
|
||||||
return True, "显示帮助信息", True
|
return True, "显示帮助信息", True
|
||||||
@@ -81,7 +88,7 @@ class SystemCommand(PlusCommand):
|
|||||||
🔧 主要功能:
|
🔧 主要功能:
|
||||||
• `/system help` - 显示此帮助
|
• `/system help` - 显示此帮助
|
||||||
• `/system permission` - 权限管理
|
• `/system permission` - 权限管理
|
||||||
• `/system plugin` - 插件管理
|
• `/system plugin` - 插件与组件管理
|
||||||
• `/system schedule` - 定时任务管理
|
• `/system schedule` - 定时任务管理
|
||||||
• `/system prompt` - 提示词注入管理
|
• `/system prompt` - 提示词注入管理
|
||||||
"""
|
"""
|
||||||
@@ -98,16 +105,18 @@ class SystemCommand(PlusCommand):
|
|||||||
"""
|
"""
|
||||||
elif target == "plugin":
|
elif target == "plugin":
|
||||||
help_text = """🔌 插件管理命令帮助
|
help_text = """🔌 插件管理命令帮助
|
||||||
📋 基本操作:
|
|
||||||
• `/system plugin help` - 显示插件管理帮助
|
|
||||||
• `/system plugin report` - 查看系统插件报告
|
|
||||||
• `/system plugin rescan` - 重新扫描所有插件目录
|
|
||||||
|
|
||||||
⚙️ 插件控制:
|
⚙️ 插件控制:
|
||||||
|
• `/system plugin rescan` - 重新扫描所有插件目录
|
||||||
• `/system plugin load <插件名>` - 加载指定插件
|
• `/system plugin load <插件名>` - 加载指定插件
|
||||||
• `/system plugin reload <插件名>` - 重新加载指定插件
|
• `/system plugin reload <插件名>` - 重新加载指定插件
|
||||||
• `/system plugin reload_all` - 重新加载所有插件
|
• `/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 <QQ号>]` - 在指定会话局部启用组件
|
• `/system plugin enable_local <名称> [group <群号> | private <QQ号>]` - 在指定会话局部启用组件
|
||||||
• `/system plugin disable_local <名称> [group <群号> | private <QQ号>]` - 在指定会话局部禁用组件
|
• `/system plugin disable_local <名称> [group <群号> | private <QQ号>]` - 在指定会话局部禁用组件
|
||||||
"""
|
"""
|
||||||
@@ -156,6 +165,15 @@ class SystemCommand(PlusCommand):
|
|||||||
await self._show_help("plugin")
|
await self._show_help("plugin")
|
||||||
elif action in ["report", "报告"]:
|
elif action in ["report", "报告"]:
|
||||||
await self._show_system_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", "重扫"]:
|
elif action in ["rescan", "重扫"]:
|
||||||
await self._rescan_plugin_dirs()
|
await self._rescan_plugin_dirs()
|
||||||
elif action in ["load", "加载"] and len(remaining_args) > 0:
|
elif action in ["load", "加载"] and len(remaining_args) > 0:
|
||||||
@@ -164,6 +182,10 @@ class SystemCommand(PlusCommand):
|
|||||||
await self._reload_plugin(remaining_args[0])
|
await self._reload_plugin(remaining_args[0])
|
||||||
elif action in ["reload_all", "重载全部"]:
|
elif action in ["reload_all", "重载全部"]:
|
||||||
await self._reload_all_plugins()
|
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:
|
elif action in ["enable_local", "局部启用"] and len(remaining_args) >= 1:
|
||||||
await self._set_local_component_state(remaining_args, enabled=True)
|
await self._set_local_component_state(remaining_args, enabled=True)
|
||||||
elif action in ["disable_local", "局部禁用"] and len(remaining_args) >= 1:
|
elif action in ["disable_local", "局部禁用"] and len(remaining_args) >= 1:
|
||||||
@@ -430,8 +452,193 @@ class SystemCommand(PlusCommand):
|
|||||||
await self.send_text("\n".join(response_parts))
|
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="❌ 你没有权限查看插件报告")
|
@require_permission("plugin.manage", deny_message="❌ 你没有权限查看插件报告")
|
||||||
async def _show_system_report(self):
|
async def _show_system_report(self):
|
||||||
@@ -784,13 +991,13 @@ class SystemCommand(PlusCommand):
|
|||||||
@register_plugin
|
@register_plugin
|
||||||
class SystemManagementPlugin(BasePlugin):
|
class SystemManagementPlugin(BasePlugin):
|
||||||
plugin_name: str = "system_management"
|
plugin_name: str = "system_management"
|
||||||
enable_plugin: bool = True
|
enable_plugin: bool = False
|
||||||
dependencies: ClassVar[list[str]] = [] # 插件依赖列表
|
dependencies: ClassVar[list[str]] = [] # 插件依赖列表
|
||||||
python_dependencies: ClassVar[list[str]] = [] # Python包依赖列表,现在使用内置API
|
python_dependencies: ClassVar[list[str]] = [] # Python包依赖列表,现在使用内置API
|
||||||
config_file_name: str = "config.toml" # 配置文件名
|
config_file_name: str = "config.toml" # 配置文件名
|
||||||
config_schema: ClassVar[dict] = {
|
config_schema: ClassVar[dict] = {
|
||||||
"plugin": {
|
"plugin": {
|
||||||
"enabled": ConfigField(bool, default=True, description="是否启用插件"),
|
"enabled": ConfigField(bool, default=False, description="是否启用插件"),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user