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

refactor(plugins): 将多个社交互动插件整合到 social_toolkit_plugin

Browse Source 
将 `set_emoji_like`, `at_user`, `poke`, `reminder`, 和 `set_typing_status` 等多个独立的内置插件的功能进行重构和整合，统一归入一个新的 `social_toolkit_plugin` 插件中。

这次重构旨在：
- 减少插件数量，简化插件管理和维护。
- 整合相似的功能（如用户互动、提醒等），提高代码复用性。
- 为未来添加更多社交互动功能提供一个统一的框架。
This commit is contained in:
minecraft1024a
2025-09-30 21:50:19 +08:00
parent 53c7b1e6fe
commit 88063f6e1b
13 changed files with 281 additions and 893 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/plugins/built_in/affinity_flow_chatter/plan_filter.py
View File

@@ -644,7 +644,7 @@ class ChatterPlanFilter:
# 为参数描述添加一个通用示例值
if action_name == "set_emoji_like" and p_name == "emoji":
# 特殊处理set_emoji_like的emoji参数
from plugins.set_emoji_like.qq_emoji_list import qq_face
from plugins.social_toolkit_plugin.qq_emoji_list import qq_face
emoji_options = [re.search(r"\[表情：(.+?)\]", name).group(1) for name in qq_face.values() if re.search(r"\[表情：(.+?)\]", name)]
example_value = f"<从'{', '.join(emoji_options[:10])}...'中选择一个>"
else:

11
src/plugins/built_in/at_user_plugin/_manifest.json
View File

@@ -1,11 +0,0 @@
{
"manifest_version": 1,
"name": "At User Plugin",
"description": "一个通过名字艾特用户的插件",
"author": "Kilo Code",
"version": "1.0.0",
"requirements": [],
"license": "MIT",
"keywords": ["at", "mention", "user"],
"categories": ["Chat", "Utility"]
}

173
src/plugins/built_in/at_user_plugin/plugin.py
View File

@@ -1,173 +0,0 @@
from typing import List, Tuple, Type
from src.plugin_system import (
BasePlugin,
BaseCommand,
CommandInfo,
register_plugin,
BaseAction,
ActionInfo,
ActionActivationType,
)
from src.person_info.person_info import get_person_info_manager
from src.common.logger import get_logger
from src.plugin_system.base.component_types import ChatType
logger = get_logger(__name__)
class AtAction(BaseAction):
"""发送艾特消息"""
# === 基本信息（必须填写）===
action_name = "at_user"
action_description = "发送艾特消息"
activation_type = ActionActivationType.LLM_JUDGE # 消息接收时激活(?)
parallel_action = False
chat_type_allow = ChatType.GROUP
# === 功能描述（必须填写）===
action_parameters = {"user_name": "需要艾特用户的名字", "at_message": "艾特用户时要发送的消息"}
action_require = [
"当用户明确要求你去'叫'、'喊'、'提醒'或'艾特'某人时使用",
"当你判断，为了让特定的人看到消息，需要代表用户去呼叫他/她时使用",
"例如：'你去叫一下张三'，'提醒一下李四开会'",
]
llm_judge_prompt = """
判定是否需要使用艾特用户动作的条件：
1. 你在对话中提到了某个具体的人，并且需要提醒他/她。
3. 上下文明确需要你艾特一个或多个人。
请回答"是"或"否"。
"""
associated_types = ["text"]
async def execute(self) -> Tuple[bool, str]:
"""执行艾特用户的动作"""
user_name = self.action_data.get("user_name")
at_message = self.action_data.get("at_message")
if not user_name or not at_message:
logger.warning("艾特用户的动作缺少必要参数。")
await self.store_action_info(
action_build_into_prompt=True,
action_prompt_display=f"执行了艾特用户动作：艾特用户 {user_name} 并发送消息: {at_message},失败了,因为没有提供必要参数",
action_done=False,
)
return False, "缺少必要参数"
user_info = await get_person_info_manager().get_person_info_by_name(user_name)
if not user_info or not user_info.get("user_id"):
logger.info(f"找不到名为 '{user_name}' 的用户。")
return False, "用户不存在"
try:
# 使用回复器生成艾特回复，而不是直接发送命令
from src.chat.replyer.default_generator import DefaultReplyer
from src.chat.message_receive.chat_stream import get_chat_manager
# 获取当前聊天流
chat_manager = get_chat_manager()
chat_stream = self.chat_stream or chat_manager.get_stream(self.chat_id)
if not chat_stream:
logger.error(f"找不到聊天流: {self.chat_stream}")
return False, "聊天流不存在"
# 创建回复器实例
replyer = DefaultReplyer(chat_stream)
# 构建回复对象，将艾特消息作为回复目标
reply_to = f"{user_name}:{at_message}"
extra_info = f"你需要艾特用户 {user_name} 并回复他们说: {at_message}"
# 使用回复器生成回复
success, llm_response, prompt = await replyer.generate_reply_with_context(
reply_to=reply_to,
extra_info=extra_info,
enable_tool=False, # 艾特回复通常不需要工具调用
from_plugin=False,
)
if success and llm_response:
# 获取生成的回复内容
reply_content = llm_response.get("content", "")
if reply_content:
# 获取用户QQ号，发送真正的艾特消息
user_id = user_info.get("user_id")
# 发送真正的艾特命令，使用回复器生成的智能内容
await self.send_command(
"SEND_AT_MESSAGE",
args={"qq_id": user_id, "text": reply_content},
display_message=f"艾特用户 {user_name} 并发送智能回复: {reply_content}",
)
await self.store_action_info(
action_build_into_prompt=True,
action_prompt_display=f"执行了艾特用户动作：艾特用户 {user_name} 并发送智能回复: {reply_content}",
action_done=True,
)
logger.info(f"成功通过回复器生成智能内容并发送真正的艾特消息给 {user_name}: {reply_content}")
return True, "智能艾特消息发送成功"
else:
logger.warning("回复器生成了空内容")
return False, "回复内容为空"
else:
logger.error("回复器生成回复失败")
return False, "回复生成失败"
except Exception as e:
logger.error(f"执行艾特用户动作时发生异常: {e}", exc_info=True)
await self.store_action_info(
action_build_into_prompt=True,
action_prompt_display=f"执行艾特用户动作失败：{str(e)}",
action_done=False,
)
return False, f"执行失败: {str(e)}"
class AtCommand(BaseCommand):
command_name: str = "at_user"
description: str = "通过名字艾特用户"
command_pattern: str = r"/at\s+@?(?P<name>[\S]+)(?:\s+(?P<text>.*))?"
async def execute(self) -> Tuple[bool, str, bool]:
name = self.matched_groups.get("name")
text = self.matched_groups.get("text", "")
if not name:
await self.send_text("请指定要艾特的用户名称。")
return False, "缺少用户名称", True
person_info_manager = get_person_info_manager()
user_info = await person_info_manager.get_person_info_by_name(name)
if not user_info or not user_info.get("user_id"):
await self.send_text(f"找不到名为 '{name}' 的用户。")
return False, "用户不存在", True
user_id = user_info.get("user_id")
await self.send_command(
"SEND_AT_MESSAGE",
args={"qq_id": user_id, "text": text},
display_message=f"艾特用户 {name} 并发送消息: {text}",
)
return True, "艾特消息已发送", True
@register_plugin
class AtUserPlugin(BasePlugin):
plugin_name: str = "at_user_plugin"
enable_plugin: bool = True
dependencies: list[str] = []
python_dependencies: list[str] = []
config_file_name: str = "config.toml"
config_schema: dict = {}
def get_plugin_components(self) -> List[Tuple[CommandInfo | ActionInfo, Type[BaseCommand] | Type[BaseAction]]]:
return [
(AtAction.get_action_info(), AtAction),
]

