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

send_api_doc与reply_to修改

Browse Source
This commit is contained in:
UnCLAS-Prommer
2025-07-28 11:47:32 +08:00
parent 0c302c9ca5
commit d643a85a0a
3 changed files with 103 additions and 539 deletions
Download Patch File Download Diff File Expand all files Collapse all files

377
docs/plugins/api/send-api.md
View File

@@ -6,86 +6,108 @@
```python
from src.plugin_system.apis import send_api
# 或者
from src.plugin_system import send_api
```
## 主要功能
### 1. 文本消息发送
### 1. 发送文本消息
```python
async def text_to_stream(
text: str,
stream_id: str,
typing: bool = False,
reply_to: str = "",
storage_message: bool = True,
) -> bool:
```
发送文本消息到指定的流
#### `text_to_group(text, group_id, platform="qq", typing=False, reply_to="", storage_message=True)`
向群聊发送文本消息
**Args:**
- `text` (str): 要发送的文本内容
- `stream_id` (str): 聊天流ID
- `typing` (bool): 是否显示正在输入
- `reply_to` (str): 回复消息，格式为"发送者:消息内容"
- `storage_message` (bool): 是否存储消息到数据库
**参数：**
- `text`：要发送的文本内容
- `group_id`：群聊ID
- `platform`：平台，默认为"qq"
- `typing`：是否显示正在输入
- `reply_to`：回复消息的格式，如"发送者:消息内容"
- `storage_message`：是否存储到数据库
**Returns:**
- `bool` - 是否发送成功
**返回：**
- `bool`：是否发送成功
### 2. 发送表情包
```python
async def emoji_to_stream(emoji_base64: str, stream_id: str, storage_message: bool = True) -> bool:
```
向指定流发送表情包。
#### `text_to_user(text, user_id, platform="qq", typing=False, reply_to="", storage_message=True)`
向用户发送私聊文本消息
**Args:**
- `emoji_base64` (str): 表情包的base64编码
- `stream_id` (str): 聊天流ID
- `storage_message` (bool): 是否存储消息到数据库
**参数与返回值同上**
**Returns:**
- `bool` - 是否发送成功
### 2. 表情包发送
### 3. 发送图片
```python
async def image_to_stream(image_base64: str, stream_id: str, storage_message: bool = True) -> bool:
```
向指定流发送图片。
#### `emoji_to_group(emoji_base64, group_id, platform="qq", storage_message=True)`
向群聊发送表情包
**Args:**
- `image_base64` (str): 图片的base64编码
- `stream_id` (str): 聊天流ID
- `storage_message` (bool): 是否存储消息到数据库
**参数：**
- `emoji_base64`：表情包的base64编码
- `group_id`：群聊ID
- `platform`：平台，默认为"qq"
- `storage_message`：是否存储到数据库
**Returns:**
- `bool` - 是否发送成功
#### `emoji_to_user(emoji_base64, user_id, platform="qq", storage_message=True)`
向用户发送表情包
### 4. 发送命令
```python
async def command_to_stream(command: Union[str, dict], stream_id: str, storage_message: bool = True, display_message: str = "") -> bool:
```
向指定流发送命令。
### 3. 图片发送
**Args:**
- `command` (Union[str, dict]): 命令内容
- `stream_id` (str): 聊天流ID
- `storage_message` (bool): 是否存储消息到数据库
- `display_message` (str): 显示消息
#### `image_to_group(image_base64, group_id, platform="qq", storage_message=True)`
向群聊发送图片
**Returns:**
- `bool` - 是否发送成功
#### `image_to_user(image_base64, user_id, platform="qq", storage_message=True)`
向用户发送图片
### 5. 发送自定义类型消息
```python
async def custom_to_stream(
message_type: str,
content: str,
stream_id: str,
display_message: str = "",
typing: bool = False,
reply_to: str = "",
storage_message: bool = True,
show_log: bool = True,
) -> bool:
```
向指定流发送自定义类型消息。
### 4. 命令发送
**Args:**
- `message_type` (str): 消息类型，如"text"、"image"、"emoji"、"video"、"file"等
- `content` (str): 消息内容（通常是base64编码或文本）
- `stream_id` (str): 聊天流ID
- `display_message` (str): 显示消息
- `typing` (bool): 是否显示正在输入
- `reply_to` (str): 回复消息，格式为"发送者:消息内容"
- `storage_message` (bool): 是否存储消息到数据库
- `show_log` (bool): 是否显示日志
#### `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`：是否存储
**Returns:**
- `bool` - 是否发送成功
## 使用示例
### 1. 基础文本发送
### 1. 基础文本发送，并回复消息
```python
from src.plugin_system.apis import send_api
@@ -93,57 +115,23 @@ 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
success = await send_api.text_to_stream(
text="Hello, world!",
stream_id=chat_stream.stream_id,
typing=True,
reply_to="User:How are you?",
storage_message=True
)
return success
```
### 2. 回复特定消息
```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 success
```
### 3. 发送表情包
### 2. 发送表情包
```python
from src.plugin_system.apis import emoji_api
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:
@@ -152,107 +140,10 @@ async def send_emoji_reaction(chat_stream, emotion):
emoji_base64, description, matched_emotion = emoji_result
# 发送表情包
if chat_stream.group_info:
success = await send_api.emoji_to_group(
success = await send_api.emoji_to_stream(
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 success
```
### 4. 在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 results
```
### 6. 智能消息发送
```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", "")
stream_id=chat_stream.stream_id,
storage_message=False # 不存储到数据库
)
return success
@@ -273,90 +164,6 @@ async def smart_send(chat_stream, message_data):
系统会自动查找匹配的原始消息并进行回复。
## 高级用法
### 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="张三")
```
## 注意事项
1. **异步操作**：所有发送函数都是异步的，必须使用`await`

2
src/chat/message_receive/message.py
View File

@@ -444,7 +444,7 @@ class MessageSending(MessageProcessBase):
is_emoji: bool = False,
thinking_start_time: float = 0,
apply_set_reply_logic: bool = False,
reply_to: str = None, # type: ignore
reply_to: Optional[str] = None,
):
# 调用父类初始化
super().__init__(

259
src/plugin_system/apis/send_api.py
View File

@@ -49,7 +49,6 @@ async def _send_to_target(
display_message: str = "",
typing: bool = False,
reply_to: str = "",
reply_to_platform_id: str = "",
storage_message: bool = True,
show_log: bool = True,
) -> bool:
@@ -60,8 +59,10 @@ async def _send_to_target(
content: 消息内容
stream_id: 目标流ID
display_message: 显示消息
typing: 是否显示正在输入
reply_to: 回复消息的格式，如"发送者:消息内容"
typing: 是否模拟打字等待。
reply_to: 回复消息，格式为"发送者:消息内容"
storage_message: 是否存储消息到数据库
show_log: 发送是否显示日志
Returns:
bool: 是否发送成功
@@ -95,8 +96,11 @@ async def _send_to_target(
# 处理回复消息
anchor_message = None
reply_to_platform_id: Optional[str] = None
if reply_to:
anchor_message = await _find_reply_message(target_stream, reply_to)
if anchor_message and anchor_message.message_info.user_info:
reply_to_platform_id = f"{anchor_message.message_info.platform}:{anchor_message.message_info.user_info.user_id}"
# 构建发送消息对象
bot_message = MessageSending(
@@ -252,7 +256,6 @@ async def text_to_stream(
stream_id: str,
typing: bool = False,
reply_to: str = "",
reply_to_platform_id: str = "",
storage_message: bool = True,
) -> bool:
"""向指定流发送文本消息
@@ -267,7 +270,7 @@ async def text_to_stream(
Returns:
bool: 是否发送成功
"""
return await _send_to_target("text", text, stream_id, "", typing, reply_to, reply_to_platform_id, storage_message)
return await _send_to_target("text", text, stream_id, "", typing, reply_to, storage_message=storage_message)
async def emoji_to_stream(emoji_base64: str, stream_id: str, storage_message: bool = True) -> bool:
@@ -350,249 +353,3 @@ async def custom_to_stream(
storage_message=storage_message,
show_log=show_log,
)
async def text_to_group(
text: str,
group_id: str,
platform: str = "qq",
typing: bool = False,
reply_to: str = "",
storage_message: bool = True,
) -> bool:
"""向群聊发送文本消息
Args:
text: 要发送的文本内容
group_id: 群聊ID
platform: 平台，默认为"qq"
typing: 是否显示正在输入
reply_to: 回复消息，格式为"发送者:消息内容"
Returns:
bool: 是否发送成功
"""
stream_id = get_chat_manager().get_stream_id(platform, group_id, True)
return await _send_to_target("text", text, stream_id, "", typing, reply_to, storage_message=storage_message)
async def text_to_user(
text: str,
user_id: str,
platform: str = "qq",
typing: bool = False,
reply_to: str = "",
storage_message: bool = True,
) -> bool:
"""向用户发送私聊文本消息
Args:
text: 要发送的文本内容
user_id: 用户ID
platform: 平台，默认为"qq"
typing: 是否显示正在输入
reply_to: 回复消息，格式为"发送者:消息内容"
Returns:
bool: 是否发送成功
"""
stream_id = get_chat_manager().get_stream_id(platform, user_id, False)
return await _send_to_target("text", text, stream_id, "", typing, reply_to, storage_message=storage_message)
async def emoji_to_group(emoji_base64: str, group_id: str, platform: str = "qq", storage_message: bool = True) -> bool:
"""向群聊发送表情包
Args:
emoji_base64: 表情包的base64编码
group_id: 群聊ID
platform: 平台，默认为"qq"
Returns:
bool: 是否发送成功
"""
stream_id = get_chat_manager().get_stream_id(platform, group_id, True)
return await _send_to_target("emoji", emoji_base64, stream_id, "", typing=False, storage_message=storage_message)
async def emoji_to_user(emoji_base64: str, user_id: str, platform: str = "qq", storage_message: bool = True) -> bool:
"""向用户发送表情包
Args:
emoji_base64: 表情包的base64编码
user_id: 用户ID
platform: 平台，默认为"qq"
Returns:
bool: 是否发送成功
"""
stream_id = get_chat_manager().get_stream_id(platform, user_id, False)
return await _send_to_target("emoji", emoji_base64, stream_id, "", typing=False, storage_message=storage_message)
async def image_to_group(image_base64: str, group_id: str, platform: str = "qq", storage_message: bool = True) -> bool:
"""向群聊发送图片
Args:
image_base64: 图片的base64编码
group_id: 群聊ID
platform: 平台，默认为"qq"
Returns:
bool: 是否发送成功
"""
stream_id = get_chat_manager().get_stream_id(platform, group_id, True)
return await _send_to_target("image", image_base64, stream_id, "", typing=False, storage_message=storage_message)
async def image_to_user(image_base64: str, user_id: str, platform: str = "qq", storage_message: bool = True) -> bool:
"""向用户发送图片
Args:
image_base64: 图片的base64编码
user_id: 用户ID
platform: 平台，默认为"qq"
Returns:
bool: 是否发送成功
"""
stream_id = get_chat_manager().get_stream_id(platform, user_id, False)
return await _send_to_target("image", image_base64, stream_id, "", typing=False)
async def command_to_group(command: str, group_id: str, platform: str = "qq", storage_message: bool = True) -> bool:
"""向群聊发送命令
Args:
command: 命令
group_id: 群聊ID
platform: 平台，默认为"qq"
Returns:
bool: 是否发送成功
"""
stream_id = get_chat_manager().get_stream_id(platform, group_id, True)
return await _send_to_target("command", command, stream_id, "", typing=False, storage_message=storage_message)
async def command_to_user(command: str, user_id: str, platform: str = "qq", storage_message: bool = True) -> bool:
"""向用户发送命令
Args:
command: 命令
user_id: 用户ID
platform: 平台，默认为"qq"
Returns:
bool: 是否发送成功
"""
stream_id = get_chat_manager().get_stream_id(platform, user_id, False)
return await _send_to_target("command", command, stream_id, "", typing=False, storage_message=storage_message)
# =============================================================================
# 通用发送函数 - 支持任意消息类型
# =============================================================================
async def custom_to_group(
message_type: str,
content: str,
group_id: str,
platform: str = "qq",
display_message: str = "",
typing: bool = False,
reply_to: str = "",
storage_message: bool = True,
) -> bool:
"""向群聊发送自定义类型消息
Args:
message_type: 消息类型，如"text"、"image"、"emoji"、"video"、"file"等
content: 消息内容（通常是base64编码或文本）
group_id: 群聊ID
platform: 平台，默认为"qq"
display_message: 显示消息
typing: 是否显示正在输入
reply_to: 回复消息，格式为"发送者:消息内容"
Returns:
bool: 是否发送成功
"""
stream_id = get_chat_manager().get_stream_id(platform, group_id, True)
return await _send_to_target(
message_type, content, stream_id, display_message, typing, reply_to, storage_message=storage_message
)
async def custom_to_user(
message_type: str,
content: str,
user_id: str,
platform: str = "qq",
display_message: str = "",
typing: bool = False,
reply_to: str = "",
storage_message: bool = True,
) -> bool:
"""向用户发送自定义类型消息
Args:
message_type: 消息类型，如"text"、"image"、"emoji"、"video"、"file"等
content: 消息内容（通常是base64编码或文本）
user_id: 用户ID
platform: 平台，默认为"qq"
display_message: 显示消息
typing: 是否显示正在输入
reply_to: 回复消息，格式为"发送者:消息内容"
Returns:
bool: 是否发送成功
"""
stream_id = get_chat_manager().get_stream_id(platform, user_id, False)
return await _send_to_target(
message_type, content, stream_id, display_message, typing, reply_to, storage_message=storage_message
)
async def custom_message(
message_type: str,
content: str,
target_id: str,
is_group: bool = True,
platform: str = "qq",
display_message: str = "",
typing: bool = False,
reply_to: str = "",
storage_message: bool = True,
) -> bool:
"""发送自定义消息的通用接口
Args:
message_type: 消息类型，如"text"、"image"、"emoji"、"video"、"file"、"audio"等
content: 消息内容
target_id: 目标ID（群ID或用户ID）
is_group: 是否为群聊，True为群聊，False为私聊
platform: 平台，默认为"qq"
display_message: 显示消息
typing: 是否显示正在输入
reply_to: 回复消息，格式为"发送者:消息内容"
Returns:
bool: 是否发送成功
示例:
# 发送视频到群聊
await send_api.custom_message("video", video_base64, "123456", True)
# 发送文件到用户
await send_api.custom_message("file", file_base64, "987654", False)
# 发送音频到群聊并回复特定消息
await send_api.custom_message("audio", audio_base64, "123456", True, reply_to="张三:你好")
"""
stream_id = get_chat_manager().get_stream_id(platform, target_id, is_group)
return await _send_to_target(
message_type, content, stream_id, display_message, typing, reply_to, storage_message=storage_message
)