311 lines
8.5 KiB
Markdown
311 lines
8.5 KiB
Markdown
# 消息API
|
||
|
||
> 消息API提供了强大的消息查询、计数和格式化功能，让你轻松处理聊天消息数据。
|
||
|
||
## 导入方式
|
||
|
||
```python
|
||
from src.plugin_system.apis import message_api
|
||
```
|
||
|
||
## 功能概述
|
||
|
||
消息API主要提供三大类功能：
|
||
- **消息查询** - 按时间、聊天、用户等条件查询消息
|
||
- **消息计数** - 统计新消息数量
|
||
- **消息格式化** - 将消息转换为可读格式
|
||
|
||
---
|
||
|
||
## 消息查询API
|
||
|
||
### 按时间查询消息
|
||
|
||
#### `get_messages_by_time(start_time, end_time, limit=0, limit_mode="latest")`
|
||
|
||
获取指定时间范围内的消息
|
||
|
||
**参数：**
|
||
- `start_time` (float): 开始时间戳
|
||
- `end_time` (float): 结束时间戳
|
||
- `limit` (int): 限制返回消息数量，0为不限制
|
||
- `limit_mode` (str): 限制模式，`"earliest"`获取最早记录，`"latest"`获取最新记录
|
||
|
||
**返回：** `List[Dict[str, Any]]` - 消息列表
|
||
|
||
**示例：**
|
||
```python
|
||
import time
|
||
|
||
# 获取最近24小时的消息
|
||
now = time.time()
|
||
yesterday = now - 24 * 3600
|
||
messages = message_api.get_messages_by_time(yesterday, now, limit=50)
|
||
```
|
||
|
||
### 按聊天查询消息
|
||
|
||
#### `get_messages_by_time_in_chat(chat_id, start_time, end_time, limit=0, limit_mode="latest")`
|
||
|
||
获取指定聊天中指定时间范围内的消息
|
||
|
||
**参数：**
|
||
- `chat_id` (str): 聊天ID
|
||
- 其他参数同上
|
||
|
||
**示例：**
|
||
```python
|
||
# 获取某个群聊最近的100条消息
|
||
messages = message_api.get_messages_by_time_in_chat(
|
||
chat_id="123456789",
|
||
start_time=yesterday,
|
||
end_time=now,
|
||
limit=100
|
||
)
|
||
```
|
||
|
||
#### `get_messages_by_time_in_chat_inclusive(chat_id, start_time, end_time, limit=0, limit_mode="latest")`
|
||
|
||
获取指定聊天中指定时间范围内的消息（包含边界时间点）
|
||
|
||
与 `get_messages_by_time_in_chat` 类似，但包含边界时间戳的消息。
|
||
|
||
#### `get_recent_messages(chat_id, hours=24.0, limit=100, limit_mode="latest")`
|
||
|
||
获取指定聊天中最近一段时间的消息（便捷方法）
|
||
|
||
**参数：**
|
||
- `chat_id` (str): 聊天ID
|
||
- `hours` (float): 最近多少小时，默认24小时
|
||
- `limit` (int): 限制返回消息数量，默认100条
|
||
- `limit_mode` (str): 限制模式
|
||
|
||
**示例：**
|
||
```python
|
||
# 获取最近6小时的消息
|
||
recent_messages = message_api.get_recent_messages(
|
||
chat_id="123456789",
|
||
hours=6.0,
|
||
limit=50
|
||
)
|
||
```
|
||
|
||
### 按用户查询消息
|
||
|
||
#### `get_messages_by_time_in_chat_for_users(chat_id, start_time, end_time, person_ids, limit=0, limit_mode="latest")`
|
||
|
||
获取指定聊天中指定用户在指定时间范围内的消息
|
||
|
||
**参数：**
|
||
- `chat_id` (str): 聊天ID
|
||
- `start_time` (float): 开始时间戳
|
||
- `end_time` (float): 结束时间戳
|
||
- `person_ids` (list): 用户ID列表
|
||
- `limit` (int): 限制返回消息数量
|
||
- `limit_mode` (str): 限制模式
|
||
|
||
**示例：**
|
||
```python
|
||
# 获取特定用户的消息
|
||
user_messages = message_api.get_messages_by_time_in_chat_for_users(
|
||
chat_id="123456789",
|
||
start_time=yesterday,
|
||
end_time=now,
|
||
person_ids=["user1", "user2"]
|
||
)
|
||
```
|
||
|
||
#### `get_messages_by_time_for_users(start_time, end_time, person_ids, limit=0, limit_mode="latest")`
|
||
|
||
获取指定用户在所有聊天中指定时间范围内的消息
|
||
|
||
### 其他查询方法
|
||
|
||
#### `get_random_chat_messages(start_time, end_time, limit=0, limit_mode="latest")`
|
||
|
||
随机选择一个聊天，返回该聊天在指定时间范围内的消息
|
||
|
||
#### `get_messages_before_time(timestamp, limit=0)`
|
||
|
||
获取指定时间戳之前的消息
|
||
|
||
#### `get_messages_before_time_in_chat(chat_id, timestamp, limit=0)`
|
||
|
||
获取指定聊天中指定时间戳之前的消息
|
||
|
||
#### `get_messages_before_time_for_users(timestamp, person_ids, limit=0)`
|
||
|
||
获取指定用户在指定时间戳之前的消息
|
||
|
||
---
|
||
|
||
## 消息计数API
|
||
|
||
### `count_new_messages(chat_id, start_time=0.0, end_time=None)`
|
||
|
||
计算指定聊天中从开始时间到结束时间的新消息数量
|
||
|
||
**参数：**
|
||
- `chat_id` (str): 聊天ID
|
||
- `start_time` (float): 开始时间戳
|
||
- `end_time` (float): 结束时间戳，如果为None则使用当前时间
|
||
|
||
**返回：** `int` - 新消息数量
|
||
|
||
**示例：**
|
||
```python
|
||
# 计算最近1小时的新消息数
|
||
import time
|
||
now = time.time()
|
||
hour_ago = now - 3600
|
||
new_count = message_api.count_new_messages("123456789", hour_ago, now)
|
||
print(f"最近1小时有{new_count}条新消息")
|
||
```
|
||
|
||
### `count_new_messages_for_users(chat_id, start_time, end_time, person_ids)`
|
||
|
||
计算指定聊天中指定用户从开始时间到结束时间的新消息数量
|
||
|
||
---
|
||
|
||
## 消息格式化API
|
||
|
||
### `build_readable_messages_to_str(messages, **options)`
|
||
|
||
将消息列表构建成可读的字符串
|
||
|
||
**参数：**
|
||
- `messages` (List[Dict[str, Any]]): 消息列表
|
||
- `replace_bot_name` (bool): 是否将机器人的名称替换为"你"，默认True
|
||
- `merge_messages` (bool): 是否合并连续消息，默认False
|
||
- `timestamp_mode` (str): 时间戳显示模式，`"relative"`或`"absolute"`，默认`"relative"`
|
||
- `read_mark` (float): 已读标记时间戳，用于分割已读和未读消息，默认0.0
|
||
- `truncate` (bool): 是否截断长消息，默认False
|
||
- `show_actions` (bool): 是否显示动作记录，默认False
|
||
|
||
**返回：** `str` - 格式化后的可读字符串
|
||
|
||
**示例：**
|
||
```python
|
||
# 获取消息并格式化为可读文本
|
||
messages = message_api.get_recent_messages("123456789", hours=2)
|
||
readable_text = message_api.build_readable_messages_to_str(
|
||
messages,
|
||
replace_bot_name=True,
|
||
merge_messages=True,
|
||
timestamp_mode="relative"
|
||
)
|
||
print(readable_text)
|
||
```
|
||
|
||
### `build_readable_messages_with_details(messages, **options)` 异步
|
||
|
||
将消息列表构建成可读的字符串，并返回详细信息
|
||
|
||
**参数：** 与 `build_readable_messages_to_str` 类似，但不包含 `read_mark` 和 `show_actions`
|
||
|
||
**返回：** `Tuple[str, List[Tuple[float, str, str]]]` - 格式化字符串和详细信息元组列表(时间戳, 昵称, 内容)
|
||
|
||
**示例：**
|
||
```python
|
||
# 异步获取详细格式化信息
|
||
readable_text, details = await message_api.build_readable_messages_with_details(
|
||
messages,
|
||
timestamp_mode="absolute"
|
||
)
|
||
|
||
for timestamp, nickname, content in details:
|
||
print(f"{timestamp}: {nickname} 说: {content}")
|
||
```
|
||
|
||
### `get_person_ids_from_messages(messages)` 异步
|
||
|
||
从消息列表中提取不重复的用户ID列表
|
||
|
||
**参数：**
|
||
- `messages` (List[Dict[str, Any]]): 消息列表
|
||
|
||
**返回：** `List[str]` - 用户ID列表
|
||
|
||
**示例：**
|
||
```python
|
||
# 获取参与对话的所有用户ID
|
||
messages = message_api.get_recent_messages("123456789")
|
||
person_ids = await message_api.get_person_ids_from_messages(messages)
|
||
print(f"参与对话的用户: {person_ids}")
|
||
```
|
||
|
||
---
|
||
|
||
## 完整使用示例
|
||
|
||
### 场景1：统计活跃度
|
||
|
||
```python
|
||
import time
|
||
from src.plugin_system.apis import message_api
|
||
|
||
async def analyze_chat_activity(chat_id: str):
|
||
"""分析聊天活跃度"""
|
||
now = time.time()
|
||
day_ago = now - 24 * 3600
|
||
|
||
# 获取最近24小时的消息
|
||
messages = message_api.get_recent_messages(chat_id, hours=24)
|
||
|
||
# 统计消息数量
|
||
total_count = len(messages)
|
||
|
||
# 获取参与用户
|
||
person_ids = await message_api.get_person_ids_from_messages(messages)
|
||
|
||
# 格式化消息内容
|
||
readable_text = message_api.build_readable_messages_to_str(
|
||
messages[-10:], # 最后10条消息
|
||
merge_messages=True,
|
||
timestamp_mode="relative"
|
||
)
|
||
|
||
return {
|
||
"total_messages": total_count,
|
||
"active_users": len(person_ids),
|
||
"recent_chat": readable_text
|
||
}
|
||
```
|
||
|
||
### 场景2：查看特定用户的历史消息
|
||
|
||
```python
|
||
def get_user_history(chat_id: str, user_id: str, days: int = 7):
|
||
"""获取用户最近N天的消息历史"""
|
||
now = time.time()
|
||
start_time = now - days * 24 * 3600
|
||
|
||
# 获取特定用户的消息
|
||
user_messages = message_api.get_messages_by_time_in_chat_for_users(
|
||
chat_id=chat_id,
|
||
start_time=start_time,
|
||
end_time=now,
|
||
person_ids=[user_id],
|
||
limit=100
|
||
)
|
||
|
||
# 格式化为可读文本
|
||
readable_history = message_api.build_readable_messages_to_str(
|
||
user_messages,
|
||
replace_bot_name=False,
|
||
timestamp_mode="absolute"
|
||
)
|
||
|
||
return readable_history
|
||
```
|
||
|
||
---
|
||
|
||
## 注意事项
|
||
|
||
1. **时间戳格式**：所有时间参数都使用Unix时间戳（float类型）
|
||
2. **异步函数**：`build_readable_messages_with_details` 和 `get_person_ids_from_messages` 是异步函数，需要使用 `await`
|
||
3. **性能考虑**：查询大量消息时建议设置合理的 `limit` 参数
|
||
4. **消息格式**：返回的消息是字典格式，包含时间戳、发送者、内容等信息
|
||
5. **用户ID**：`person_ids` 参数接受字符串列表，用于筛选特定用户的消息 |