mirror of https://github.com/Mai-with-u/MaiBot.git
Add files via upload
A0000Xz
GitHub
parent
ed1724b29c
commit
d1de43b09a
|
|
@ -1,230 +1,231 @@
|
||||||
from abc import ABC, abstractmethod
|
from abc import ABC, abstractmethod
|
||||||
from typing import Dict, Tuple, Optional, List
|
from typing import Dict, Tuple, Optional, List
|
||||||
from src.common.logger import get_logger
|
from src.common.logger import get_logger
|
||||||
from src.plugin_system.base.component_types import CommandInfo, ComponentType
|
from src.plugin_system.base.component_types import CommandInfo, ComponentType
|
||||||
from src.chat.message_receive.message import MessageRecv
|
from src.chat.message_receive.message import MessageRecv
|
||||||
from src.plugin_system.apis import send_api
|
from src.plugin_system.apis import send_api
|
||||||
|
|
||||||
logger = get_logger("base_command")
|
logger = get_logger("base_command")
|
||||||
|
|
||||||
|
|
||||||
class BaseCommand(ABC):
|
class BaseCommand(ABC):
|
||||||
"""Command组件基类
|
"""Command组件基类
|
||||||
|
|
||||||
Command是插件的一种组件类型，用于处理命令请求
|
Command是插件的一种组件类型，用于处理命令请求
|
||||||
|
|
||||||
子类可以通过类属性定义命令模式：
|
子类可以通过类属性定义命令模式：
|
||||||
- command_pattern: 命令匹配的正则表达式
|
- command_pattern: 命令匹配的正则表达式
|
||||||
- command_help: 命令帮助信息
|
- command_help: 命令帮助信息
|
||||||
- command_examples: 命令使用示例列表
|
- command_examples: 命令使用示例列表
|
||||||
- intercept_message: 是否拦截消息处理（默认True拦截，False继续传递）
|
- intercept_message: 是否拦截消息处理（默认True拦截，False继续传递）
|
||||||
"""
|
"""
|
||||||
|
|
||||||
command_name: str = ""
|
command_name: str = ""
|
||||||
command_description: str = ""
|
command_description: str = ""
|
||||||
|
|
||||||
# 默认命令设置（子类可以覆盖）
|
# 默认命令设置（子类可以覆盖）
|
||||||
command_pattern: str = ""
|
command_pattern: str = ""
|
||||||
command_help: str = ""
|
command_help: str = ""
|
||||||
command_examples: List[str] = []
|
command_examples: List[str] = []
|
||||||
intercept_message: bool = True # 默认拦截消息，不继续处理
|
intercept_message: bool = True # 默认拦截消息，不继续处理
|
||||||
|
|
||||||
def __init__(self, message: MessageRecv, plugin_config: dict = None):
|
def __init__(self, message: MessageRecv, plugin_config: dict = None):
|
||||||
"""初始化Command组件
|
"""初始化Command组件
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
message: 接收到的消息对象
|
message: 接收到的消息对象
|
||||||
plugin_config: 插件配置字典
|
plugin_config: 插件配置字典
|
||||||
"""
|
"""
|
||||||
self.message = message
|
self.message = message
|
||||||
self.matched_groups: Dict[str, str] = {} # 存储正则表达式匹配的命名组
|
self.matched_groups: Dict[str, str] = {} # 存储正则表达式匹配的命名组
|
||||||
self.plugin_config = plugin_config or {} # 直接存储插件配置字典
|
self.plugin_config = plugin_config or {} # 直接存储插件配置字典
|
||||||
|
|
||||||
self.log_prefix = "[Command]"
|
self.log_prefix = "[Command]"
|
||||||
|
|
||||||
logger.debug(f"{self.log_prefix} Command组件初始化完成")
|
logger.debug(f"{self.log_prefix} Command组件初始化完成")
|
||||||
|
|
||||||
def set_matched_groups(self, groups: Dict[str, str]) -> None:
|
def set_matched_groups(self, groups: Dict[str, str]) -> None:
|
||||||
"""设置正则表达式匹配的命名组
|
"""设置正则表达式匹配的命名组
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
groups: 正则表达式匹配的命名组
|
groups: 正则表达式匹配的命名组
|
||||||
"""
|
"""
|
||||||
self.matched_groups = groups
|
self.matched_groups = groups
|
||||||
|
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
async def execute(self) -> Tuple[bool, Optional[str]]:
|
async def execute(self) -> Tuple[bool, Optional[str]]:
|
||||||
"""执行Command的抽象方法，子类必须实现
|
"""执行Command的抽象方法，子类必须实现
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Tuple[bool, Optional[str]]: (是否执行成功, 可选的回复消息)
|
Tuple[bool, Optional[str]]: (是否执行成功, 可选的回复消息)
|
||||||
"""
|
"""
|
||||||
pass
|
pass
|
||||||
|
|
||||||
def get_config(self, key: str, default=None):
|
def get_config(self, key: str, default=None):
|
||||||
"""获取插件配置值，支持嵌套键访问
|
"""获取插件配置值，支持嵌套键访问
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
key: 配置键名，支持嵌套访问如 "section.subsection.key"
|
key: 配置键名，支持嵌套访问如 "section.subsection.key"
|
||||||
default: 默认值
|
default: 默认值
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Any: 配置值或默认值
|
Any: 配置值或默认值
|
||||||
"""
|
"""
|
||||||
if not self.plugin_config:
|
if not self.plugin_config:
|
||||||
return default
|
return default
|
||||||
|
|
||||||
# 支持嵌套键访问
|
# 支持嵌套键访问
|
||||||
keys = key.split(".")
|
keys = key.split(".")
|
||||||
current = self.plugin_config
|
current = self.plugin_config
|
||||||
|
|
||||||
for k in keys:
|
for k in keys:
|
||||||
if isinstance(current, dict) and k in current:
|
if isinstance(current, dict) and k in current:
|
||||||
current = current[k]
|
current = current[k]
|
||||||
else:
|
else:
|
||||||
return default
|
return default
|
||||||
|
|
||||||
return current
|
return current
|
||||||
|
|
||||||
async def send_text(self, content: str, reply_to: str = "") -> bool:
|
async def send_text(self, content: str, reply_to: str = "") -> bool:
|
||||||
"""发送回复消息
|
"""发送回复消息
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
content: 回复内容
|
content: 回复内容
|
||||||
reply_to: 回复消息，格式为"发送者:消息内容"
|
reply_to: 回复消息，格式为"发送者:消息内容"
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
bool: 是否发送成功
|
bool: 是否发送成功
|
||||||
"""
|
"""
|
||||||
# 获取聊天流信息
|
# 获取聊天流信息
|
||||||
chat_stream = self.message.chat_stream
|
chat_stream = self.message.chat_stream
|
||||||
if not chat_stream or not hasattr(chat_stream, "stream_id"):
|
if not chat_stream or not hasattr(chat_stream, "stream_id"):
|
||||||
logger.error(f"{self.log_prefix} 缺少聊天流或stream_id")
|
logger.error(f"{self.log_prefix} 缺少聊天流或stream_id")
|
||||||
return False
|
return False
|
||||||
|
|
||||||
return await send_api.text_to_stream(text=content, stream_id=chat_stream.stream_id, reply_to=reply_to)
|
return await send_api.text_to_stream(text=content, stream_id=chat_stream.stream_id, reply_to=reply_to)
|
||||||
|
|
||||||
async def send_type(
|
async def send_type(
|
||||||
self, message_type: str, content: str, display_message: str = "", typing: bool = False, reply_to: str = ""
|
self, message_type: str, content: str, display_message: str = "", typing: bool = False, reply_to: str = ""
|
||||||
) -> bool:
|
) -> bool:
|
||||||
"""发送指定类型的回复消息到当前聊天环境
|
"""发送指定类型的回复消息到当前聊天环境
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
message_type: 消息类型，如"text"、"image"、"emoji"等
|
message_type: 消息类型，如"text"、"image"、"emoji"等
|
||||||
content: 消息内容
|
content: 消息内容
|
||||||
display_message: 显示消息（可选）
|
display_message: 显示消息（可选）
|
||||||
typing: 是否显示正在输入
|
typing: 是否显示正在输入
|
||||||
reply_to: 回复消息，格式为"发送者:消息内容"
|
reply_to: 回复消息，格式为"发送者:消息内容"
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
bool: 是否发送成功
|
bool: 是否发送成功
|
||||||
"""
|
"""
|
||||||
# 获取聊天流信息
|
# 获取聊天流信息
|
||||||
chat_stream = self.message.chat_stream
|
chat_stream = self.message.chat_stream
|
||||||
if not chat_stream or not hasattr(chat_stream, "stream_id"):
|
if not chat_stream or not hasattr(chat_stream, "stream_id"):
|
||||||
logger.error(f"{self.log_prefix} 缺少聊天流或stream_id")
|
logger.error(f"{self.log_prefix} 缺少聊天流或stream_id")
|
||||||
return False
|
return False
|
||||||
|
|
||||||
return await send_api.custom_to_stream(
|
return await send_api.custom_to_stream(
|
||||||
message_type=message_type,
|
message_type=message_type,
|
||||||
content=content,
|
content=content,
|
||||||
stream_id=chat_stream.stream_id,
|
stream_id=chat_stream.stream_id,
|
||||||
display_message=display_message,
|
display_message=display_message,
|
||||||
typing=typing,
|
typing=typing,
|
||||||
reply_to=reply_to,
|
reply_to=reply_to,
|
||||||
)
|
)
|
||||||
|
|
||||||
async def send_command(
|
async def send_command(
|
||||||
self, command_name: str, args: dict = None, display_message: str = "", storage_message: bool = True
|
self, command_name: str, args: dict = None, display_message: str = "", storage_message: bool = True
|
||||||
) -> bool:
|
) -> bool:
|
||||||
"""发送命令消息
|
"""发送命令消息
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
command_name: 命令名称
|
command_name: 命令名称
|
||||||
args: 命令参数
|
args: 命令参数
|
||||||
display_message: 显示消息
|
display_message: 显示消息
|
||||||
storage_message: 是否存储消息到数据库
|
storage_message: 是否存储消息到数据库
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
bool: 是否发送成功
|
bool: 是否发送成功
|
||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
# 获取聊天流信息
|
# 获取聊天流信息
|
||||||
chat_stream = self.message.chat_stream
|
chat_stream = self.message.chat_stream
|
||||||
if not chat_stream or not hasattr(chat_stream, "stream_id"):
|
if not chat_stream or not hasattr(chat_stream, "stream_id"):
|
||||||
logger.error(f"{self.log_prefix} 缺少聊天流或stream_id")
|
logger.error(f"{self.log_prefix} 缺少聊天流或stream_id")
|
||||||
return False
|
return False
|
||||||
|
|
||||||
# 构造命令数据
|
# 构造命令数据
|
||||||
command_data = {"name": command_name, "args": args or {}}
|
command_data = {"name": command_name, "args": args or {}}
|
||||||
|
|
||||||
success = await send_api.command_to_stream(
|
success = await send_api.command_to_stream(
|
||||||
command=command_data,
|
command=command_data,
|
||||||
stream_id=chat_stream.stream_id,
|
stream_id=chat_stream.stream_id,
|
||||||
storage_message=storage_message,
|
storage_message=storage_message,
|
||||||
)
|
display_message=display_message
|
||||||
|
)
|
||||||
if success:
|
|
||||||
logger.info(f"{self.log_prefix} 成功发送命令: {command_name}")
|
if success:
|
||||||
else:
|
logger.info(f"{self.log_prefix} 成功发送命令: {command_name}")
|
||||||
logger.error(f"{self.log_prefix} 发送命令失败: {command_name}")
|
else:
|
||||||
|
logger.error(f"{self.log_prefix} 发送命令失败: {command_name}")
|
||||||
return success
|
|
||||||
|
return success
|
||||||
except Exception as e:
|
|
||||||
logger.error(f"{self.log_prefix} 发送命令时出错: {e}")
|
except Exception as e:
|
||||||
return False
|
logger.error(f"{self.log_prefix} 发送命令时出错: {e}")
|
||||||
|
return False
|
||||||
async def send_emoji(self, emoji_base64: str) -> bool:
|
|
||||||
"""发送表情包
|
async def send_emoji(self, emoji_base64: str) -> bool:
|
||||||
|
"""发送表情包
|
||||||
Args:
|
|
||||||
emoji_base64: 表情包的base64编码
|
Args:
|
||||||
|
emoji_base64: 表情包的base64编码
|
||||||
Returns:
|
|
||||||
bool: 是否发送成功
|
Returns:
|
||||||
"""
|
bool: 是否发送成功
|
||||||
chat_stream = self.message.chat_stream
|
"""
|
||||||
if not chat_stream or not hasattr(chat_stream, "stream_id"):
|
chat_stream = self.message.chat_stream
|
||||||
logger.error(f"{self.log_prefix} 缺少聊天流或stream_id")
|
if not chat_stream or not hasattr(chat_stream, "stream_id"):
|
||||||
return False
|
logger.error(f"{self.log_prefix} 缺少聊天流或stream_id")
|
||||||
|
return False
|
||||||
return await send_api.emoji_to_stream(emoji_base64, chat_stream.stream_id)
|
|
||||||
|
return await send_api.emoji_to_stream(emoji_base64, chat_stream.stream_id)
|
||||||
async def send_image(self, image_base64: str) -> bool:
|
|
||||||
"""发送图片
|
async def send_image(self, image_base64: str) -> bool:
|
||||||
|
"""发送图片
|
||||||
Args:
|
|
||||||
image_base64: 图片的base64编码
|
Args:
|
||||||
|
image_base64: 图片的base64编码
|
||||||
Returns:
|
|
||||||
bool: 是否发送成功
|
Returns:
|
||||||
"""
|
bool: 是否发送成功
|
||||||
chat_stream = self.message.chat_stream
|
"""
|
||||||
if not chat_stream or not hasattr(chat_stream, "stream_id"):
|
chat_stream = self.message.chat_stream
|
||||||
logger.error(f"{self.log_prefix} 缺少聊天流或stream_id")
|
if not chat_stream or not hasattr(chat_stream, "stream_id"):
|
||||||
return False
|
logger.error(f"{self.log_prefix} 缺少聊天流或stream_id")
|
||||||
|
return False
|
||||||
return await send_api.image_to_stream(image_base64, chat_stream.stream_id)
|
|
||||||
|
return await send_api.image_to_stream(image_base64, chat_stream.stream_id)
|
||||||
@classmethod
|
|
||||||
def get_command_info(cls) -> "CommandInfo":
|
@classmethod
|
||||||
"""从类属性生成CommandInfo
|
def get_command_info(cls) -> "CommandInfo":
|
||||||
|
"""从类属性生成CommandInfo
|
||||||
Args:
|
|
||||||
name: Command名称，如果不提供则使用类名
|
Args:
|
||||||
description: Command描述，如果不提供则使用类文档字符串
|
name: Command名称，如果不提供则使用类名
|
||||||
|
description: Command描述，如果不提供则使用类文档字符串
|
||||||
Returns:
|
|
||||||
CommandInfo: 生成的Command信息对象
|
Returns:
|
||||||
"""
|
CommandInfo: 生成的Command信息对象
|
||||||
|
"""
|
||||||
return CommandInfo(
|
|
||||||
name=cls.command_name,
|
return CommandInfo(
|
||||||
component_type=ComponentType.COMMAND,
|
name=cls.command_name,
|
||||||
description=cls.command_description,
|
component_type=ComponentType.COMMAND,
|
||||||
command_pattern=cls.command_pattern,
|
description=cls.command_description,
|
||||||
command_help=cls.command_help,
|
command_pattern=cls.command_pattern,
|
||||||
command_examples=cls.command_examples.copy() if cls.command_examples else [],
|
command_help=cls.command_help,
|
||||||
intercept_message=cls.intercept_message,
|
command_examples=cls.command_examples.copy() if cls.command_examples else [],
|
||||||
)
|
intercept_message=cls.intercept_message,
|
||||||
|
)
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue