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

Merge branch 'master' of https://github.com/MoFox-Studio/MoFox_Bot

Browse Source
This commit is contained in:
minecraft1024a
2025-08-28 18:42:37 +08:00
parent 7ba3b7f34d b28116e17b
commit 7bd1320504
11 changed files with 184 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/chat/chat_loop/cycle_processor.py
View File

@@ -100,7 +100,7 @@ class CycleProcessor:
from src.plugin_system.core.event_manager import event_manager from src.plugin_system.core.event_manager import event_manager
from src.plugin_system.base.component_types import EventType from src.plugin_system.base.component_types import EventType
# 触发 ON_PLAN 事件 # 触发 ON_PLAN 事件
result = await event_manager.trigger_event(EventType.ON_PLAN, stream_id=self.context.stream_id) result = await event_manager.trigger_event(EventType.ON_PLAN, plugin_name="SYSTEM", stream_id=self.context.stream_id)
if result and not result.all_continue_process(): if result and not result.all_continue_process():
return return

2
src/chat/message_receive/bot.py
View File

@@ -437,7 +437,7 @@ class ChatBot:
logger.info(f"命令处理完成，跳过后续消息处理: {cmd_result}") logger.info(f"命令处理完成，跳过后续消息处理: {cmd_result}")
return return
result = await event_manager.trigger_event(EventType.ON_MESSAGE,message=message) result = await event_manager.trigger_event(EventType.ON_MESSAGE,plugin_name="SYSTEM",message=message)
if not result.all_continue_process(): if not result.all_continue_process():
raise UserWarning(f"插件{result.get_summary().get('stopped_handlers','')}于消息到达时取消了消息处理") raise UserWarning(f"插件{result.get_summary().get('stopped_handlers','')}于消息到达时取消了消息处理")

4
src/chat/replyer/default_generator.py
View File

@@ -370,7 +370,7 @@ class DefaultReplyer:
from src.plugin_system.core.event_manager import event_manager from src.plugin_system.core.event_manager import event_manager
if not from_plugin: if not from_plugin:
result = await event_manager.trigger_event(EventType.POST_LLM,prompt=prompt,stream_id=stream_id) result = await event_manager.trigger_event(EventType.POST_LLM,plugin_name="SYSTEM",prompt=prompt,stream_id=stream_id)
if not result.all_continue_process(): if not result.all_continue_process():
raise UserWarning(f"插件{result.get_summary().get('stopped_handlers', '')}于请求前中断了内容生成") raise UserWarning(f"插件{result.get_summary().get('stopped_handlers', '')}于请求前中断了内容生成")
@@ -390,7 +390,7 @@ class DefaultReplyer:
} }
# 触发 AFTER_LLM 事件 # 触发 AFTER_LLM 事件
if not from_plugin: if not from_plugin:
result = await event_manager.trigger_event(EventType.AFTER_LLM,prompt=prompt,llm_response=llm_response,stream_id=stream_id) result = await event_manager.trigger_event(EventType.AFTER_LLM,plugin_name="SYSTEM",prompt=prompt,llm_response=llm_response,stream_id=stream_id)
if not result.all_continue_process(): if not result.all_continue_process():
raise UserWarning(f"插件{result.get_summary().get('stopped_handlers','')}于请求后取消了内容生成") raise UserWarning(f"插件{result.get_summary().get('stopped_handlers','')}于请求后取消了内容生成")
except UserWarning as e: except UserWarning as e:

2
src/main.py
View File

@@ -254,7 +254,7 @@ MoFox_Bot(第三方修改版)
try: try:
await event_manager.trigger_event(EventType.ON_START) await event_manager.trigger_event(EventType.ON_START,plugin_name="SYSTEM")
init_time = int(1000 * (time.time() - init_start_time)) init_time = int(1000 * (time.time() - init_start_time))
logger.info(f"初始化完成，神经元放电{init_time}次") logger.info(f"初始化完成，神经元放电{init_time}次")
except Exception as e: except Exception as e:

11
src/plugin_system/base/base_event.py
View File

