Mofox-Core/docs/plugins/api/message-api.md

📡 消息API

📖 概述

消息API提供了发送各种类型消息的接口，支持文本、表情、图片等多种消息类型。新版API格式更加简洁直观，自动处理群聊/私聊判断。

🔄 基础消息发送

发送文本消息

# 新API格式 - 自动判断群聊/私聊
await self.send_text("这是一条文本消息")

# 发送多行文本
message = """
这是第一行
这是第二行
这是第三行
"""
await self.send_text(message.strip())

发送表情消息

# 新API格式 - 发送表情
await self.send_emoji("😊")
await self.send_emoji("🎉") 
await self.send_emoji("👋")

发送特定类型消息

# 发送图片
await self.send_type("image", "https://example.com/image.jpg")

# 发送音频
await self.send_type("audio", "audio_file_path")

# 发送视频
await self.send_type("video", "video_file_path")

# 发送文件
await self.send_type("file", "file_path")

🎯 跨目标消息发送

使用send_api模块发送消息

# 导入send_api
from src.plugin_system.apis import send_api

# 向指定群聊发送文本消息
success = await send_api.text_to_group(
    text="这是发送到群聊的消息",
    group_id="123456789",
    platform="qq"
)

# 向指定用户发送私聊消息
success = await send_api.text_to_user(
    text="这是私聊消息",
    user_id="987654321", 
    platform="qq"
)

# 向指定群聊发送表情
success = await send_api.emoji_to_group(
    emoji="😊",
    group_id="123456789",
    platform="qq"
)

# 向指定用户发送表情
success = await send_api.emoji_to_user(
    emoji="🎉",
    user_id="987654321",
    platform="qq"
)

通用目标消息发送

# 向任意目标发送任意类型消息
success = await send_api.message_to_target(
    message_type="text",           # 消息类型
    content="消息内容",            # 消息内容
    platform="qq",                # 平台
    target_id="123456789",        # 目标ID
    is_group=True,                # 是否为群聊
    display_message="显示消息"     # 可选：显示消息
)

📨 消息类型支持

支持的消息类型

类型	说明	新API方法	send_api方法
text	普通文本消息	await self.send_text()	await send_api.text_to_group()
emoji	表情消息	await self.send_emoji()	await send_api.emoji_to_group()
image	图片消息	await self.send_type("image", url)	await send_api.message_to_target()
audio	音频消息	await self.send_type("audio", path)	await send_api.message_to_target()
video	视频消息	await self.send_type("video", path)	await send_api.message_to_target()
file	文件消息	await self.send_type("file", path)	await send_api.message_to_target()

新API格式示例

class ExampleAction(BaseAction):
    async def execute(self) -> Tuple[bool, str]:
        # 文本消息 - 最常用
        await self.send_text("普通文本消息")
        
        # 表情消息 - 直接方法
        await self.send_emoji("🎉")
        
        # 图片消息
        await self.send_type("image", "/path/to/image.jpg")
        
        # 音频消息
        await self.send_type("audio", "/path/to/audio.mp3")
        
        # 文件消息
        await self.send_type("file", "/path/to/document.pdf")
        
        return True, "发送了多种类型的消息"

🔍 消息信息获取

获取当前消息信息

# 新API格式 - 直接属性访问
class ExampleCommand(BaseCommand):
    async def execute(self) -> Tuple[bool, str]:
        # 用户信息
        user_id = self.user_id
        user_nickname = self.user_nickname
        
        # 聊天信息
        is_group_chat = self.is_group
        chat_id = self.chat_id
        platform = self.platform
        
        # 消息内容
        message_text = self.message.processed_plain_text
        
        # 构建信息显示
        info = f"""
👤 用户: {user_nickname}({user_id})
💬 类型: {'群聊' if is_group_chat else '私聊'}
📱 平台: {platform}
📝 内容: {message_text}
        """.strip()
        
        await self.send_text(info)
        return True, "显示了消息信息"

获取群聊信息（如果适用）

# 在Action或Command中检查群聊信息
if self.is_group:
    group_info = self.message.message_info.group_info
    if group_info:
        group_id = group_info.group_id
        group_name = getattr(group_info, 'group_name', '未知群聊')
        
        await self.send_text(f"当前群聊: {group_name}({group_id})")
else:
    await self.send_text("当前是私聊对话")

🌐 平台支持

支持的平台

平台	标识	说明
QQ	qq	QQ聊天平台
微信	wechat	微信聊天平台
Discord	discord	Discord聊天平台

平台适配示例

class PlatformAdaptiveAction(BaseAction):
    async def execute(self) -> Tuple[bool, str]:
        # 获取当前平台
        current_platform = self.platform
        
        # 根据平台调整消息格式
        if current_platform == "qq":
            await self.send_text("[QQ] 这是QQ平台的消息")
            await self.send_emoji("🐧")  # QQ企鹅表情
        elif current_platform == "wechat":
            await self.send_text("【微信】这是微信平台的消息")
            await self.send_emoji("💬")  # 微信气泡表情
        elif current_platform == "discord":
            await self.send_text("**Discord** 这是Discord平台的消息")
            await self.send_emoji("🎮")  # Discord游戏表情
        else:
            await self.send_text(f"未知平台: {current_platform}")
        
        return True, f"发送了{current_platform}平台适配消息"

