285 lines
10 KiB
Python
285 lines
10 KiB
Python
from typing import List, Tuple, Type, Any, Optional
|
||
from src.plugin_system import (
|
||
BasePlugin,
|
||
register_plugin,
|
||
BaseAction,
|
||
BaseCommand,
|
||
BaseTool,
|
||
ComponentInfo,
|
||
ActionActivationType,
|
||
ConfigField,
|
||
ToolParamType
|
||
)
|
||
|
||
|
||
from src.plugin_system.base.base_command import BaseCommand
|
||
from src.plugin_system.apis import send_api
|
||
from src.common.logger import get_logger
|
||
from src.plugin_system.base.component_types import ChatType
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
|
||
class GetGroupListCommand(BaseCommand):
|
||
"""获取群列表命令"""
|
||
|
||
command_name = "get_groups"
|
||
command_description = "获取机器人加入的群列表"
|
||
command_pattern = r"^/get_groups$"
|
||
command_help = "获取机器人加入的群列表"
|
||
command_examples = ["/get_groups"]
|
||
intercept_message = True
|
||
|
||
|
||
|
||
|
||
|
||
async def execute(self) -> Tuple[bool, str, bool]:
|
||
try:
|
||
# 调用适配器命令API
|
||
domain = "user.qzone.qq.com"
|
||
response = await send_api.adapter_command_to_stream(
|
||
action="get_cookies",
|
||
platform="qq",
|
||
params={"domain": domain},
|
||
timeout=40.0,
|
||
storage_message=False
|
||
)
|
||
text = str(response)
|
||
await self.send_text(text)
|
||
return True, "获取群列表成功", True
|
||
|
||
except Exception as e:
|
||
await self.send_text(f"获取群列表失败: {str(e)}")
|
||
return False, "获取群列表失败", True
|
||
|
||
class CompareNumbersTool(BaseTool):
|
||
"""比较两个数大小的工具"""
|
||
|
||
name = "compare_numbers"
|
||
description = "使用工具 比较两个数的大小，返回较大的数"
|
||
parameters = [
|
||
("num1", ToolParamType.FLOAT, "第一个数字", True, None),
|
||
("num2", ToolParamType.FLOAT, "第二个数字", True, None),
|
||
]
|
||
|
||
async def execute(self, function_args: dict[str, Any]) -> dict[str, Any]:
|
||
"""执行比较两个数的大小
|
||
|
||
Args:
|
||
function_args: 工具参数
|
||
|
||
Returns:
|
||
dict: 工具执行结果
|
||
"""
|
||
num1: int | float = function_args.get("num1") # type: ignore
|
||
num2: int | float = function_args.get("num2") # type: ignore
|
||
|
||
try:
|
||
if num1 > num2:
|
||
result = f"{num1} 大于 {num2}"
|
||
elif num1 < num2:
|
||
result = f"{num1} 小于 {num2}"
|
||
else:
|
||
result = f"{num1} 等于 {num2}"
|
||
|
||
return {"name": self.name, "content": result}
|
||
except Exception as e:
|
||
return {"name": self.name, "content": f"比较数字失败，炸了: {str(e)}"}
|
||
|
||
|
||
# ===== Action组件 =====
|
||
class HelloAction(BaseAction):
|
||
"""问候Action - 简单的问候动作"""
|
||
|
||
# === 基本信息（必须填写）===
|
||
action_name = "hello_greeting"
|
||
action_description = "向用户发送问候消息"
|
||
activation_type = ActionActivationType.ALWAYS # 始终激活
|
||
|
||
# === 功能描述（必须填写）===
|
||
action_parameters = {"greeting_message": "要发送的问候消息"}
|
||
action_require = ["需要发送友好问候时使用", "当有人向你问好时使用", "当你遇见没有见过的人时使用"]
|
||
associated_types = ["text"]
|
||
|
||
async def execute(self) -> Tuple[bool, str]:
|
||
"""执行问候动作 - 这是核心功能"""
|
||
# 发送问候消息
|
||
greeting_message = self.action_data.get("greeting_message", "")
|
||
base_message = self.get_config("greeting.message", "嗨！很开心见到你！😊")
|
||
message = base_message + greeting_message
|
||
await self.send_text(message)
|
||
|
||
return True, "发送了问候消息"
|
||
|
||
|
||
class ByeAction(BaseAction):
|
||
"""告别Action - 只在用户说再见时激活"""
|
||
|
||
action_name = "bye_greeting"
|
||
action_description = "向用户发送告别消息"
|
||
|
||
# 使用关键词激活
|
||
activation_type = ActionActivationType.KEYWORD
|
||
|
||
# 关键词设置
|
||
activation_keywords = ["再见", "bye", "88", "拜拜"]
|
||
keyword_case_sensitive = False
|
||
|
||
action_parameters = {"bye_message": "要发送的告别消息"}
|
||
action_require = [
|
||
"用户要告别时使用",
|
||
"当有人要离开时使用",
|
||
"当有人和你说再见时使用",
|
||
]
|
||
associated_types = ["text"]
|
||
|
||
async def execute(self) -> Tuple[bool, str]:
|
||
bye_message = self.action_data.get("bye_message", "")
|
||
|
||
message = f"再见！期待下次聊天！👋{bye_message}"
|
||
await self.send_text(message)
|
||
return True, "发送了告别消息"
|
||
|
||
|
||
class TimeCommand(BaseCommand):
|
||
"""时间查询Command - 响应/time命令"""
|
||
|
||
command_name = "time"
|
||
command_description = "获取当前时间"
|
||
|
||
# === 命令设置（必须填写）===
|
||
command_pattern = r"^/time$" # 精确匹配 "/time" 命令
|
||
chat_type_allow = ChatType.GROUP # 仅在群聊中可用
|
||
|
||
async def execute(self) -> Tuple[bool, str, bool]:
|
||
"""执行时间查询"""
|
||
import datetime
|
||
|
||
# 获取当前时间
|
||
time_format: str = self.get_config("time.format", "%Y-%m-%d %H:%M:%S") # type: ignore
|
||
now = datetime.datetime.now()
|
||
time_str = now.strftime(time_format)
|
||
|
||
# 发送时间信息
|
||
message = f"⏰ 当前时间：{time_str}"
|
||
await self.send_text(message)
|
||
|
||
return True, f"显示了当前x时间: {time_str}", True
|
||
|
||
|
||
# class PrintMessage(BaseEventHandler):
|
||
# """打印消息事件处理器 - 处理打印消息事件"""
|
||
#
|
||
# event_type = EventType.ON_MESSAGE
|
||
# handler_name = "print_message_handler"
|
||
# handler_description = "打印接收到的消息"
|
||
#
|
||
# async def execute(self, message: MaiMessages) -> Tuple[bool, bool, str | None]:
|
||
# """执行打印消息事件处理"""
|
||
# # 打印接收到的消息
|
||
#
|
||
# if self.get_config("print_message.enabled", False):
|
||
# print(f"接收到消息: {message.raw_message}")
|
||
# return True, True, "消息已打印1"
|
||
|
||
|
||
# ===== 插件注册 =====
|
||
|
||
|
||
@register_plugin
|
||
class HelloWorldPlugin(BasePlugin):
|
||
"""Hello World插件 - 你的第一个MaiCore插件"""
|
||
|
||
# 插件基本信息
|
||
plugin_name: str = "hello_world_plugin" # 内部标识符
|
||
enable_plugin: bool = True
|
||
dependencies: List[str] = [] # 插件依赖列表
|
||
python_dependencies: List[str] = [] # Python包依赖列表
|
||
config_file_name: str = "config.toml" # 配置文件名
|
||
|
||
# 配置节描述
|
||
config_section_descriptions = {"plugin": "插件基本信息", "greeting": "问候功能配置", "time": "时间查询配置"}
|
||
|
||
# 配置Schema定义
|
||
config_schema: dict = {
|
||
"plugin": {
|
||
"name": ConfigField(type=str, default="hello_world_plugin", description="插件名称"),
|
||
"version": ConfigField(type=str, default="1.0.0", description="插件版本"),
|
||
"enabled": ConfigField(type=bool, default=False, description="是否启用插件"),
|
||
},
|
||
"greeting": {
|
||
"message": ConfigField(
|
||
type=list, default=["嗨！很开心见到你！😊", "Ciallo～(∠・ω< )⌒★"], description="默认问候消息"
|
||
),
|
||
"enable_emoji": ConfigField(type=bool, default=True, description="是否启用表情符号"),
|
||
},
|
||
"time": {"format": ConfigField(type=str, default="%Y-%m-%d %H:%M:%S", description="时间显示格式")},
|
||
"print_message": {"enabled": ConfigField(type=bool, default=True, description="是否启用打印")},
|
||
}
|
||
|
||
def get_plugin_components(self) -> List[Tuple[ComponentInfo, Type]]:
|
||
return [
|
||
(HelloAction.get_action_info(), HelloAction),
|
||
(CompareNumbersTool.get_tool_info(), CompareNumbersTool), # 添加比较数字工具
|
||
(ByeAction.get_action_info(), ByeAction), # 添加告别Action
|
||
(TimeCommand.get_command_info(), TimeCommand), # 现在只能在群聊中使用
|
||
(GetGroupListCommand.get_command_info(), GetGroupListCommand), # 添加获取群列表命令
|
||
(PrivateInfoCommand.get_command_info(), PrivateInfoCommand), # 私聊专用命令
|
||
(GroupOnlyAction.get_action_info(), GroupOnlyAction), # 群聊专用动作
|
||
# (PrintMessage.get_handler_info(), PrintMessage),
|
||
]
|
||
|
||
|
||
# @register_plugin
|
||
# class HelloWorldEventPlugin(BaseEPlugin):
|
||
# """Hello World事件插件 - 处理问候和告别事件"""
|
||
|
||
# plugin_name = "hello_world_event_plugin"
|
||
# enable_plugin = False
|
||
# dependencies = []
|
||
# python_dependencies = []
|
||
# config_file_name = "event_config.toml"
|
||
|
||
# config_schema = {
|
||
# "plugin": {
|
||
# "name": ConfigField(type=str, default="hello_world_event_plugin", description="插件名称"),
|
||
# "version": ConfigField(type=str, default="1.0.0", description="插件版本"),
|
||
# "enabled": ConfigField(type=bool, default=True, description="是否启用插件"),
|
||
# },
|
||
# }
|
||
|
||
# def get_plugin_components(self) -> List[Tuple[ComponentInfo, Type]]:
|
||
# return [(PrintMessage.get_handler_info(), PrintMessage)]
|
||
|
||
# 添加一个新的私聊专用命令
|
||
class PrivateInfoCommand(BaseCommand):
|
||
command_name = "private_info"
|
||
command_description = "获取私聊信息"
|
||
command_pattern = r"^/私聊信息$"
|
||
chat_type_allow = ChatType.PRIVATE # 仅在私聊中可用
|
||
|
||
async def execute(self) -> Tuple[bool, Optional[str], bool]:
|
||
"""执行私聊信息命令"""
|
||
try:
|
||
await self.send_text("这是一个只能在私聊中使用的命令！")
|
||
return True, "私聊信息命令执行成功", False
|
||
except Exception as e:
|
||
logger.error(f"私聊信息命令执行失败: {e}")
|
||
return False, f"命令执行失败: {e}", False
|
||
|
||
# 添加一个新的仅群聊可用的Action
|
||
class GroupOnlyAction(BaseAction):
|
||
action_name = "group_only_test"
|
||
action_description = "群聊专用测试动作"
|
||
chat_type_allow = ChatType.GROUP # 仅在群聊中可用
|
||
|
||
async def execute(self) -> Tuple[bool, str]:
|
||
"""执行群聊专用测试动作"""
|
||
try:
|
||
await self.send_text("这是一个只能在群聊中执行的动作！")
|
||
return True, "群聊专用动作执行成功"
|
||
except Exception as e:
|
||
logger.error(f"群聊专用动作执行失败: {e}")
|
||
return False, f"动作执行失败: {e}"
|