@@ -9,7 +9,7 @@ class HandlerResult:
所有事件处理器必须返回此类的实例 所有事件处理器必须返回此类的实例
""" """
def __init__(self, success: bool, continue_process: bool, message: str = "", handler_name: str = ""): def __init__(self, success: bool, continue_process: bool, message: Any = {}, handler_name: str = ""):
self.success = success self.success = success
self.continue_process = continue_process self.continue_process = continue_process
self.message = message self.message = message
@@ -67,9 +67,16 @@ class HandlerResultsCollection:
} }
class BaseEvent: class BaseEvent:
def __init__(self, name: str): def __init__(
self,
name: str,
allowed_subscribers: List[str]=[],
allowed_triggers: List[str]=[]
):
self.name = name self.name = name
self.enabled = True self.enabled = True
self.allowed_subscribers = allowed_subscribers # 记录事件处理器名
self.allowed_triggers = allowed_triggers # 记录插件名
from src.plugin_system.base.base_events_handler import BaseEventHandler from src.plugin_system.base.base_events_handler import BaseEventHandler
self.subscribers: List["BaseEventHandler"] = [] # 订阅该事件的事件处理器列表 self.subscribers: List["BaseEventHandler"] = [] # 订阅该事件的事件处理器列表

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

@@ -51,6 +51,7 @@ class CommandArgs:
return self._parsed_args return self._parsed_args
@property
def is_empty(self) -> bool: def is_empty(self) -> bool:
"""检查参数是否为空 """检查参数是否为空
@@ -74,6 +75,7 @@ class CommandArgs:
return args[index] return args[index]
return default return default
@property
def get_first(self, default: str = "") -> str: def get_first(self, default: str = "") -> str:
"""获取第一个参数 """获取第一个参数
@@ -85,7 +87,7 @@ class CommandArgs:
""" """
return self.get_arg(0, default) return self.get_arg(0, default)
def get_remaining(self, start_index: int = 1) -> str: def get_remaining(self, start_index: int = 0) -> str:
"""获取从指定索引开始的剩余参数字符串 """获取从指定索引开始的剩余参数字符串
Args: Args:

30
src/plugin_system/core/event_manager.py
View File

@@ -40,12 +40,18 @@ class EventManager:
self._initialized = True self._initialized = True
logger.info("EventManager 单例初始化完成") logger.info("EventManager 单例初始化完成")
def register_event(self, event_name: Union[EventType, str]) -> bool: def register_event(
self,
event_name: Union[EventType, str],
allowed_subscribers: List[str]=[],
allowed_triggers: List[str]=[]
) -> bool:
"""注册一个新的事件 """注册一个新的事件
Args: Args:
event_name Union[EventType, str]: 事件名称 event_name Union[EventType, str]: 事件名称
allowed_subscribers: List[str]: 事件订阅者白名单,
allowed_triggers: List[str]: 事件触发插件白名单
Returns: Returns:
bool: 注册成功返回True，已存在返回False bool: 注册成功返回True，已存在返回False
""" """
@@ -53,7 +59,7 @@ class EventManager:
logger.warning(f"事件 {event_name} 已存在，跳过注册") logger.warning(f"事件 {event_name} 已存在，跳过注册")
return False return False
event = BaseEvent(event_name) event = BaseEvent(event_name,allowed_subscribers,allowed_triggers)
self._events[event_name] = event self._events[event_name] = event
logger.info(f"事件 {event_name} 注册成功") logger.info(f"事件 {event_name} 注册成功")
@@ -211,6 +217,11 @@ class EventManager:
logger.warning(f"事件处理器 {handler_name} 已经订阅了事件 {event_name}，跳过重复订阅") logger.warning(f"事件处理器 {handler_name} 已经订阅了事件 {event_name}，跳过重复订阅")
return True return True
# 白名单检查
if event.allowed_subscribers and handler_name not in event.allowed_subscribers:
logger.warning(f"事件处理器 {handler_name} 不在事件 {event_name} 的订阅者白名单中，无法订阅")
return False
event.subscribers.append(handler_instance) event.subscribers.append(handler_instance)
# 按权重从高到低排序订阅者 # 按权重从高到低排序订阅者
@@ -264,11 +275,12 @@ class EventManager:
return {handler.handler_name: handler for handler in event.subscribers} return {handler.handler_name: handler for handler in event.subscribers}
async def trigger_event(self, event_name: Union[EventType, str], **kwargs) -> Optional[HandlerResultsCollection]: async def trigger_event(self, event_name: Union[EventType, str], plugin_name: Optional[str]="", **kwargs) -> Optional[HandlerResultsCollection]:
"""触发指定事件 """触发指定事件
Args: Args:
event_name Union[EventType, str]: 事件名称 event_name Union[EventType, str]: 事件名称
plugin_name str: 触发事件的插件名
**kwargs: 传递给处理器的参数 **kwargs: 传递给处理器的参数
Returns: Returns:
@@ -281,6 +293,14 @@ class EventManager:
logger.error(f"事件 {event_name} 不存在，无法触发") logger.error(f"事件 {event_name} 不存在，无法触发")
return None return None
# 插件白名单检查
if event.allowed_triggers and not plugin_name:
logger.warning(f"事件 {event_name} 存在触发者白名单，缺少plugin_name无法验证权限，已拒绝触发！")
return None
elif event.allowed_triggers and plugin_name not in event.allowed_triggers:
logger.warning(f"插件 {plugin_name} 没有权限触发事件 {event_name}，已拒绝触发！")
return None
return await event.activate(params) return await event.activate(params)
def init_default_events(self) -> None: def init_default_events(self) -> None:
@@ -297,7 +317,7 @@ class EventManager:
] ]
for event_name in default_events: for event_name in default_events:
self.register_event(event_name) self.register_event(event_name,allowed_triggers=["SYSTEM"])
logger.info("默认事件初始化完成") logger.info("默认事件初始化完成")

64
src/plugin_system/utils/permission_decorators.py
View File

@@ -9,9 +9,9 @@ from typing import Callable, Optional
from inspect import iscoroutinefunction from inspect import iscoroutinefunction
from src.plugin_system.apis.permission_api import permission_api from src.plugin_system.apis.permission_api import permission_api
from src.plugin_system.apis.send_api import send_message from src.plugin_system.apis.send_api import text_to_stream
from src.plugin_system.apis.logging_api import get_logger from src.plugin_system.apis.logging_api import get_logger
from src.common.message import ChatStream from src.chat.message_receive.chat_stream import ChatStream
logger = get_logger(__name__) logger = get_logger(__name__)
@@ -37,6 +37,8 @@ def require_permission(permission_node: str, deny_message: Optional[str] = None)
async def async_wrapper(*args, **kwargs): async def async_wrapper(*args, **kwargs):
# 尝试从参数中提取 ChatStream 对象 # 尝试从参数中提取 ChatStream 对象
chat_stream = None chat_stream = None
# 首先检查位置参数中的 ChatStream
for arg in args: for arg in args:
if isinstance(arg, ChatStream): if isinstance(arg, ChatStream):
chat_stream = arg chat_stream = arg
@@ -46,21 +48,31 @@ def require_permission(permission_node: str, deny_message: Optional[str] = None)
if chat_stream is None: if chat_stream is None:
chat_stream = kwargs.get('chat_stream') chat_stream = kwargs.get('chat_stream')
# 如果还没找到，检查是否是 PlusCommand 方法调用
if chat_stream is None and args:
# 检查第一个参数是否有 message.chat_stream 属性（PlusCommand 实例）
instance = args[0]
if hasattr(instance, 'message') and hasattr(instance.message, 'chat_stream'):
chat_stream = instance.message.chat_stream
if chat_stream is None: if chat_stream is None:
logger.error(f"权限装饰器无法找到 ChatStream 对象，函数: {func.__name__}") logger.error(f"权限装饰器无法找到 ChatStream 对象，函数: {func.__name__}")
return return
# 检查权限 # 检查权限
has_permission = permission_api.check_permission( has_permission = permission_api.check_permission(
chat_stream.user_platform, chat_stream.platform,
chat_stream.user_id, chat_stream.user_info.user_id,
permission_node permission_node
) )
if not has_permission: if not has_permission:
# 权限不足，发送拒绝消息 # 权限不足，发送拒绝消息
message = deny_message or f"❌ 你没有执行此操作的权限\n需要权限: {permission_node}" message = deny_message or f"❌ 你没有执行此操作的权限\n需要权限: {permission_node}"
await send_message(chat_stream, message) await text_to_stream(message, chat_stream.stream_id)
# 对于PlusCommand的execute方法，需要返回适当的元组
if func.__name__ == 'execute' and hasattr(args[0], 'send_text'):
return False, "权限不足", True
return return
# 权限检查通过，执行原函数 # 权限检查通过，执行原函数
@@ -83,13 +95,13 @@ def require_permission(permission_node: str, deny_message: Optional[str] = None)
# 检查权限 # 检查权限
has_permission = permission_api.check_permission( has_permission = permission_api.check_permission(
chat_stream.user_platform, chat_stream.platform,
chat_stream.user_id, chat_stream.user_info.user_id,
permission_node permission_node
) )
if not has_permission: if not has_permission:
logger.warning(f"用户 {chat_stream.user_platform}:{chat_stream.user_id} 没有权限 {permission_node}") logger.warning(f"用户 {chat_stream.platform}:{chat_stream.user_info.user_id} 没有权限 {permission_node}")
return return
# 权限检查通过，执行原函数 # 权限检查通过，执行原函数
@@ -124,6 +136,8 @@ def require_master(deny_message: Optional[str] = None):
async def async_wrapper(*args, **kwargs): async def async_wrapper(*args, **kwargs):
# 尝试从参数中提取 ChatStream 对象 # 尝试从参数中提取 ChatStream 对象
chat_stream = None chat_stream = None
# 首先检查位置参数中的 ChatStream
for arg in args: for arg in args:
if isinstance(arg, ChatStream): if isinstance(arg, ChatStream):
chat_stream = arg chat_stream = arg
@@ -133,20 +147,28 @@ def require_master(deny_message: Optional[str] = None):
if chat_stream is None: if chat_stream is None:
chat_stream = kwargs.get('chat_stream') chat_stream = kwargs.get('chat_stream')
# 如果还没找到，检查是否是 PlusCommand 方法调用
if chat_stream is None and args:
# 检查第一个参数是否有 message.chat_stream 属性（PlusCommand 实例）
instance = args[0]
if hasattr(instance, 'message') and hasattr(instance.message, 'chat_stream'):
chat_stream = instance.message.chat_stream
if chat_stream is None: if chat_stream is None:
logger.error(f"Master权限装饰器无法找到 ChatStream 对象，函数: {func.__name__}") logger.error(f"Master权限装饰器无法找到 ChatStream 对象，函数: {func.__name__}")
return return
# 检查是否为Master用户 # 检查是否为Master用户
is_master = permission_api.is_master( is_master = permission_api.is_master(
chat_stream.user_platform, chat_stream.platform,
chat_stream.user_id chat_stream.user_info.user_id
) )
if not is_master: if not is_master:
# 权限不足，发送拒绝消息
message = deny_message or "❌ 此操作仅限Master用户执行" message = deny_message or "❌ 此操作仅限Master用户执行"
await send_message(chat_stream, message) await text_to_stream(message, chat_stream.stream_id)
if func.__name__ == 'execute' and hasattr(args[0], 'send_text'):
return False, "需要Master权限", True
return return
# 权限检查通过，执行原函数 # 权限检查通过，执行原函数
@@ -169,12 +191,12 @@ def require_master(deny_message: Optional[str] = None):
# 检查是否为Master用户 # 检查是否为Master用户
is_master = permission_api.is_master( is_master = permission_api.is_master(
chat_stream.user_platform, chat_stream.platform,
chat_stream.user_id chat_stream.user_info.user_id
) )
if not is_master: if not is_master:
logger.warning(f"用户 {chat_stream.user_platform}:{chat_stream.user_id} 不是Master用户") logger.warning(f"用户 {chat_stream.platform}:{chat_stream.user_info.user_id} 不是Master用户")
return return
# 权限检查通过，执行原函数 # 权限检查通过，执行原函数
@@ -209,8 +231,8 @@ class PermissionChecker:
bool: 是否拥有权限 bool: 是否拥有权限
""" """
return permission_api.check_permission( return permission_api.check_permission(
chat_stream.user_platform, chat_stream.platform,
chat_stream.user_id, chat_stream.user_info.user_id,
permission_node permission_node
) )
@@ -226,8 +248,8 @@ class PermissionChecker:
bool: 是否为Master用户 bool: 是否为Master用户
""" """
return permission_api.is_master( return permission_api.is_master(
chat_stream.user_platform, chat_stream.platform,
chat_stream.user_id chat_stream.user_info.user_id
) )
@staticmethod @staticmethod
@@ -248,7 +270,7 @@ class PermissionChecker:
if not has_permission: if not has_permission:
message = deny_message or f"❌ 你没有执行此操作的权限\n需要权限: {permission_node}" message = deny_message or f"❌ 你没有执行此操作的权限\n需要权限: {permission_node}"
await send_message(chat_stream, message) await text_to_stream(message, chat_stream.stream_id)
return has_permission return has_permission
@@ -269,6 +291,6 @@ class PermissionChecker:
if not is_master: if not is_master:
message = deny_message or "❌ 此操作仅限Master用户执行" message = deny_message or "❌ 此操作仅限Master用户执行"
await send_message(chat_stream, message) await text_to_stream(message, chat_stream.stream_id)
return is_master return is_master

36
src/plugins/built_in/maizone_refactored/commands/send_feed_command.py
View File

@@ -5,52 +5,32 @@
from typing import Tuple from typing import Tuple
from src.common.logger import get_logger from src.common.logger import get_logger
from src.plugin_system import BaseCommand from src.plugin_system.base.plus_command import PlusCommand
from src.plugin_system.base.command_args import CommandArgs
from src.plugin_system.utils.permission_decorators import require_permission
from ..services.manager import get_qzone_service, get_config_getter from ..services.manager import get_qzone_service, get_config_getter
logger = get_logger("MaiZone.SendFeedCommand") logger = get_logger("MaiZone.SendFeedCommand")
class SendFeedCommand(BaseCommand): class SendFeedCommand(PlusCommand):
""" """
响应用户通过 `/send_feed` 命令发送说说的请求。 响应用户通过 `/send_feed` 命令发送说说的请求。
""" """
command_name: str = "send_feed" command_name: str = "send_feed"
command_description: str = "发送一条QQ空间说说" command_description: str = "发送一条QQ空间说说"
command_pattern: str = r"^/send_feed(?:\s+(?P<topic>.*))?$" command_aliases = ["发空间"]
command_help: str = "使用 /send_feed [主题] 来发送一条说说"
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs) super().__init__(*args, **kwargs)
def _check_permission(self) -> bool: @require_permission("plugin.send.permission")
"""检查当前用户是否有权限执行此命令""" async def execute(self, args: CommandArgs) -> Tuple[bool, str, bool]:
user_id = self.message.message_info.user_info.user_id
if not user_id:
return False
get_config = get_config_getter()
permission_list = get_config("send.permission", [])
permission_type = get_config("send.permission_type", "whitelist")
if not isinstance(permission_list, list):
return False
if permission_type == 'whitelist':
return user_id in permission_list
elif permission_type == 'blacklist':
return user_id not in permission_list
return False
async def execute(self) -> Tuple[bool, str, bool]:
""" """
执行命令的核心逻辑。 执行命令的核心逻辑。
""" """
if not self._check_permission():
await self.send_text("抱歉，你没有权限使用这个命令哦。")
return False, "权限不足", True
topic = self.matched_groups.get("topic", "") topic = args.get_remaining()
stream_id = self.message.chat_stream.stream_id stream_id = self.message.chat_stream.stream_id
await self.send_text(f"收到！正在为你生成关于“{topic or '随机'}”的说说，请稍候...") await self.send_text(f"收到！正在为你生成关于“{topic or '随机'}”的说说，请稍候...")

10
src/plugins/built_in/maizone_refactored/plugin.py
View File

@@ -13,6 +13,7 @@ from src.plugin_system import (
register_plugin register_plugin
) )
from src.plugin_system.base.config_types import ConfigField from src.plugin_system.base.config_types import ConfigField
from src.plugin_system.apis.permission_api import permission_api
from .actions.read_feed_action import ReadFeedAction from .actions.read_feed_action import ReadFeedAction
from .actions.send_feed_action import SendFeedAction from .actions.send_feed_action import SendFeedAction
@@ -82,7 +83,12 @@ class MaiZoneRefactoredPlugin(BasePlugin):
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs) super().__init__(*args, **kwargs)
permission_api.register_permission_node(
"plugin.send.permission",
"是否可以使用机器人发送说说",
"maiZone",
False
)
content_service = ContentService(self.get_config) content_service = ContentService(self.get_config)
image_service = ImageService(self.get_config) image_service = ImageService(self.get_config)
cookie_service = CookieService(self.get_config) cookie_service = CookieService(self.get_config)
@@ -102,5 +108,5 @@ class MaiZoneRefactoredPlugin(BasePlugin):
return [ return [
(SendFeedAction.get_action_info(), SendFeedAction), (SendFeedAction.get_action_info(), SendFeedAction),
(ReadFeedAction.get_action_info(), ReadFeedAction), (ReadFeedAction.get_action_info(), ReadFeedAction),
(SendFeedCommand.get_command_info(), SendFeedCommand), (SendFeedCommand.get_plus_command_info(), SendFeedCommand),
] ]