🎨 消息格式化和高级功能

长消息分割

# 自动处理长消息分割
long_message = "这是一条很长的消息..." * 100

# 新API会自动处理长消息分割
await self.send_text(long_message)

消息模板和格式化

class TemplateMessageAction(BaseAction):
    async def execute(self) -> Tuple[bool, str]:
        # 使用配置中的消息模板
        template = self.get_config("messages.greeting_template", "你好 {username}！")
        
        # 格式化消息
        formatted_message = template.format(
            username=self.user_nickname,
            time=datetime.now().strftime("%H:%M"),
            platform=self.platform
        )
        
        await self.send_text(formatted_message)
        
        # 根据配置决定是否发送表情
        if self.get_config("messages.include_emoji", True):
            await self.send_emoji("😊")
        
        return True, "发送了模板化消息"

条件消息发送

class ConditionalMessageAction(BaseAction):
    async def execute(self) -> Tuple[bool, str]:
        # 根据用户类型发送不同消息
        if self.is_group:
            await self.send_text(f"群聊消息 - 当前群成员: @{self.user_nickname}")
        else:
            await self.send_text(f"私聊消息 - 你好 {self.user_nickname}！")
        
        # 根据时间发送不同表情
        from datetime import datetime
        hour = datetime.now().hour
        
        if 6 <= hour < 12:
            await self.send_emoji("🌅")  # 早上
        elif 12 <= hour < 18:
            await self.send_emoji("☀️")  # 下午
        else:
            await self.send_emoji("🌙")  # 晚上
        
        return True, "发送了条件化消息"

🛠️ 高级消息发送功能

批量消息发送

class BatchMessageAction(BaseAction):
    async def execute(self) -> Tuple[bool, str]:
        messages = [
            ("text", "第一条消息"),
            ("emoji", "🎉"),
            ("text", "第二条消息"),
            ("emoji", "✨")
        ]
        
        for msg_type, content in messages:
            if msg_type == "text":
                await self.send_text(content)
            elif msg_type == "emoji":
                await self.send_emoji(content)
            
            # 可选：添加延迟避免消息发送过快
            import asyncio
            await asyncio.sleep(0.5)
        
        return True, "发送了批量消息"

错误处理和重试

class ReliableMessageAction(BaseAction):
    async def execute(self) -> Tuple[bool, str]:
        max_retries = 3
        retry_count = 0
        
        while retry_count < max_retries:
            try:
                await self.send_text("重要消息")
                return True, "消息发送成功"
            except Exception as e:
                retry_count += 1
                logger.warning(f"消息发送失败 (尝试 {retry_count}/{max_retries}): {e}")
                
                if retry_count < max_retries:
                    import asyncio
                    await asyncio.sleep(1)  # 等待1秒后重试
        
        return False, "消息发送失败，已达到最大重试次数"

📝 最佳实践

1. 消息发送最佳实践

# ✅ 好的做法
class GoodMessageAction(BaseAction):
    async def execute(self) -> Tuple[bool, str]:
        # 1. 检查配置
        if not self.get_config("features.enable_messages", True):
            return True, "消息功能已禁用"
        
        # 2. 简洁的消息发送
        await self.send_text("简洁明了的消息")
        
        # 3. 适当的表情使用
        if self.get_config("features.enable_emoji", True):
            await self.send_emoji("😊")
        
        return True, "消息发送完成"

# ❌ 避免的做法
class BadMessageAction(BaseAction):
    async def execute(self) -> Tuple[bool, str]:
        # 避免：过长的消息
        await self.send_text("这是一条非常非常长的消息" * 50)
        
        # 避免：过多的表情
        for emoji in ["😊", "🎉", "✨", "🌟", "💫"]:
            await self.send_emoji(emoji)
        
        return True, "发送了糟糕的消息"

2. 错误处理

# ✅ 推荐的错误处理
class SafeMessageAction(BaseAction):
    async def execute(self) -> Tuple[bool, str]:
        try:
            message = self.get_config("messages.default", "默认消息")
            await self.send_text(message)
            return True, "消息发送成功"
        except Exception as e:
            logger.error(f"消息发送失败: {e}")
            # 可选：发送备用消息
            await self.send_text("消息发送遇到问题，请稍后再试")
            return False, f"发送失败: {str(e)}"

3. 性能优化

# ✅ 性能友好的消息发送
class OptimizedMessageAction(BaseAction):
    async def execute(self) -> Tuple[bool, str]:
        # 合并多个短消息为一条长消息
        parts = [
            "第一部分信息",
            "第二部分信息", 
            "第三部分信息"
        ]
        
        combined_message = "\n".join(parts)
        await self.send_text(combined_message)
        
        return True, "发送了优化的消息"

通过新的API格式，消息发送变得更加简洁高效！