mirror of https://github.com/Mai-with-u/MaiBot.git
400 lines
15 KiB
Python
400 lines
15 KiB
Python
from typing import List, Tuple, Type
|
||
from src.plugin_system import (
|
||
BasePlugin,
|
||
register_plugin,
|
||
BaseCommand,
|
||
ComponentInfo,
|
||
ConfigField,
|
||
ReplyContentType,
|
||
emoji_api,
|
||
)
|
||
from maim_message import Seg
|
||
from src.common.logger import get_logger
|
||
|
||
logger = get_logger("emoji_manage_plugin")
|
||
|
||
|
||
class AddEmojiCommand(BaseCommand):
|
||
command_name = "add_emoji"
|
||
command_description = "添加表情包"
|
||
command_pattern = r".*/emoji add.*"
|
||
|
||
async def execute(self) -> Tuple[bool, str, bool]:
|
||
# 查找消息中的表情包
|
||
# logger.info(f"查找消息中的表情包: {self.message.message_segment}")
|
||
|
||
emoji_base64_list = self.find_and_return_emoji_in_message(self.message.message_segment)
|
||
|
||
if not emoji_base64_list:
|
||
return False, "未在消息中找到表情包或图片", False
|
||
|
||
# 注册找到的表情包
|
||
success_count = 0
|
||
fail_count = 0
|
||
results = []
|
||
|
||
for i, emoji_base64 in enumerate(emoji_base64_list):
|
||
try:
|
||
# 使用emoji_api注册表情包（让API自动生成唯一文件名）
|
||
result = await emoji_api.register_emoji(emoji_base64)
|
||
|
||
if result["success"]:
|
||
success_count += 1
|
||
description = result.get("description", "未知描述")
|
||
emotions = result.get("emotions", [])
|
||
replaced = result.get("replaced", False)
|
||
|
||
result_msg = f"表情包 {i + 1} 注册成功{'(替换旧表情包)' if replaced else '(新增表情包)'}"
|
||
if description:
|
||
result_msg += f"\n描述: {description}"
|
||
if emotions:
|
||
result_msg += f"\n情感标签: {', '.join(emotions)}"
|
||
|
||
results.append(result_msg)
|
||
else:
|
||
fail_count += 1
|
||
error_msg = result.get("message", "注册失败")
|
||
results.append(f"表情包 {i + 1} 注册失败: {error_msg}")
|
||
|
||
except Exception as e:
|
||
fail_count += 1
|
||
results.append(f"表情包 {i + 1} 注册时发生错误: {str(e)}")
|
||
|
||
# 构建返回消息
|
||
total_count = success_count + fail_count
|
||
summary_msg = f"表情包注册完成: 成功 {success_count} 个，失败 {fail_count} 个，共处理 {total_count} 个"
|
||
|
||
# 如果有结果详情，添加到返回消息中
|
||
details_msg = ""
|
||
if results:
|
||
details_msg = "\n" + "\n".join(results)
|
||
final_msg = summary_msg + details_msg
|
||
else:
|
||
final_msg = summary_msg
|
||
|
||
# 使用表达器重写回复
|
||
try:
|
||
from src.plugin_system.apis import generator_api
|
||
|
||
# 构建重写数据
|
||
rewrite_data = {
|
||
"raw_reply": summary_msg,
|
||
"reason": f"注册了表情包：{details_msg}\n",
|
||
}
|
||
|
||
# 调用表达器重写
|
||
result_status, data = await generator_api.rewrite_reply(
|
||
chat_stream=self.message.chat_stream,
|
||
reply_data=rewrite_data,
|
||
)
|
||
|
||
if result_status:
|
||
# 发送重写后的回复
|
||
for reply_seg in data.reply_set.reply_data:
|
||
send_data = reply_seg.content
|
||
await self.send_text(send_data)
|
||
|
||
return success_count > 0, final_msg, success_count > 0
|
||
else:
|
||
# 如果重写失败，发送原始消息
|
||
await self.send_text(final_msg)
|
||
return success_count > 0, final_msg, success_count > 0
|
||
|
||
except Exception as e:
|
||
# 如果表达器调用失败，发送原始消息
|
||
logger.error(f"[add_emoji] 表达器重写失败: {e}")
|
||
await self.send_text(final_msg)
|
||
return success_count > 0, final_msg, success_count > 0
|
||
|
||
def find_and_return_emoji_in_message(self, message_segments) -> List[str]:
|
||
emoji_base64_list = []
|
||
|
||
# 处理单个Seg对象的情况
|
||
if isinstance(message_segments, Seg):
|
||
if message_segments.type == "emoji":
|
||
emoji_base64_list.append(message_segments.data)
|
||
elif message_segments.type == "image":
|
||
# 假设图片数据是base64编码的
|
||
emoji_base64_list.append(message_segments.data)
|
||
elif message_segments.type == "seglist":
|
||
# 递归处理嵌套的Seg列表
|
||
emoji_base64_list.extend(self.find_and_return_emoji_in_message(message_segments.data))
|
||
return emoji_base64_list
|
||
|
||
# 处理Seg列表的情况
|
||
for seg in message_segments:
|
||
if seg.type == "emoji":
|
||
emoji_base64_list.append(seg.data)
|
||
elif seg.type == "image":
|
||
# 假设图片数据是base64编码的
|
||
emoji_base64_list.append(seg.data)
|
||
elif seg.type == "seglist":
|
||
# 递归处理嵌套的Seg列表
|
||
emoji_base64_list.extend(self.find_and_return_emoji_in_message(seg.data))
|
||
return emoji_base64_list
|
||
|
||
|
||
class ListEmojiCommand(BaseCommand):
|
||
"""列表表情包Command - 响应/emoji list命令"""
|
||
|
||
command_name = "emoji_list"
|
||
command_description = "列表表情包"
|
||
|
||
# === 命令设置（必须填写）===
|
||
command_pattern = r"^/emoji list(\s+\d+)?$" # 匹配 "/emoji list" 或 "/emoji list 数量"
|
||
|
||
async def execute(self) -> Tuple[bool, str, bool]:
|
||
"""执行列表表情包"""
|
||
from src.plugin_system.apis import emoji_api
|
||
import datetime
|
||
|
||
# 解析命令参数
|
||
import re
|
||
|
||
match = re.match(r"^/emoji list(?:\s+(\d+))?$", self.message.raw_message)
|
||
max_count = 10 # 默认显示10个
|
||
if match and match.group(1):
|
||
max_count = min(int(match.group(1)), 50) # 最多显示50个
|
||
|
||
# 获取当前时间
|
||
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)
|
||
|
||
# 获取表情包信息
|
||
emoji_count = emoji_api.get_count()
|
||
emoji_info = emoji_api.get_info()
|
||
|
||
# 构建返回消息
|
||
message_lines = [
|
||
f"📊 表情包统计信息 ({time_str})",
|
||
f"• 总数: {emoji_count} / {emoji_info['max_count']}",
|
||
f"• 可用: {emoji_info['available_emojis']}",
|
||
]
|
||
|
||
if emoji_count == 0:
|
||
message_lines.append("\n❌ 暂无表情包")
|
||
final_message = "\n".join(message_lines)
|
||
await self.send_text(final_message)
|
||
return True, final_message, True
|
||
|
||
# 获取所有表情包
|
||
all_emojis = await emoji_api.get_all()
|
||
if not all_emojis:
|
||
message_lines.append("\n❌ 无法获取表情包列表")
|
||
final_message = "\n".join(message_lines)
|
||
await self.send_text(final_message)
|
||
return False, final_message, True
|
||
|
||
# 显示前N个表情包
|
||
display_emojis = all_emojis[:max_count]
|
||
message_lines.append(f"\n📋 显示前 {len(display_emojis)} 个表情包:")
|
||
|
||
for i, (_, description, emotion) in enumerate(display_emojis, 1):
|
||
# 截断过长的描述
|
||
short_desc = description[:50] + "..." if len(description) > 50 else description
|
||
message_lines.append(f"{i}. {short_desc} [{emotion}]")
|
||
|
||
# 如果还有更多表情包，显示总数
|
||
if len(all_emojis) > max_count:
|
||
message_lines.append(f"\n💡 还有 {len(all_emojis) - max_count} 个表情包未显示")
|
||
|
||
final_message = "\n".join(message_lines)
|
||
|
||
# 直接发送文本消息
|
||
await self.send_text(final_message)
|
||
|
||
return True, final_message, True
|
||
|
||
|
||
class DeleteEmojiCommand(BaseCommand):
|
||
command_name = "delete_emoji"
|
||
command_description = "删除表情包"
|
||
command_pattern = r".*/emoji delete.*"
|
||
|
||
async def execute(self) -> Tuple[bool, str, bool]:
|
||
# 查找消息中的表情包图片
|
||
logger.info(f"查找消息中的表情包用于删除: {self.message.message_segment}")
|
||
|
||
emoji_base64_list = self.find_and_return_emoji_in_message(self.message.message_segment)
|
||
|
||
if not emoji_base64_list:
|
||
return False, "未在消息中找到表情包或图片", False
|
||
|
||
# 删除找到的表情包
|
||
success_count = 0
|
||
fail_count = 0
|
||
results = []
|
||
|
||
for i, emoji_base64 in enumerate(emoji_base64_list):
|
||
try:
|
||
# 计算图片的哈希值来查找对应的表情包
|
||
import base64
|
||
import hashlib
|
||
|
||
# 确保base64字符串只包含ASCII字符
|
||
if isinstance(emoji_base64, str):
|
||
emoji_base64_clean = emoji_base64.encode("ascii", errors="ignore").decode("ascii")
|
||
else:
|
||
emoji_base64_clean = str(emoji_base64)
|
||
|
||
# 计算哈希值
|
||
image_bytes = base64.b64decode(emoji_base64_clean)
|
||
emoji_hash = hashlib.md5(image_bytes).hexdigest()
|
||
|
||
# 使用emoji_api删除表情包
|
||
result = await emoji_api.delete_emoji(emoji_hash)
|
||
|
||
if result["success"]:
|
||
success_count += 1
|
||
description = result.get("description", "未知描述")
|
||
count_before = result.get("count_before", 0)
|
||
count_after = result.get("count_after", 0)
|
||
emotions = result.get("emotions", [])
|
||
|
||
result_msg = f"表情包 {i + 1} 删除成功"
|
||
if description:
|
||
result_msg += f"\n描述: {description}"
|
||
if emotions:
|
||
result_msg += f"\n情感标签: {', '.join(emotions)}"
|
||
result_msg += f"\n表情包数量: {count_before} → {count_after}"
|
||
|
||
results.append(result_msg)
|
||
else:
|
||
fail_count += 1
|
||
error_msg = result.get("message", "删除失败")
|
||
results.append(f"表情包 {i + 1} 删除失败: {error_msg}")
|
||
|
||
except Exception as e:
|
||
fail_count += 1
|
||
results.append(f"表情包 {i + 1} 删除时发生错误: {str(e)}")
|
||
|
||
# 构建返回消息
|
||
total_count = success_count + fail_count
|
||
summary_msg = f"表情包删除完成: 成功 {success_count} 个，失败 {fail_count} 个，共处理 {total_count} 个"
|
||
|
||
# 如果有结果详情，添加到返回消息中
|
||
details_msg = ""
|
||
if results:
|
||
details_msg = "\n" + "\n".join(results)
|
||
final_msg = summary_msg + details_msg
|
||
else:
|
||
final_msg = summary_msg
|
||
|
||
# 使用表达器重写回复
|
||
try:
|
||
from src.plugin_system.apis import generator_api
|
||
|
||
# 构建重写数据
|
||
rewrite_data = {
|
||
"raw_reply": summary_msg,
|
||
"reason": f"删除了表情包：{details_msg}\n",
|
||
}
|
||
|
||
# 调用表达器重写
|
||
result_status, data = await generator_api.rewrite_reply(
|
||
chat_stream=self.message.chat_stream,
|
||
reply_data=rewrite_data,
|
||
)
|
||
|
||
if result_status:
|
||
# 发送重写后的回复
|
||
for reply_seg in data.reply_set.reply_data:
|
||
send_data = reply_seg.content
|
||
await self.send_text(send_data)
|
||
|
||
return success_count > 0, final_msg, success_count > 0
|
||
else:
|
||
# 如果重写失败，发送原始消息
|
||
await self.send_text(final_msg)
|
||
return success_count > 0, final_msg, success_count > 0
|
||
|
||
except Exception as e:
|
||
# 如果表达器调用失败，发送原始消息
|
||
logger.error(f"[delete_emoji] 表达器重写失败: {e}")
|
||
await self.send_text(final_msg)
|
||
return success_count > 0, final_msg, success_count > 0
|
||
|
||
def find_and_return_emoji_in_message(self, message_segments) -> List[str]:
|
||
emoji_base64_list = []
|
||
|
||
# 处理单个Seg对象的情况
|
||
if isinstance(message_segments, Seg):
|
||
if message_segments.type == "emoji":
|
||
emoji_base64_list.append(message_segments.data)
|
||
elif message_segments.type == "image":
|
||
# 假设图片数据是base64编码的
|
||
emoji_base64_list.append(message_segments.data)
|
||
elif message_segments.type == "seglist":
|
||
# 递归处理嵌套的Seg列表
|
||
emoji_base64_list.extend(self.find_and_return_emoji_in_message(message_segments.data))
|
||
return emoji_base64_list
|
||
|
||
# 处理Seg列表的情况
|
||
for seg in message_segments:
|
||
if seg.type == "emoji":
|
||
emoji_base64_list.append(seg.data)
|
||
elif seg.type == "image":
|
||
# 假设图片数据是base64编码的
|
||
emoji_base64_list.append(seg.data)
|
||
elif seg.type == "seglist":
|
||
# 递归处理嵌套的Seg列表
|
||
emoji_base64_list.extend(self.find_and_return_emoji_in_message(seg.data))
|
||
return emoji_base64_list
|
||
|
||
|
||
class RandomEmojis(BaseCommand):
|
||
command_name = "random_emojis"
|
||
command_description = "发送多张随机表情包"
|
||
command_pattern = r"^/random_emojis$"
|
||
|
||
async def execute(self):
|
||
emojis = await emoji_api.get_random(5)
|
||
if not emojis:
|
||
return False, "未找到表情包", False
|
||
emoji_base64_list = []
|
||
for emoji in emojis:
|
||
emoji_base64_list.append(emoji[0])
|
||
return await self.forward_images(emoji_base64_list)
|
||
|
||
async def forward_images(self, images: List[str]):
|
||
"""
|
||
把多张图片用合并转发的方式发给用户
|
||
"""
|
||
success = await self.send_forward([("0", "神秘用户", [(ReplyContentType.IMAGE, img)]) for img in images])
|
||
return (True, "已发送随机表情包", True) if success else (False, "发送随机表情包失败", False)
|
||
|
||
|
||
# ===== 插件注册 =====
|
||
|
||
|
||
@register_plugin
|
||
class EmojiManagePlugin(BasePlugin):
|
||
"""表情包管理插件 - 管理表情包"""
|
||
|
||
# 插件基本信息
|
||
plugin_name: str = "emoji_manage_plugin" # 内部标识符
|
||
enable_plugin: bool = False
|
||
dependencies: List[str] = [] # 插件依赖列表
|
||
python_dependencies: List[str] = [] # Python包依赖列表
|
||
config_file_name: str = "config.toml" # 配置文件名
|
||
|
||
# 配置节描述
|
||
config_section_descriptions = {"plugin": "插件基本信息", "emoji": "表情包功能配置"}
|
||
|
||
# 配置Schema定义
|
||
config_schema: dict = {
|
||
"plugin": {
|
||
"enabled": ConfigField(type=bool, default=True, description="是否启用插件"),
|
||
"config_version": ConfigField(type=str, default="1.0.1", description="配置文件版本"),
|
||
},
|
||
}
|
||
|
||
def get_plugin_components(self) -> List[Tuple[ComponentInfo, Type]]:
|
||
return [
|
||
(RandomEmojis.get_command_info(), RandomEmojis),
|
||
(AddEmojiCommand.get_command_info(), AddEmojiCommand),
|
||
(ListEmojiCommand.get_command_info(), ListEmojiCommand),
|
||
(DeleteEmojiCommand.get_command_info(), DeleteEmojiCommand),
|
||
]
|