142
src/plugins/built_in/permission_management/plugin.py
View File

@@ -16,6 +16,7 @@ from src.plugin_system.apis.permission_api import permission_api
from src.plugin_system.apis.logging_api import get_logger from src.plugin_system.apis.logging_api import get_logger
from src.plugin_system.base.component_types import PlusCommandInfo, ChatType from src.plugin_system.base.component_types import PlusCommandInfo, ChatType
from src.plugin_system.base.config_types import ConfigField from src.plugin_system.base.config_types import ConfigField
from src.plugin_system.utils.permission_decorators import require_permission, require_master, PermissionChecker
logger = get_logger("Permission") logger = get_logger("Permission")
@@ -49,67 +50,35 @@ class PermissionCommand(PlusCommand):
async def execute(self, args: CommandArgs) -> Tuple[bool, Optional[str], bool]: async def execute(self, args: CommandArgs) -> Tuple[bool, Optional[str], bool]:
"""执行权限管理命令""" """执行权限管理命令"""
if args.is_empty(): if args.is_empty:
await self._show_help() await self._show_help()
return True, "显示帮助信息", True return True, "显示帮助信息", True
subcommand = args.get_first().lower() subcommand = args.get_first.lower()
remaining_args = args.get_args()[1:] # 获取除第一个参数外的所有参数 remaining_args = args.get_args()[1:] # 获取除第一个参数外的所有参数
chat_stream = self.message.chat_stream chat_stream = self.message.chat_stream
# 检查基本查看权限
can_view = permission_api.check_permission(
chat_stream.platform,
chat_stream.user_info.user_id,
"plugin.permission.view"
) or permission_api.is_master(chat_stream.platform, chat_stream.user_info.user_id)
# 检查管理权限
can_manage = permission_api.check_permission(
chat_stream.platform,
chat_stream.user_info.user_id,
"plugin.permission.manage"
) or permission_api.is_master(chat_stream.platform, chat_stream.user_info.user_id)
if subcommand in ["grant", "授权", "give"]: if subcommand in ["grant", "授权", "give"]:
if not can_manage:
await self.send_text("❌ 你没有权限管理的权限")
return True, "权限不足", True
await self._grant_permission(chat_stream, remaining_args) await self._grant_permission(chat_stream, remaining_args)
return True, "执行授权命令", True return True, "执行授权命令", True
elif subcommand in ["revoke", "撤销", "remove"]: elif subcommand in ["revoke", "撤销", "remove"]:
if not can_manage:
await self.send_text("❌ 你没有权限管理的权限")
return True, "权限不足", True
await self._revoke_permission(chat_stream, remaining_args) await self._revoke_permission(chat_stream, remaining_args)
return True, "执行撤销命令", True return True, "执行撤销命令", True
elif subcommand in ["list", "列表", "ls"]: elif subcommand in ["list", "列表", "ls"]:
if not can_view:
await self.send_text("❌ 你没有查看权限的权限")
return True, "权限不足", True
await self._list_permissions(chat_stream, remaining_args) await self._list_permissions(chat_stream, remaining_args)
return True, "执行列表命令", True return True, "执行列表命令", True
elif subcommand in ["check", "检查"]: elif subcommand in ["check", "检查"]:
if not can_view:
await self.send_text("❌ 你没有查看权限的权限")
return True, "权限不足", True
await self._check_permission(chat_stream, remaining_args) await self._check_permission(chat_stream, remaining_args)
return True, "执行检查命令", True return True, "执行检查命令", True
elif subcommand in ["nodes", "节点"]: elif subcommand in ["nodes", "节点"]:
if not can_view:
await self.send_text("❌ 你没有查看权限的权限")
return True, "权限不足", True
await self._list_nodes(chat_stream, remaining_args) await self._list_nodes(chat_stream, remaining_args)
return True, "执行节点命令", True return True, "执行节点命令", True
elif subcommand in ["allnodes", "全部节点", "all"]: elif subcommand in ["allnodes", "全部节点", "all"]:
if not can_view:
await self.send_text("❌ 你没有查看权限的权限")
return True, "权限不足", True
await self._list_all_nodes_with_description(chat_stream) await self._list_all_nodes_with_description(chat_stream)
return True, "执行全部节点命令", True return True, "执行全部节点命令", True
@@ -149,74 +118,113 @@ class PermissionCommand(PlusCommand):
await self.send_text(help_text) await self.send_text(help_text)
def _parse_user_mention(self, mention: str) -> Optional[str]: def _parse_user_mention(self, mention: str) -> Optional[str]:
"""解析用户提及，提取QQ号""" """解析用户提及，提取QQ号
支持的格式:
- @<用户名:QQ号> 格式
- [CQ:at,qq=QQ号] 格式
- 直接的QQ号
"""
# 匹配 @<用户名:QQ号> 格式，提取QQ号 # 匹配 @<用户名:QQ号> 格式，提取QQ号
at_match = re.search(r'@<[^:]+:(\d+)>', mention) at_match = re.search(r'@<[^:]+:(\d+)>', mention)
if at_match: if at_match:
return at_match.group(1) return at_match.group(1)
# 直接是数字 # 直接是数字
if mention.isdigit(): if mention.isdigit():
return mention return mention
return None return None
@staticmethod
def parse_user_from_args(args: CommandArgs, index: int = 0) -> Optional[str]:
"""从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]): async def _grant_permission(self, chat_stream, args: List[str]):
"""授权用户权限""" """授权用户权限"""
if len(args) < 2: if len(args) < 2:
await self.send_text("❌ 用法: /permission grant <@用户|QQ号> <权限节点>") await self.send_text("❌ 用法: /permission grant <@用户|QQ号> <权限节点>")
return return
user_mention = args[0] # 解析用户ID - 使用新的解析方法
permission_node = args[1] user_id = self._parse_user_mention(args[0])
# 解析用户ID
user_id = self._parse_user_mention(user_mention)
if not user_id: if not user_id:
await self.send_text("❌ 无效的用户格式，请使用 @用户 或直接输入QQ号") await self.send_text("❌ 无效的用户格式，请使用 @<用户名:QQ号> 或直接输入QQ号")
return return
permission_node = args[1]
# 执行授权 # 执行授权
success = permission_api.grant_permission(chat_stream.platform, user_id, permission_node) success = permission_api.grant_permission(chat_stream.platform, user_id, permission_node)
if success: if success:
await self.send_text(f"✅ 已授权用户 {user_id} 权限节点 {permission_node}") await self.send_text(f"✅ 已授权用户 {user_id} 权限节点 `{permission_node}`")
else: else:
await self.send_text("❌ 授权失败，请检查权限节点是否存在") await self.send_text("❌ 授权失败，请检查权限节点是否存在")
@require_permission("plugin.permission.manage", "❌ 你没有权限管理的权限")
async def _revoke_permission(self, chat_stream, args: List[str]): async def _revoke_permission(self, chat_stream, args: List[str]):
"""撤销用户权限""" """撤销用户权限"""
if len(args) < 2: if len(args) < 2:
await self.send_text("❌ 用法: /permission revoke <@用户|QQ号> <权限节点>") await self.send_text("❌ 用法: /permission revoke <@用户|QQ号> <权限节点>")
return return
user_mention = args[0] # 解析用户ID - 使用新的解析方法
permission_node = args[1] user_id = self._parse_user_mention(args[0])
# 解析用户ID
user_id = self._parse_user_mention(user_mention)
if not user_id: if not user_id:
await self.send_text("❌ 无效的用户格式，请使用 @用户 或直接输入QQ号") await self.send_text("❌ 无效的用户格式，请使用 @<用户名:QQ号> 或直接输入QQ号")
return return
permission_node = args[1]
# 执行撤销 # 执行撤销
success = permission_api.revoke_permission(chat_stream.platform, user_id, permission_node) success = permission_api.revoke_permission(chat_stream.platform, user_id, permission_node)
if success: if success:
await self.send_text(f"✅ 已撤销用户 {user_id} 权限节点 {permission_node}") await self.send_text(f"✅ 已撤销用户 {user_id} 权限节点 `{permission_node}`")
else: else:
await self.send_text("❌ 撤销失败，请检查权限节点是否存在") await self.send_text("❌ 撤销失败，请检查权限节点是否存在")
@require_permission("plugin.permission.view", "❌ 你没有查看权限的权限")
async def _list_permissions(self, chat_stream, args: List[str]): async def _list_permissions(self, chat_stream, args: List[str]):
"""列出用户权限""" """列出用户权限"""
target_user_id = None target_user_id = None
if args: if args:
# 指定了用户 # 指定了用户 - 使用新的解析方法
user_mention = args[0] target_user_id = self._parse_user_mention(args[0])
target_user_id = self._parse_user_mention(user_mention)
if not target_user_id: if not target_user_id:
await self.send_text("❌ 无效的用户格式，请使用 @用户 或直接输入QQ号") await self.send_text("❌ 无效的用户格式，请使用 @<用户名:QQ号> 或直接输入QQ号")
return return
else: else:
# 查看自己的权限 # 查看自己的权限
@@ -229,45 +237,46 @@ class PermissionCommand(PlusCommand):
permissions = permission_api.get_user_permissions(chat_stream.platform, target_user_id) permissions = permission_api.get_user_permissions(chat_stream.platform, target_user_id)
if is_master: if is_master:
response = f"👑 用户 {target_user_id} 是Master用户，拥有所有权限" response = f"👑 用户 `{target_user_id}` 是Master用户，拥有所有权限"
else: else:
if permissions: if permissions:
perm_list = "\n".join([f"• {perm}" for perm in permissions]) perm_list = "\n".join([f"• `{perm}`" for perm in permissions])
response = f"📋 用户 {target_user_id} 拥有的权限：\n{perm_list}" response = f"📋 用户 `{target_user_id}` 拥有的权限：\n{perm_list}"
else: else:
response = f"📋 用户 {target_user_id} 没有任何权限" response = f"📋 用户 `{target_user_id}` 没有任何权限"
await self.send_text(response) await self.send_text(response)
@require_permission("plugin.permission.view", "❌ 你没有查看权限的权限")
async def _check_permission(self, chat_stream, args: List[str]): async def _check_permission(self, chat_stream, args: List[str]):
"""检查用户权限""" """检查用户权限"""
if len(args) < 2: if len(args) < 2:
await self.send_text("❌ 用法: /permission check <@用户|QQ号> <权限节点>") await self.send_text("❌ 用法: /permission check <@用户|QQ号> <权限节点>")
return return
user_mention = args[0] # 解析用户ID - 使用新的解析方法
permission_node = args[1] user_id = self._parse_user_mention(args[0])
# 解析用户ID
user_id = self._parse_user_mention(user_mention)
if not user_id: if not user_id:
await self.send_text("❌ 无效的用户格式，请使用 @用户 或直接输入QQ号") await self.send_text("❌ 无效的用户格式，请使用 @<用户名:QQ号> 或直接输入QQ号")
return return
permission_node = args[1]
# 检查权限 # 检查权限
has_permission = permission_api.check_permission(chat_stream.platform, user_id, permission_node) has_permission = permission_api.check_permission(chat_stream.platform, user_id, permission_node)
is_master = permission_api.is_master(chat_stream.platform, user_id) is_master = permission_api.is_master(chat_stream.platform, user_id)
if has_permission: if has_permission:
if is_master: if is_master:
response = f"✅ 用户 {user_id} 拥有权限 {permission_node}（Master用户）" response = f"✅ 用户 `{user_id}` 拥有权限 `{permission_node}`（Master用户）"
else: else:
response = f"✅ 用户 {user_id} 拥有权限 {permission_node}" response = f"✅ 用户 `{user_id}` 拥有权限 `{permission_node}`"
else: else:
response = f"❌ 用户 {user_id} 没有权限 {permission_node}" response = f"❌ 用户 `{user_id}` 没有权限 `{permission_node}`"
await self.send_text(response) await self.send_text(response)
@require_permission("plugin.permission.view", "❌ 你没有查看权限的权限")
async def _list_nodes(self, chat_stream, args: List[str]): async def _list_nodes(self, chat_stream, args: List[str]):
"""列出权限节点""" """列出权限节点"""
plugin_name = args[0] if args else None plugin_name = args[0] if args else None
@@ -300,6 +309,7 @@ class PermissionCommand(PlusCommand):
await self.send_text(response) await self.send_text(response)
@require_permission("plugin.permission.view", "❌ 你没有查看权限的权限")
async def _list_all_nodes_with_description(self, chat_stream): async def _list_all_nodes_with_description(self, chat_stream):
"""列出所有插件的权限节点（带详细描述）""" """列出所有插件的权限节点（带详细描述）"""
# 获取所有权限节点 # 获取所有权限节点