消息发送API
消息发送API模块专门负责发送各种类型的消息，支持文本、表情包、图片等多种消息类型。
导入方式
python
from src.plugin_system.apis import send_api主要功能
1. 文本消息发送
text_to_group(text, group_id, platform="qq", typing=False, reply_to="", storage_message=True)
向群聊发送文本消息
参数：
text：要发送的文本内容group_id：群聊IDplatform：平台，默认为"qq"typing：是否显示正在输入reply_to：回复消息的格式，如"发送者:消息内容"storage_message：是否存储到数据库
返回：
bool：是否发送成功
text_to_user(text, user_id, platform="qq", typing=False, reply_to="", storage_message=True)
向用户发送私聊文本消息
参数与返回值同上
2. 表情包发送
emoji_to_group(emoji_base64, group_id, platform="qq", storage_message=True)
向群聊发送表情包
参数：
emoji_base64：表情包的base64编码group_id：群聊IDplatform：平台，默认为"qq"storage_message：是否存储到数据库
emoji_to_user(emoji_base64, user_id, platform="qq", storage_message=True)
向用户发送表情包
3. 图片发送
image_to_group(image_base64, group_id, platform="qq", storage_message=True)
向群聊发送图片
image_to_user(image_base64, user_id, platform="qq", storage_message=True)
向用户发送图片
4. 命令发送
command_to_group(command, group_id, platform="qq", storage_message=True)
向群聊发送命令
command_to_user(command, user_id, platform="qq", storage_message=True)
向用户发送命令
5. 自定义消息发送
custom_to_group(message_type, content, group_id, platform="qq", display_message="", typing=False, reply_to="", storage_message=True)
向群聊发送自定义类型消息
custom_to_user(message_type, content, user_id, platform="qq", display_message="", typing=False, reply_to="", storage_message=True)
向用户发送自定义类型消息
custom_message(message_type, content, target_id, is_group=True, platform="qq", display_message="", typing=False, reply_to="", storage_message=True)
通用的自定义消息发送
参数：
message_type：消息类型，如"text"、"image"、"emoji"等content：消息内容target_id：目标ID（群ID或用户ID）is_group：是否为群聊platform：平台display_message：显示消息typing：是否显示正在输入reply_to：回复消息storage_message：是否存储
使用示例
1. 基础文本发送
python
from src.plugin_system.apis import send_api
async def send_hello(chat_stream):
"""发送问候消息"""
if chat_stream.group_info:
# 群聊
success = await send_api.text_to_group(
text="大家好！",
group_id=chat_stream.group_info.group_id,
typing=True
)
else:
# 私聊
success = await send_api.text_to_user(
text="你好！",
user_id=chat_stream.user_info.user_id,
typing=True
)
return success2. 回复特定消息
python
async def reply_to_message(chat_stream, reply_text, original_sender, original_message):
"""回复特定消息"""
# 构建回复格式
reply_to = f"{original_sender}:{original_message}"
if chat_stream.group_info:
success = await send_api.text_to_group(
text=reply_text,
group_id=chat_stream.group_info.group_id,
reply_to=reply_to
)
else:
success = await send_api.text_to_user(
text=reply_text,
user_id=chat_stream.user_info.user_id,
reply_to=reply_to
)
return success3. 发送表情包
python
async def send_emoji_reaction(chat_stream, emotion):
"""根据情感发送表情包"""
from src.plugin_system.apis import emoji_api
# 获取表情包
emoji_result = await emoji_api.get_by_emotion(emotion)
if not emoji_result:
return False
emoji_base64, description, matched_emotion = emoji_result
# 发送表情包
if chat_stream.group_info:
success = await send_api.emoji_to_group(
emoji_base64=emoji_base64,
group_id=chat_stream.group_info.group_id
)
else:
success = await send_api.emoji_to_user(
emoji_base64=emoji_base64,
user_id=chat_stream.user_info.user_id
)
return success4. 在Action中发送消息
python
from src.plugin_system.base import BaseAction
class MessageAction(BaseAction):
async def execute(self, action_data, chat_stream):
message_type = action_data.get("type", "text")
content = action_data.get("content", "")
if message_type == "text":
success = await self.send_text(chat_stream, content)
elif message_type == "emoji":
success = await self.send_emoji(chat_stream, content)
elif message_type == "image":
success = await self.send_image(chat_stream, content)
else:
success = False
return {"success": success}
async def send_text(self, chat_stream, text):
if chat_stream.group_info:
return await send_api.text_to_group(text, chat_stream.group_info.group_id)
else:
return await send_api.text_to_user(text, chat_stream.user_info.user_id)
async def send_emoji(self, chat_stream, emoji_base64):
if chat_stream.group_info:
return await send_api.emoji_to_group(emoji_base64, chat_stream.group_info.group_id)
else:
return await send_api.emoji_to_user(emoji_base64, chat_stream.user_info.user_id)
async def send_image(self, chat_stream, image_base64):
if chat_stream.group_info:
return await send_api.image_to_group(image_base64, chat_stream.group_info.group_id)
else:
return await send_api.image_to_user(image_base64, chat_stream.user_info.user_id)5. 批量发送消息
python
async def broadcast_message(message: str, target_groups: list):
"""向多个群组广播消息"""
results = {}
for group_id in target_groups:
try:
success = await send_api.text_to_group(
text=message,
group_id=group_id,
typing=True
)
results[group_id] = success
except Exception as e:
results[group_id] = False
print(f"发送到群 {group_id} 失败: {e}")
return results6. 智能消息发送
python
async def smart_send(chat_stream, message_data):
"""智能发送不同类型的消息"""
message_type = message_data.get("type", "text")
content = message_data.get("content", "")
options = message_data.get("options", {})
# 根据聊天流类型选择发送方法
target_id = (chat_stream.group_info.group_id if chat_stream.group_info
else chat_stream.user_info.user_id)
is_group = chat_stream.group_info is not None
# 使用通用发送方法
success = await send_api.custom_message(
message_type=message_type,
content=content,
target_id=target_id,
is_group=is_group,
typing=options.get("typing", False),
reply_to=options.get("reply_to", ""),
display_message=options.get("display_message", "")
)
return success消息类型说明
支持的消息类型
"text"：纯文本消息"emoji"：表情包消息"image"：图片消息"command"：命令消息"video"：视频消息（如果支持）"audio"：音频消息（如果支持）
回复格式
回复消息使用格式："发送者:消息内容" 或 "发送者：消息内容"
系统会自动查找匹配的原始消息并进行回复。
高级用法
1. 消息发送队列
python
import asyncio
class MessageQueue:
def __init__(self):
self.queue = asyncio.Queue()
self.running = False
async def add_message(self, chat_stream, message_type, content, options=None):
"""添加消息到队列"""
message_item = {
"chat_stream": chat_stream,
"type": message_type,
"content": content,
"options": options or {}
}
await self.queue.put(message_item)
async def process_queue(self):
"""处理消息队列"""
self.running = True
while self.running:
try:
message_item = await asyncio.wait_for(self.queue.get(), timeout=1.0)
# 发送消息
success = await smart_send(
message_item["chat_stream"],
{
"type": message_item["type"],
"content": message_item["content"],
"options": message_item["options"]
}
)
# 标记任务完成
self.queue.task_done()
# 发送间隔
await asyncio.sleep(0.5)
except asyncio.TimeoutError:
continue
except Exception as e:
print(f"处理消息队列出错: {e}")2. 消息模板系统
python
class MessageTemplate:
def __init__(self):
self.templates = {
"welcome": "欢迎 {nickname} 加入群聊！",
"goodbye": "{nickname} 离开了群聊。",
"notification": "🔔 通知：{message}",
"error": "❌ 错误：{error_message}",
"success": "✅ 成功：{message}"
}
def format_message(self, template_name: str, **kwargs) -> str:
"""格式化消息模板"""
template = self.templates.get(template_name, "{message}")
return template.format(**kwargs)
async def send_template(self, chat_stream, template_name: str, **kwargs):
"""发送模板消息"""
message = self.format_message(template_name, **kwargs)
if chat_stream.group_info:
return await send_api.text_to_group(message, chat_stream.group_info.group_id)
else:
return await send_api.text_to_user(message, chat_stream.user_info.user_id)
# 使用示例
template_system = MessageTemplate()
await template_system.send_template(chat_stream, "welcome", nickname="张三")注意事项
- 异步操作：所有发送函数都是异步的，必须使用
await - 错误处理：发送失败时返回False，成功时返回True
- 发送频率：注意控制发送频率，避免被平台限制
- 内容限制：注意平台对消息内容和长度的限制
- 权限检查：确保机器人有发送消息的权限
- 编码格式：图片和表情包需要使用base64编码
- 存储选项：可以选择是否将发送的消息存储到数据库