83
src/plugins/built_in/poke_plugin/_manifest.json
View File

@@ -1,83 +0,0 @@
{
"manifest_version": 1,
"name": "戳一戳插件 (Poke Plugin)",
"version": "1.0.0",
"description": "智能戳一戳互动插件，支持主动戳用户和自动反戳机制，提供多种反应模式包括AI回复",
"author": {
"name": "MaiBot-Plus开发团队",
"url": "https://github.com/MaiBot-Plus"
},
"license": "GPL-v3.0-or-later",
"host_application": {
"min_version": "0.8.0"
},
"homepage_url": "https://github.com/MoFox-Studio/MoFox_Bot",
"repository_url": "https://github.com/MoFox-Studio/MoFox_Bot",
"keywords": ["poke", "interaction", "fun", "social", "ai-reply", "auto-response"],
"categories": ["Social", "Interactive", "Fun"],
"default_locale": "zh-CN",
"locales_path": "_locales",
"plugin_info": {
"is_built_in": false,
"plugin_type": "interactive",
"components": [
{
"type": "action",
"name": "poke_user",
"description": "向指定用户发送戳一戳动作",
"parameters": {
"user_name": "需要戳一戳的用户名称",
"times": "戳一戳次数（可选，默认为1）"
},
"activation_modes": ["llm_judge", "manual"],
"llm_conditions": [
"用户明确要求使用戳一戳",
"想以有趣的方式提醒或与某人互动",
"上下文明确需要戳一个或多个人"
]
},
{
"type": "command",
"name": "poke_back",
"description": "检测戳一戳消息并自动反戳",
"pattern": "(?P<poker_name>\\S+)\\s*戳了戳\\s*(?P<target_name>\\S+)",
"features": [
"正则表达式匹配戳一戳文本",
"智能识别戳击目标",
"防抖机制避免频繁反戳",
"多种反戳模式选择"
]
}
],
"features": [
"主动戳一戳功能 - 可指定用户和次数",
"智能反戳机制 - 自动检测并回应戳一戳",
"多种反戳模式 - 戳回去/AI回复/随机选择",
"AI智能回复 - 集成回复生成器API",
"冷却时间控制 - 防止频繁反戳",
"灵活配置选项 - 支持自定义回复消息",
"安全过滤机制 - 只对针对机器人的戳一戳反应"
],
"configuration": {
"poke_back_mode": {
"type": "string",
"options": ["poke", "reply", "random"],
"default": "poke",
"description": "反戳模式选择"
},
"poke_back_cooldown": {
"type": "integer",
"default": 5,
"description": "反戳冷却时间（秒）"
},
"enable_typo_in_reply": {
"type": "boolean",
"default": false,
"description": "AI回复时是否启用错字生成"
}
}
}
}

307
src/plugins/built_in/poke_plugin/plugin.py
View File

@@ -1,307 +0,0 @@
import asyncio
import random
from typing import List, Tuple, Type
from src.plugin_system import (
BasePlugin,
register_plugin,
BaseAction,
BaseCommand,
ComponentInfo,
ActionActivationType,
ConfigField,
)
from src.common.logger import get_logger
from src.person_info.person_info import get_person_info_manager
from src.plugin_system.apis import generator_api
logger = get_logger("poke_plugin")
# ===== Action组件 =====
class PokeAction(BaseAction):
"""发送戳一戳动作"""
# === 基本信息（必须填写）===
action_name = "poke_user"
action_description = "向用户发送戳一戳"
activation_type = ActionActivationType.ALWAYS
parallel_action = True
# === 功能描述（必须填写）===
action_parameters = {
"user_name": "需要戳一戳的用户的名字 (可选)",
"user_id": "需要戳一戳的用户的ID (可选，优先级更高)",
"times": "需要戳一戳的次数 (默认为 1)",
}
action_require = ["当需要戳某个用户时使用", "当你想提醒特定用户时使用"]
llm_judge_prompt = """
判定是否需要使用戳一戳动作的条件：
1. 用户明确要求使用戳一戳。
2. 你想以一种有趣的方式提醒或与某人互动。
3. 上下文明确需要你戳一个或多个人。
请回答"是"或"否"。
"""
associated_types = ["text"]
async def execute(self) -> Tuple[bool, str]:
"""执行戳一戳的动作"""
user_id = self.action_data.get("user_id")
user_name = self.action_data.get("user_name")
try:
times = int(self.action_data.get("times", 1))
except (ValueError, TypeError):
times = 1
# 优先使用 user_id
if not user_id:
if not user_name:
logger.warning("戳一戳动作缺少 'user_id' 或 'user_name' 参数。")
return False, "缺少用户标识参数"
# 备用方案：通过 user_name 查找
user_info = await get_person_info_manager().get_person_info_by_name(user_name)
if not user_info or not user_info.get("user_id"):
logger.info(f"找不到名为 '{user_name}' 的用户。")
return False, f"找不到名为 '{user_name}' 的用户"
user_id = user_info.get("user_id")
display_name = user_name or user_id
for i in range(times):
logger.info(f"正在向 {display_name} ({user_id}) 发送第 {i + 1}/{times} 次戳一戳...")
await self.send_command(
"SEND_POKE", args={"qq_id": user_id}, display_message=f"戳了戳 {display_name} ({i + 1}/{times})"
)
# 添加一个小的延迟，以避免发送过快
await asyncio.sleep(0.5)
success_message = f"已向 {display_name} 发送 {times} 次戳一戳。"
await self.store_action_info(
action_build_into_prompt=True, action_prompt_display=success_message, action_done=True
)
return True, success_message
# ===== Command组件 =====
class PokeBackCommand(BaseCommand):
"""反戳命令组件"""
command_name = "poke_back"
command_description = "检测到戳一戳时自动反戳回去"
# 匹配戳一戳的正则表达式 - 匹配 "xxx戳了戳xxx" 的格式
command_pattern = r"(?P<poker_name>\S+)\s*戳了戳\s*(?P<target_name>\S+)"
async def execute(self) -> Tuple[bool, str, bool]:
"""执行反戳逻辑"""
# 检查反戳功能是否启用
if not self.get_config("components.command_poke_back", True):
return False, "", False
# 获取匹配的用户名
poker_name = self.matched_groups.get("poker_name", "")
target_name = self.matched_groups.get("target_name", "")
if not poker_name or not target_name:
logger.debug("戳一戳消息格式不匹配，跳过反戳")
return False, "", False
# 只有当目标是机器人自己时才反戳
if target_name not in ["我", "bot", "机器人", "麦麦"]:
logger.debug(f"戳一戳目标不是机器人 ({target_name}), 跳过反戳")
return False, "", False
# 获取戳我的用户信息
poker_info = await get_person_info_manager().get_person_info_by_name(poker_name)
if not poker_info or not poker_info.get("user_id"):
logger.info(f"找不到名为 '{poker_name}' 的用户信息，无法反戳")
return False, "", False
poker_id = poker_info.get("user_id")
if not isinstance(poker_id, (int, str)):
logger.error(f"获取到的用户ID类型不正确: {type(poker_id)}")
return False, "", False
# 确保poker_id是整数类型
try:
poker_id = int(poker_id)
except (ValueError, TypeError):
logger.error(f"无法将用户ID转换为整数: {poker_id}")
return False, "", False
# 检查反戳冷却时间（防止频繁反戳）
cooldown_seconds = self.get_config("components.poke_back_cooldown", 5)
current_time = asyncio.get_event_loop().time()
# 使用类变量存储上次反戳时间
if not hasattr(PokeBackCommand, "_last_poke_back_time"):
PokeBackCommand._last_poke_back_time = {}
last_time = PokeBackCommand._last_poke_back_time.get(poker_id, 0)
if current_time - last_time < cooldown_seconds:
logger.info(f"反戳冷却中，跳过对 {poker_name} 的反戳")
return False, "", False
# 记录本次反戳时间
PokeBackCommand._last_poke_back_time[poker_id] = current_time
# 执行反戳
logger.info(f"检测到 {poker_name} 戳了我，准备反戳回去")
try:
# 获取反戳模式
poke_back_mode = self.get_config("components.poke_back_mode", "poke") # "poke", "reply", "random"
if poke_back_mode == "random":
# 随机选择模式
poke_back_mode = random.choice(["poke", "reply"])
if poke_back_mode == "poke":
# 戳回去模式
await self._poke_back(poker_id, poker_name)
elif poke_back_mode == "reply":
# 回复模式
await self._reply_back(poker_name)
else:
logger.warning(f"未知的反戳模式: {poke_back_mode}")
return False, "", False
logger.info(f"成功反戳了 {poker_name} (模式: {poke_back_mode})")
return True, f"反戳了 {poker_name}", False # 不拦截消息继续处理
except Exception as e:
logger.error(f"反戳失败: {e}")
return False, "", False
async def _poke_back(self, poker_id: int, poker_name: str):
"""执行戳一戳反击"""
await self.send_command(
"SEND_POKE",
args={"qq_id": poker_id},
display_message=f"反戳了 {poker_name}",
storage_message=False, # 不存储到消息历史中
)
# 可选：发送一个随机的反戳回复
poke_back_messages = self.get_config(
"components.poke_back_messages",
[
"哼，戳回去！",
"戳我干嘛~",
"反戳！",
"你戳我，我戳你！",
"（戳回去）",
],
)
if poke_back_messages and self.get_config("components.send_poke_back_message", False):
reply_message = random.choice(poke_back_messages)
await self.send_text(reply_message)
async def _reply_back(self, poker_name: str):
"""生成AI回复"""
# 构造回复上下文
extra_info = f"{poker_name}戳了我一下，需要生成一个有趣的回应。"
# 获取配置，确保类型正确
enable_typo = self.get_config("components.enable_typo_in_reply", False)
if not isinstance(enable_typo, bool):
enable_typo = False
# 使用generator_api生成回复
success, reply_set, _ = await generator_api.generate_reply(
chat_stream=self.message.chat_stream,
extra_info=extra_info,
enable_tool=False,
enable_splitter=True,
enable_chinese_typo=enable_typo,
from_plugin=True,
)
if success and reply_set:
# 发送生成的回复
for reply_item in reply_set:
message_type, content = reply_item
if message_type == "text":
await self.send_text(content)
else:
await self.send_type(message_type, content)
else:
# 如果AI回复失败，发送一个默认回复
fallback_messages = self.get_config(
"components.fallback_reply_messages",
[
"被戳了！",
"诶？",
"做什么呢~",
"怎么了？",
],
)
# 确保fallback_messages是列表
if isinstance(fallback_messages, list) and fallback_messages:
fallback_reply = random.choice(fallback_messages)
await self.send_text(fallback_reply)
else:
await self.send_text("被戳了！")
# ===== 插件注册 =====
@register_plugin
class PokePlugin(BasePlugin):
"""戳一戳插件"""
# 插件基本信息
plugin_name: str = "poke_plugin"
enable_plugin: bool = True
dependencies: List[str] = []
python_dependencies: List[str] = []
config_file_name: str = "config.toml"
# 配置节描述
config_section_descriptions = {"plugin": "插件基本信息", "components": "插件组件"}
# 配置Schema定义
config_schema: dict = {
"plugin": {
"name": ConfigField(type=str, default="poke_plugin", description="插件名称"),
"version": ConfigField(type=str, default="1.0.0", description="插件版本"),
"enabled": ConfigField(type=bool, default=True, description="是否启用插件"),
"config_version": ConfigField(type=str, default="1.0", description="配置版本"),
},
"components": {
"action_poke_user": ConfigField(type=bool, default=True, description="是否启用戳一戳功能"),
"command_poke_back": ConfigField(type=bool, default=True, description="是否启用反戳功能"),
"poke_back_mode": ConfigField(
type=str, default="poke", description="反戳模式: poke(戳回去), reply(AI回复), random(随机)"
),
"poke_back_cooldown": ConfigField(type=int, default=5, description="反戳冷却时间（秒）"),
"send_poke_back_message": ConfigField(type=bool, default=False, description="戳回去时是否发送文字回复"),
"enable_typo_in_reply": ConfigField(type=bool, default=False, description="AI回复时是否启用错字生成"),
"poke_back_messages": ConfigField(
type=list,
default=["哼，戳回去！", "戳我干嘛~", "反戳！", "你戳我，我戳你！", "（戳回去）"],
description="戳回去时的随机回复消息列表",
),
"fallback_reply_messages": ConfigField(
type=list,
default=["被戳了！", "诶？", "做什么呢~", "怎么了？"],
description="AI回复失败时的备用回复消息列表",
),
},
}
def get_plugin_components(self) -> List[Tuple[ComponentInfo, Type]]:
components = []
# 添加戳一戳动作组件
if self.get_config("components.action_poke_user"):
components.append((PokeAction.get_action_info(), PokeAction))
# 添加反戳命令组件
if self.get_config("components.command_poke_back"):
components.append((PokeBackCommand.get_command_info(), PokeBackCommand))
return components

9
src/plugins/built_in/reminder_plugin/_manifest.json
View File

@@ -1,9 +0,0 @@
{
"manifest_version": 1,
"name": "智能提醒插件",
"version": "1.0.0",
"description": "一个能从对话中智能识别并设置定时提醒的插件。",
"author": {
"name": "墨墨"
}
}

18
src/plugins/built_in/set_typing_status/_manifest.json
View File

@@ -1,18 +0,0 @@
{
"manifest_version": 1,
"name": "Set Typing Status",
"description": "一个在LLM生成回复时设置私聊输入状态的插件。",
"version": "1.0.0",
"author": {
"name": "MoFox-Studio"
},
"license": "MIT",
"homepage_url": "",
"repository_url": "",
"keywords": ["typing", "status", "private chat"],
"categories": ["utility"],
"host_application": {
"min_version": "0.10.0"
}
}

63
src/plugins/built_in/set_typing_status/plugin.py
View File

@@ -1,63 +0,0 @@
from typing import List, Tuple, Type
import logging
from src.plugin_system import (
BasePlugin,
register_plugin,
ComponentInfo,
BaseEventHandler,
EventType,
)
from src.plugin_system.base.base_event import HandlerResult
from src.plugin_system.apis import send_api
logger = logging.getLogger(__name__)
class SetTypingStatusHandler(BaseEventHandler):
"""在LLM处理私聊消息后设置“正在输入”状态的事件处理器。"""
handler_name = "set_typing_status_handler"
handler_description = "在LLM生成回复后，将用户的聊天状态设置为“正在输入”。"
init_subscribe = [EventType.POST_LLM]
async def execute(self, params: dict) -> HandlerResult:
message = params.get("message")
if not message or not message.is_private_message:
return HandlerResult(success=True, continue_process=True)
user_id = message.message_info.user_info.user_id
if not user_id:
return HandlerResult(success=False, continue_process=True, message="无法获取用户ID")
try:
params = {"user_id": user_id, "event_type": 1}
await send_api.adapter_command_to_stream(
action="set_input_status",
params=params,
stream_id=message.stream_id,
)
logger.debug(f"成功为用户 {user_id} 设置“正在输入”状态。")
return HandlerResult(success=True, continue_process=True)
except Exception as e:
logger.error(f"为用户 {user_id} 设置“正在输入”状态时出错: {e}")
return HandlerResult(success=False, continue_process=True, message=str(e))
@register_plugin
class SetTypingStatusPlugin(BasePlugin):
"""一个在LLM生成回复时设置私聊输入状态的插件。"""
plugin_name = "set_typing_status"
enable_plugin = True
dependencies = []
python_dependencies = []
config_file_name = ""
config_schema = {}
def get_plugin_components(self) -> List[Tuple[ComponentInfo, Type]]:
"""注册插件的功能组件。"""
return [(SetTypingStatusHandler.get_handler_info(), SetTypingStatusHandler)]
def register_plugin(self) -> bool:
return True

0
src/plugins/built_in/at_user_plugin/__init__.py → src/plugins/social_toolkit_plugin/__init__.py
View File

25
src/plugins/social_toolkit_plugin/_manifest.json Normal file
View File

@@ -0,0 +1,25 @@
{
"manifest_version": 1,
"name": "MoFox-Bot工具箱",
"version": "1.0.0",
"description": "一个集合多种实用功能的插件，旨在提升聊天体验和效率。",
"author": {
"name": "MoFox-Studio",
"url": "https://github.com/MoFox-Studio"
},
"license": "GPL-v3.0-or-later",
"host_application": {
"min_version": "0.10.0"
},
"keywords": ["emoji", "reaction", "like", "表情", "回应", "点赞"],
"categories": ["Chat", "Integration"],
"default_locale": "zh-CN",
"locales_path": "_locales",
"plugin_info": {
"is_built_in": "true",
"plugin_type": "functional"
}
}

293
src/plugins/built_in/reminder_plugin/plugin.py → src/plugins/social_toolkit_plugin/plugin.py
View File

@@ -1,24 +1,26 @@
import asyncio
from datetime import datetime
from typing import List, Tuple, Type, Optional
import re
from typing import List, Tuple, Type
from dateutil.parser import parse as parse_datetime
from src.common.logger import get_logger
from src.manager.async_task_manager import AsyncTask, async_task_manager
from src.person_info.person_info import get_person_info_manager
from src.plugin_system import (
BaseAction,
ActionInfo,
BasePlugin,
register_plugin,
BaseAction,
ComponentInfo,
ActionActivationType,
ConfigField,
)
from src.common.logger import get_logger
from .qq_emoji_list import qq_face
from src.plugin_system.base.component_types import ChatType
from src.person_info.person_info import get_person_info_manager
from dateutil.parser import parse as parse_datetime
from src.manager.async_task_manager import AsyncTask, async_task_manager
from src.plugin_system.apis import send_api, llm_api, generator_api
from src.plugin_system.base.component_types import ComponentType
import asyncio
import datetime
logger = get_logger(__name__)
logger = get_logger("set_emoji_like_plugin")
# ============================ AsyncTask ============================
@@ -89,21 +91,217 @@ class ReminderTask(AsyncTask):
# =============================== Actions ===============================
def get_emoji_id(emoji_input: str) -> str | None:
"""根据输入获取表情ID"""
# 如果输入本身就是数字ID，直接返回
if emoji_input.isdigit() or (isinstance(emoji_input, str) and emoji_input.startswith("😊")):
if emoji_input in qq_face:
return emoji_input
# 尝试从 "[表情：xxx]" 格式中提取
match = re.search(r"\[表情：(.+?)\]", emoji_input)
if match:
emoji_name = match.group(1).strip()
else:
emoji_name = emoji_input.strip()
# 遍历查找
for key, value in qq_face.items():
# value 的格式是 "[表情：xxx]"
if f"[表情：{emoji_name}]" == value:
return key
return None
# ===== Action组件 =====
class PokeAction(BaseAction):
"""发送戳一戳动作"""
# === 基本信息（必须填写）===
action_name = "poke_user"
action_description = "向用户发送戳一戳"
activation_type = ActionActivationType.ALWAYS
parallel_action = True
# === 功能描述（必须填写）===
action_parameters = {
"user_name": "需要戳一戳的用户的名字 (可选)",
"user_id": "需要戳一戳的用户的ID (可选，优先级更高)",
"times": "需要戳一戳的次数 (默认为 1)",
}
action_require = ["当需要戳某个用户时使用", "当你想提醒特定用户时使用"]
llm_judge_prompt = """
判定是否需要使用戳一戳动作的条件：
1. 用户明确要求使用戳一戳。
2. 你想以一种有趣的方式提醒或与某人互动。
3. 上下文明确需要你戳一个或多个人。
请回答"是"或"否"。
"""
associated_types = ["text"]
async def execute(self) -> Tuple[bool, str]:
"""执行戳一戳的动作"""
user_id = self.action_data.get("user_id")
user_name = self.action_data.get("user_name")
try:
times = int(self.action_data.get("times", 1))
except (ValueError, TypeError):
times = 1
# 优先使用 user_id
if not user_id:
if not user_name:
logger.warning("戳一戳动作缺少 'user_id' 或 'user_name' 参数。")
return False, "缺少用户标识参数"
# 备用方案：通过 user_name 查找
user_info = await get_person_info_manager().get_person_info_by_name(user_name)
if not user_info or not user_info.get("user_id"):
logger.info(f"找不到名为 '{user_name}' 的用户。")
return False, f"找不到名为 '{user_name}' 的用户"
user_id = user_info.get("user_id")
display_name = user_name or user_id
for i in range(times):
logger.info(f"正在向 {display_name} ({user_id}) 发送第 {i + 1}/{times} 次戳一戳...")
await self.send_command(
"SEND_POKE", args={"qq_id": user_id}, display_message=f"戳了戳 {display_name} ({i + 1}/{times})"
)
# 添加一个小的延迟，以避免发送过快
await asyncio.sleep(0.5)
success_message = f"已向 {display_name} 发送 {times} 次戳一戳。"
await self.store_action_info(
action_build_into_prompt=True, action_prompt_display=success_message, action_done=True
)
return True, success_message
class SetEmojiLikeAction(BaseAction):
"""设置消息表情回应"""
# === 基本信息（必须填写）===
action_name = "set_emoji_like"
action_description = "为某条已经存在的消息添加‘贴表情’回应（类似点赞），而不是发送新消息。可以在觉得某条消息非常有趣、值得赞同或者需要特殊情感回应时主动使用。"
activation_type = ActionActivationType.ALWAYS # 消息接收时激活(?)
chat_type_allow = ChatType.GROUP
parallel_action = True
# === 功能描述（必须填写）===
# 从 qq_face 字典中提取所有表情名称用于提示
emoji_options = []
for name in qq_face.values():
match = re.search(r"\[表情：(.+?)\]", name)
if match:
emoji_options.append(match.group(1))
action_parameters = {
"emoji": f"要回应的表情,必须从以下表情中选择: {', '.join(emoji_options)}",
"set": "是否设置回应 (True/False)",
}
action_require = [
"当需要对一个已存在消息进行‘贴表情’回应时使用",
"这是一个对旧消息的操作，而不是发送新消息",
"如果你想发送一个新的表情包消息，请使用 'emoji' 动作",
]
llm_judge_prompt = """
判定是否需要使用贴表情动作的条件：
1. 用户明确要求使用贴表情包
2. 这是一个适合表达强烈情绪的场合
3. 不要发送太多表情包，如果你已经发送过多个表情包则回答"否"
请回答"是"或"否"。
"""
associated_types = ["text"]
async def execute(self) -> Tuple[bool, str]:
"""执行设置表情回应的动作"""
message_id = None
if self.has_action_message:
logger.debug(str(self.action_message))
if isinstance(self.action_message, dict):
message_id = self.action_message.get("message_id")
logger.info(f"获取到的消息ID: {message_id}")
else:
logger.error("未提供消息ID")
await self.store_action_info(
action_build_into_prompt=True,
action_prompt_display=f"执行了set_emoji_like动作：{self.action_name},失败: 未提供消息ID",
action_done=False,
)
return False, "未提供消息ID"
emoji_input = self.action_data.get("emoji")
set_like = self.action_data.get("set", True)
if not emoji_input:
logger.error("未提供表情")
return False, "未提供表情"
logger.info(f"设置表情回应: {emoji_input}, 是否设置: {set_like}")
emoji_id = get_emoji_id(emoji_input)
if not emoji_id:
logger.error(f"找不到表情: '{emoji_input}'。请从可用列表中选择。")
await self.store_action_info(
action_build_into_prompt=True,
action_prompt_display=f"执行了set_emoji_like动作：{self.action_name},失败: 找不到表情: '{emoji_input}'",
action_done=False,
)
return False, f"找不到表情: '{emoji_input}'。请从可用列表中选择。"
# 4. 使用适配器API发送命令
if not message_id:
logger.error("未提供消息ID")
await self.store_action_info(
action_build_into_prompt=True,
action_prompt_display=f"执行了set_emoji_like动作：{self.action_name},失败: 未提供消息ID",
action_done=False,
)
return False, "未提供消息ID"
try:
# 使用适配器API发送贴表情命令
success = await self.send_command(
command_name="set_emoji_like", args={"message_id": message_id, "emoji_id": emoji_id, "set": set_like}, storage_message=False
)
if success:
logger.info("设置表情回应成功")
await self.store_action_info(
action_build_into_prompt=True,
action_prompt_display=f"执行了set_emoji_like动作,{emoji_input},设置表情回应: {emoji_id}, 是否设置: {set_like}",
action_done=True,
)
return True, "成功设置表情回应"
else:
logger.error("设置表情回应失败")
await self.store_action_info(
action_build_into_prompt=True,
action_prompt_display=f"执行了set_emoji_like动作：{self.action_name},失败",
action_done=False,
)
return False, "设置表情回应失败"
except Exception as e:
logger.error(f"设置表情回应失败: {e}")
await self.store_action_info(
action_build_into_prompt=True,
action_prompt_display=f"执行了set_emoji_like动作：{self.action_name},失败: {e}",
action_done=False,
)
return False, f"设置表情回应失败: {e}"
class RemindAction(BaseAction):
"""一个能从对话中智能识别并设置定时提醒的动作。"""
# === 基本信息 ===
action_name = "set_reminder"
action_description = "根据用户的对话内容，智能地设置一个未来的提醒事项。"
@staticmethod
def get_action_info() -> ActionInfo:
return ActionInfo(
name="set_reminder",
component_type=ComponentType.ACTION,
activation_type=ActionActivationType.KEYWORD,
activation_keywords=["提醒", "叫我", "记得", "别忘了"]
)
activation_type=ActionActivationType.KEYWORD,
activation_keywords=["提醒", "叫我", "记得", "别忘了"]
# === LLM 判断与参数提取 ===
llm_judge_prompt = ""
@@ -319,23 +517,42 @@ class RemindAction(BaseAction):
await self.send_text("抱歉，设置提醒时发生了一点内部错误。")
return False, "设置提醒时发生内部错误"
# =============================== Plugin ===============================
# ===== 插件注册 =====
@register_plugin
class ReminderPlugin(BasePlugin):
"""一个能从对话中智能识别并设置定时提醒的插件。"""
class SetEmojiLikePlugin(BasePlugin):
"""一个集合多种实用功能的插件，旨在提升聊天体验和效率。"""
# --- 插件基础信息 ---
plugin_name = "reminder_plugin"
enable_plugin = True
dependencies = []
python_dependencies = []
config_file_name = "config.toml"
config_schema = {}
# 插件基本信息
plugin_name: str = "social_toolkit_plugin" # 内部标识符
enable_plugin: bool = True
dependencies: List[str] = [] # 插件依赖列表
python_dependencies: List[str] = [] # Python包依赖列表，现在使用内置API
config_file_name: str = "config.toml" # 配置文件名
def get_plugin_components(self) -> List[Tuple[ActionInfo, Type[BaseAction]]]:
"""注册插件的所有功能组件。"""
return [
(RemindAction.get_action_info(), RemindAction)
]
# 配置节描述
config_section_descriptions = {"plugin": "插件基本信息", "components": "插件组件"}
# 配置Schema定义
config_schema: dict = {
"plugin": {
"name": ConfigField(type=str, default="set_emoji_like", description="插件名称"),
"version": ConfigField(type=str, default="1.0.0", description="插件版本"),
"enabled": ConfigField(type=bool, default=True, description="是否启用插件"),
"config_version": ConfigField(type=str, default="1.1", description="配置版本"),
},
"components": {
"action_set_emoji_like": ConfigField(type=bool, default=True, description="是否启用设置表情回应功能"),
"action_poke_enable": ConfigField(type=bool, default=True, description="是否启用戳一戳功能"),
"action_set_reminder_enable": ConfigField(type=bool, default=True, description="是否启用定时提醒功能"),
},
}
def get_plugin_components(self) -> List[Tuple[ComponentInfo, Type]]:
enable_components = []
if self.get_config("components.action_set_emoji_like"):
enable_components.append((SetEmojiLikeAction.get_action_info(), SetEmojiLikeAction))
if self.get_config("components.action_poke_enable"):
enable_components.append((PokeAction.get_action_info(), PokeAction))
if self.get_config("components.action_set_reminder_enable"):
enable_components.append((RemindAction.get_action_info(), RemindAction))
return enable_components

252
src/plugins/social_toolkit_plugin/qq_emoji_list.py Normal file
View File

@@ -0,0 +1,252 @@
qq_face: dict = {
"0": "[表情：惊讶]",
"1": "[表情：撇嘴]",
"2": "[表情：色]",
"3": "[表情：发呆]",
"4": "[表情：得意]",
"5": "[表情：流泪]",
"6": "[表情：害羞]",
"7": "[表情：闭嘴]",
"8": "[表情：睡]",
"9": "[表情：大哭]",
"10": "[表情：尴尬]",
"11": "[表情：发怒]",
"12": "[表情：调皮]",
"13": "[表情：呲牙]",
"14": "[表情：微笑]",
"15": "[表情：难过]",
"16": "[表情：酷]",
"18": "[表情：抓狂]",
"19": "[表情：吐]",
"20": "[表情：偷笑]",
"21": "[表情：可爱]",
"22": "[表情：白眼]",
"23": "[表情：傲慢]",
"24": "[表情：饥饿]",
"25": "[表情：困]",
"26": "[表情：惊恐]",
"27": "[表情：流汗]",
"28": "[表情：憨笑]",
"29": "[表情：悠闲]",
"30": "[表情：奋斗]",
"31": "[表情：咒骂]",
"32": "[表情：疑问]",
"33": "[表情： 嘘]",
"34": "[表情：晕]",
"35": "[表情：折磨]",
"36": "[表情：衰]",
"37": "[表情：骷髅]",
"38": "[表情：敲打]",
"39": "[表情：再见]",
"41": "[表情：发抖]",
"42": "[表情：爱情]",
"43": "[表情：跳跳]",
"46": "[表情：猪头]",
"49": "[表情：拥抱]",
"53": "[表情：蛋糕]",
"56": "[表情：刀]",
"59": "[表情：便便]",
"60": "[表情：咖啡]",
"63": "[表情：玫瑰]",
"64": "[表情：凋谢]",
"66": "[表情：爱心]",
"67": "[表情：心碎]",
"74": "[表情：太阳]",
"75": "[表情：月亮]",
"76": "[表情：赞]",
"77": "[表情：踩]",
"78": "[表情：握手]",
"79": "[表情：胜利]",
"85": "[表情：飞吻]",
"86": "[表情：怄火]",
"89": "[表情：西瓜]",
"96": "[表情：冷汗]",
"97": "[表情：擦汗]",
"98": "[表情：抠鼻]",
"99": "[表情：鼓掌]",
"100": "[表情：糗大了]",
"101": "[表情：坏笑]",
"102": "[表情：左哼哼]",
"103": "[表情：右哼哼]",
"104": "[表情：哈欠]",
"105": "[表情：鄙视]",
"106": "[表情：委屈]",
"107": "[表情：快哭了]",
"108": "[表情：阴险]",
"109": "[表情：左亲亲]",
"110": "[表情：吓]",
"111": "[表情：可怜]",
"112": "[表情：菜刀]",
"114": "[表情：篮球]",
"116": "[表情：示爱]",
"118": "[表情：抱拳]",
"119": "[表情：勾引]",
"120": "[表情：拳头]",
"121": "[表情：差劲]",
"123": "[表情：NO]",
"124": "[表情：OK]",
"125": "[表情：转圈]",
"129": "[表情：挥手]",
"137": "[表情：鞭炮]",
"144": "[表情：喝彩]",
"146": "[表情：爆筋]",
"147": "[表情：棒棒糖]",
"169": "[表情：手枪]",
"171": "[表情：茶]",
"172": "[表情：眨眼睛]",
"173": "[表情：泪奔]",
"174": "[表情：无奈]",
"175": "[表情：卖萌]",
"176": "[表情：小纠结]",
"177": "[表情：喷血]",
"178": "[表情：斜眼笑]",
"179": "[表情：doge]",
"181": "[表情：戳一戳]",
"182": "[表情：笑哭]",
"183": "[表情：我最美]",
"185": "[表情：羊驼]",
"187": "[表情：幽灵]",
"201": "[表情：点赞]",
"212": "[表情：托腮]",
"262": "[表情：脑阔疼]",
"263": "[表情：沧桑]",
"264": "[表情：捂脸]",
"265": "[表情：辣眼睛]",
"266": "[表情：哦哟]",
"267": "[表情：头秃]",
"268": "[表情：问号脸]",
"269": "[表情：暗中观察]",
"270": "[表情：emm]",
"271": "[表情：吃 瓜]",
"272": "[表情：呵呵哒]",
"273": "[表情：我酸了]",
"277": "[表情：汪汪]",
"281": "[表情：无眼笑]",
"282": "[表情：敬礼]",
"283": "[表情：狂笑]",
"284": "[表情：面无表情]",
"285": "[表情：摸鱼]",
"286": "[表情：魔鬼笑]",
"287": "[表情：哦]",
"289": "[表情：睁眼]",
"293": "[表情：摸锦鲤]",
"294": "[表情：期待]",
"295": "[表情：拿到红包]",
"297": "[表情：拜谢]",
"298": "[表情：元宝]",
"299": "[表情：牛啊]",
"300": "[表情：胖三斤]",
"302": "[表情：左拜年]",
"303": "[表情：右拜年]",
"305": "[表情：右亲亲]",
"306": "[表情：牛气冲天]",
"307": "[表情：喵喵]",
"311": "[表情：打call]",
"312": "[表情：变形]",
"314": "[表情：仔细分析]",
"317": "[表情：菜汪]",
"318": "[表情：崇拜]",
"319": "[表情： 比心]",
"320": "[表情：庆祝]",
"323": "[表情：嫌弃]",
"324": "[表情：吃糖]",
"325": "[表情：惊吓]",
"326": "[表情：生气]",
"332": "[表情：举牌牌]",
"333": "[表情：烟花]",
"334": "[表情：虎虎生威]",
"336": "[表情：豹富]",
"337": "[表情：花朵脸]",
"338": "[表情：我想开了]",
"339": "[表情：舔屏]",
"341": "[表情：打招呼]",
"342": "[表情：酸Q]",
"343": "[表情：我方了]",
"344": "[表情：大怨种]",
"345": "[表情：红包多多]",
"346": "[表情：你真棒棒]",
"347": "[表情：大展宏兔]",
"349": "[表情：坚强]",
"350": "[表情：贴贴]",
"351": "[表情：敲敲]",
"352": "[表情：咦]",
"353": "[表情：拜托]",
"354": "[表情：尊嘟假嘟]",
"355": "[表情：耶]",
"356": "[表情：666]",
"357": "[表情：裂开]",
"392": "[表情：龙年 快乐]",
"393": "[表情：新年中龙]",
"394": "[表情：新年大龙]",
"395": "[表情：略略略]",
"396": "[表情：龙年快乐]",
"424": " [表情：按钮]",
"😊": "[表情：嘿嘿]",
"😌": "[表情：羞涩]",
"😚": "[ 表情：亲亲]",
"😓": "[表情：汗]",
"😰": "[表情：紧张]",
"😝": "[表情：吐舌]",
"😁": "[表情：呲牙]",
"😜": "[表情：淘气]",
"☺": "[表情：可爱]",
"😍": "[表情：花痴]",
"😔": "[表情：失落]",
"😄": "[表情：高兴]",
"😏": "[表情：哼哼]",
"😒": "[表情：不屑]",
"😳": "[表情：瞪眼]",
"😘": "[表情：飞吻]",
"😭": "[表情：大哭]",
"😱": "[表情：害怕]",
"😂": "[表情：激动]",
"💪": "[表情：肌肉]",
"👊": "[表情：拳头]",
"👍": "[表情 ：厉害]",
"👏": "[表情：鼓掌]",
"👎": "[表情：鄙视]",
"🙏": "[表情：合十]",
"👌": "[表情：好的]",
"👆": "[表情：向上]",
"👀": "[表情：眼睛]",
"🍜": "[表情：拉面]",
"🍧": "[表情：刨冰]",
"🍞": "[表情：面包]",
"🍺": "[表情：啤酒]",
"🍻": "[表情：干杯]",
"☕": "[表情：咖啡]",
"🍎": "[表情：苹果]",
"🍓": "[表情：草莓]",
"🍉": "[表情：西瓜]",
"🚬": "[表情：吸烟]",
"🌹": "[表情：玫瑰]",
"🎉": "[表情：庆祝]",
"💝": "[表情：礼物]",
"💣": "[表情：炸弹]",
"✨": "[表情：闪光]",
"💨": "[表情：吹气]",
"💦": "[表情：水]",
"🔥": "[表情：火]",
"💤": "[表情：睡觉]",
"💩": "[表情：便便]",
"💉": "[表情：打针]",
"📫": "[表情：邮箱]",
"🐎": "[表情：骑马]",
"👧": "[表情：女孩]",
"👦": "[表情：男孩]",
"🐵": "[表情：猴]",
"🐷": "[表情：猪]",
"🐮": "[表情：牛]",
"🐔": "[表情：公鸡]",
"🐸": "[表情：青蛙]",
"👻": "[表情：幽灵]",
"🐛": "[表情：虫]",
"🐶": "[表情：狗]",
"🐳": "[表情：鲸鱼]",
"👢": "[表情：靴子]",
"☀": "[表情：晴天]",
"❔": "[表情：问号]",
"🔫": "[表情：手枪]",
"💓": "[表情：爱 心]",
"🏪": "[表情：便利店]",
}