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

message_api_doc

Browse Source
This commit is contained in:
UnCLAS-Prommer
2025-07-27 22:11:14 +08:00
parent e893b62580
commit 55ce050cc2
2 changed files with 299 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files

529
docs/plugins/api/message-api.md
View File

@@ -1,11 +1,13 @@
# 消息API # 消息API
> 消息API提供了强大的消息查询、计数和格式化功能，让你轻松处理聊天消息数据。 消息API提供了强大的消息查询、计数和格式化功能，让你轻松处理聊天消息数据。
## 导入方式 ## 导入方式
```python ```python
from src.plugin_system.apis import message_api from src.plugin_system.apis import message_api
# 或者
from src.plugin_system import message_api
``` ```
## 功能概述 ## 功能概述
@@ -15,297 +17,356 @@ from src.plugin_system.apis import message_api
- **消息计数** - 统计新消息数量 - **消息计数** - 统计新消息数量
- **消息格式化** - 将消息转换为可读格式 - **消息格式化** - 将消息转换为可读格式
--- ## 主要功能
## 消息查询API ### 1. 按照事件查询消息
```python
def get_messages_by_time(
start_time: float, end_time: float, limit: int = 0, limit_mode: str = "latest", filter_mai: bool = False
) -> List[Dict[str, Any]]:
```
获取指定时间范围内的消息。
### 按时间查询消息 **Args:**
#### `get_messages_by_time(start_time, end_time, limit=0, limit_mode="latest")`
获取指定时间范围内的消息
**参数：**
- `start_time` (float): 开始时间戳 - `start_time` (float): 开始时间戳
- `end_time` (float): 结束时间戳 - `end_time` (float): 结束时间戳
- `limit` (int): 限制返回消息数量，0为不限制 - `limit` (int): 限制返回消息数量，0为不限制
- `limit_mode` (str): 限制模式，`"earliest"`获取最早记录，`"latest"`获取最新记录 - `limit_mode` (str): 限制模式，`"earliest"`获取最早记录，`"latest"`获取最新记录
- `filter_mai` (bool): 是否过滤掉机器人的消息，默认False
**返回：** `List[Dict[str, Any]]` - 消息列表 **Returns:**
- `List[Dict[str, Any]]` - 消息列表
**示例：** 消息列表中包含的键与`Messages`类的属性一致。（位于`src.common.database.database_model`）
### 2. 获取指定聊天中指定时间范围内的信息
```python ```python
import time def get_messages_by_time_in_chat(
chat_id: str,
# 获取最近24小时的消息 start_time: float,
now = time.time() end_time: float,
yesterday = now - 24 * 3600 limit: int = 0,
messages = message_api.get_messages_by_time(yesterday, now, limit=50) limit_mode: str = "latest",
filter_mai: bool = False,
) -> List[Dict[str, Any]]:
``` ```
获取指定聊天中指定时间范围内的消息。
### 按聊天查询消息 **Args:**
#### `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 - `chat_id` (str): 聊天ID
- `start_time` (float): 开始时间戳 - `start_time` (float): 开始时间戳
- `end_time` (float): 结束时间戳 - `end_time` (float): 结束时间戳
- `person_ids` (list): 用户ID列表 - `limit` (int): 限制返回消息数量，0为不限制
- `limit` (int): 限制返回消息数量 - `limit_mode` (str): 限制模式，`"earliest"`获取最早记录，`"latest"`获取最新记录
- `limit_mode` (str): 限制模式 - `filter_mai` (bool): 是否过滤掉机器人的消息，默认False
**示例：** **Returns:**
- `List[Dict[str, Any]]` - 消息列表
### 3. 获取指定聊天中指定时间范围内的信息（包含边界）
```python ```python
# 获取特定用户的消息 def get_messages_by_time_in_chat_inclusive(
user_messages = message_api.get_messages_by_time_in_chat_for_users( chat_id: str,
chat_id="123456789", start_time: float,
start_time=yesterday, end_time: float,
end_time=now, limit: int = 0,
person_ids=["user1", "user2"] limit_mode: str = "latest",
) filter_mai: bool = False,
filter_command: bool = False,
) -> List[Dict[str, Any]]:
``` ```
获取指定聊天中指定时间范围内的消息（包含边界）。
#### `get_messages_by_time_for_users(start_time, end_time, person_ids, limit=0, limit_mode="latest")` **Args:**
- `chat_id` (str): 聊天ID
- `start_time` (float): 开始时间戳（包含）
- `end_time` (float): 结束时间戳（包含）
- `limit` (int): 限制返回消息数量，0为不限制
- `limit_mode` (str): 限制模式，`"earliest"`获取最早记录，`"latest"`获取最新记录
- `filter_mai` (bool): 是否过滤掉机器人的消息，默认False
- `filter_command` (bool): 是否过滤命令消息，默认False
获取指定用户在所有聊天中指定时间范围内的消息 **Returns:**
- `List[Dict[str, Any]]` - 消息列表
### 其他查询方法
#### `get_random_chat_messages(start_time, end_time, limit=0, limit_mode="latest")` ### 4. 获取指定聊天中指定用户在指定时间范围内的消息
```python
def get_messages_by_time_in_chat_for_users(
chat_id: str,
start_time: float,
end_time: float,
person_ids: List[str],
limit: int = 0,
limit_mode: str = "latest",
) -> List[Dict[str, Any]]:
```
获取指定聊天中指定用户在指定时间范围内的消息。
随机选择一个聊天，返回该聊天在指定时间范围内的消息 **Args:**
#### `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 - `chat_id` (str): 聊天ID
- `start_time` (float): 开始时间戳 - `start_time` (float): 开始时间戳
- `end_time` (float): 结束时间戳，如果为None则使用当前时间 - `end_time` (float): 结束时间戳
- `person_ids` (List[str]): 用户ID列表
- `limit` (int): 限制返回消息数量，0为不限制
- `limit_mode` (str): 限制模式，`"earliest"`获取最早记录，`"latest"`获取最新记录
**返回：** `int` - 新消息数量 **Returns:**
- `List[Dict[str, Any]]` - 消息列表
**示例：**
### 5. 随机选择一个聊天，返回该聊天在指定时间范围内的消息
```python ```python
# 计算最近1小时的新消息数 def get_random_chat_messages(
import time start_time: float,
now = time.time() end_time: float,
hour_ago = now - 3600 limit: int = 0,
new_count = message_api.count_new_messages("123456789", hour_ago, now) limit_mode: str = "latest",
print(f"最近1小时有{new_count}条新消息") filter_mai: bool = False,
) -> List[Dict[str, Any]]:
``` ```
随机选择一个聊天，返回该聊天在指定时间范围内的消息。
### `count_new_messages_for_users(chat_id, start_time, end_time, person_ids)` **Args:**
- `start_time` (float): 开始时间戳
- `end_time` (float): 结束时间戳
- `limit` (int): 限制返回消息数量，0为不限制
- `limit_mode` (str): 限制模式，`"earliest"`获取最早记录，`"latest"`获取最新记录
- `filter_mai` (bool): 是否过滤掉机器人的消息，默认False
计算指定聊天中指定用户从开始时间到结束时间的新消息数量 **Returns:**
- `List[Dict[str, Any]]` - 消息列表
---
## 消息格式化API ### 6. 获取指定用户在所有聊天中指定时间范围内的消息
```python
def get_messages_by_time_for_users(
start_time: float,
end_time: float,
person_ids: List[str],
limit: int = 0,
limit_mode: str = "latest",
) -> List[Dict[str, Any]]:
```
获取指定用户在所有聊天中指定时间范围内的消息。
### `build_readable_messages_to_str(messages, **options)` **Args:**
- `start_time` (float): 开始时间戳
- `end_time` (float): 结束时间戳
- `person_ids` (List[str]): 用户ID列表
- `limit` (int): 限制返回消息数量，0为不限制
- `limit_mode` (str): 限制模式，`"earliest"`获取最早记录，`"latest"`获取最新记录
将消息列表构建成可读的字符串 **Returns:**
- `List[Dict[str, Any]]` - 消息列表
**参数：**
### 7. 获取指定时间戳之前的消息
```python
def get_messages_before_time(
timestamp: float,
limit: int = 0,
filter_mai: bool = False,
) -> List[Dict[str, Any]]:
```
获取指定时间戳之前的消息。
**Args:**
- `timestamp` (float): 时间戳
- `limit` (int): 限制返回消息数量，0为不限制
- `filter_mai` (bool): 是否过滤掉机器人的消息，默认False
**Returns:**
- `List[Dict[str, Any]]` - 消息列表
### 8. 获取指定聊天中指定时间戳之前的消息
```python
def get_messages_before_time_in_chat(
chat_id: str,
timestamp: float,
limit: int = 0,
filter_mai: bool = False,
) -> List[Dict[str, Any]]:
```
获取指定聊天中指定时间戳之前的消息。
**Args:**
- `chat_id` (str): 聊天ID
- `timestamp` (float): 时间戳
- `limit` (int): 限制返回消息数量，0为不限制
- `filter_mai` (bool): 是否过滤掉机器人的消息，默认False
**Returns:**
- `List[Dict[str, Any]]` - 消息列表
### 9. 获取指定用户在指定时间戳之前的消息
```python
def get_messages_before_time_for_users(
timestamp: float,
person_ids: List[str],
limit: int = 0,
) -> List[Dict[str, Any]]:
```
获取指定用户在指定时间戳之前的消息。
**Args:**
- `timestamp` (float): 时间戳
- `person_ids` (List[str]): 用户ID列表
- `limit` (int): 限制返回消息数量，0为不限制
**Returns:**
- `List[Dict[str, Any]]` - 消息列表
### 10. 获取指定聊天中最近一段时间的消息
```python
def get_recent_messages(
chat_id: str,
hours: float = 24.0,
limit: int = 100,
limit_mode: str = "latest",
filter_mai: bool = False,
) -> List[Dict[str, Any]]:
```
获取指定聊天中最近一段时间的消息。
**Args:**
- `chat_id` (str): 聊天ID
- `hours` (float): 最近多少小时，默认24小时
- `limit` (int): 限制返回消息数量，默认100条
- `limit_mode` (str): 限制模式，`"earliest"`获取最早记录，`"latest"`获取最新记录
- `filter_mai` (bool): 是否过滤掉机器人的消息，默认False
**Returns:**
- `List[Dict[str, Any]]` - 消息列表
### 11. 计算指定聊天中从开始时间到结束时间的新消息数量
```python
def count_new_messages(
chat_id: str,
start_time: float = 0.0,
end_time: Optional[float] = None,
) -> int:
```
计算指定聊天中从开始时间到结束时间的新消息数量。
**Args:**
- `chat_id` (str): 聊天ID
- `start_time` (float): 开始时间戳
- `end_time` (Optional[float]): 结束时间戳，如果为None则使用当前时间
**Returns:**
- `int` - 新消息数量
### 12. 计算指定聊天中指定用户从开始时间到结束时间的新消息数量
```python
def count_new_messages_for_users(
chat_id: str,
start_time: float,
end_time: float,
person_ids: List[str],
) -> int:
```
计算指定聊天中指定用户从开始时间到结束时间的新消息数量。
**Args:**
- `chat_id` (str): 聊天ID
- `start_time` (float): 开始时间戳
- `end_time` (float): 结束时间戳
- `person_ids` (List[str]): 用户ID列表
**Returns:**
- `int` - 新消息数量
### 13. 将消息列表构建成可读的字符串
```python
def build_readable_messages_to_str(
messages: List[Dict[str, Any]],
replace_bot_name: bool = True,
merge_messages: bool = False,
timestamp_mode: str = "relative",
read_mark: float = 0.0,
truncate: bool = False,
show_actions: bool = False,
) -> str:
```
将消息列表构建成可读的字符串。
**Args:**
- `messages` (List[Dict[str, Any]]): 消息列表 - `messages` (List[Dict[str, Any]]): 消息列表
- `replace_bot_name` (bool): 是否将机器人的名称替换为"你"，默认True - `replace_bot_name` (bool): 是否将机器人的名称替换为"你"
- `merge_messages` (bool): 是否合并连续消息，默认False - `merge_messages` (bool): 是否合并连续消息
- `timestamp_mode` (str): 时间戳显示模式，`"relative"`或`"absolute"`，默认`"relative"` - `timestamp_mode` (str): 时间戳显示模式，`"relative"`或`"absolute"`
- `read_mark` (float): 已读标记时间戳，用于分割已读和未读消息，默认0.0 - `read_mark` (float): 已读标记时间戳，用于分割已读和未读消息
- `truncate` (bool): 是否截断长消息，默认False - `truncate` (bool): 是否截断长消息
- `show_actions` (bool): 是否显示动作记录，默认False - `show_actions` (bool): 是否显示动作记录
**返回：** `str` - 格式化后的可读字符串 **Returns:**
- `str` - 格式化后的可读字符串
**示例：**
### 14. 将消息列表构建成可读的字符串，并返回详细信息
```python ```python
# 获取消息并格式化为可读文本 async def build_readable_messages_with_details(
messages = message_api.get_recent_messages("123456789", hours=2) messages: List[Dict[str, Any]],
readable_text = message_api.build_readable_messages_to_str( replace_bot_name: bool = True,
messages, merge_messages: bool = False,
replace_bot_name=True, timestamp_mode: str = "relative",
merge_messages=True, truncate: bool = False,
timestamp_mode="relative" ) -> Tuple[str, List[Tuple[float, str, str]]]:
)
print(readable_text)
``` ```
将消息列表构建成可读的字符串，并返回详细信息。
### `build_readable_messages_with_details(messages, **options)` 异步 **Args:**
- `messages` (List[Dict[str, Any]]): 消息列表
- `replace_bot_name` (bool): 是否将机器人的名称替换为"你"
- `merge_messages` (bool): 是否合并连续消息
- `timestamp_mode` (str): 时间戳显示模式，`"relative"`或`"absolute"`
- `truncate` (bool): 是否截断长消息
将消息列表构建成可读的字符串，并返回详细信息 **Returns:**
- `Tuple[str, List[Tuple[float, str, str]]]` - 格式化后的可读字符串和详细信息元组列表(时间戳, 昵称, 内容)
**参数：** 与 `build_readable_messages_to_str` 类似，但不包含 `read_mark` 和 `show_actions`
**返回：** `Tuple[str, List[Tuple[float, str, str]]]` - 格式化字符串和详细信息元组列表(时间戳, 昵称, 内容) ### 15. 从消息列表中提取不重复的用户ID列表
**示例：**
```python ```python
# 异步获取详细格式化信息 async def get_person_ids_from_messages(
readable_text, details = await message_api.build_readable_messages_with_details( messages: List[Dict[str, Any]],
messages, ) -> List[str]:
timestamp_mode="absolute"
)
for timestamp, nickname, content in details:
print(f"{timestamp}: {nickname} 说: {content}")
``` ```
从消息列表中提取不重复的用户ID列表。
### `get_person_ids_from_messages(messages)` 异步 **Args:**
从消息列表中提取不重复的用户ID列表
**参数：**
- `messages` (List[Dict[str, Any]]): 消息列表 - `messages` (List[Dict[str, Any]]): 消息列表
**返回：** `List[str]` - 用户ID列表 **Returns:**
- `List[str]` - 用户ID列表
**示例：**
### 16. 从消息列表中移除机器人的消息
```python ```python
# 获取参与对话的所有用户ID def filter_mai_messages(
messages = message_api.get_recent_messages("123456789") messages: List[Dict[str, Any]],
person_ids = await message_api.get_person_ids_from_messages(messages) ) -> List[Dict[str, Any]]:
print(f"参与对话的用户: {person_ids}")
``` ```
从消息列表中移除机器人的消息。
--- **Args:**
- `messages` (List[Dict[str, Any]]): 消息列表，每个元素是消息字典
## 完整使用示例 **Returns:**
- `List[Dict[str, Any]]` - 过滤后的消息列表
### 场景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类型） 1. **时间戳格式**：所有时间参数都使用Unix时间戳（float类型）
2. **异步函数**：`build_readable_messages_with_details` 和 `get_person_ids_from_messages` 是异步函数，需要使用 `await` 2. **异步函数**：部分函数是异步函数，需要使用 `await`
3. **性能考虑**：查询大量消息时建议设置合理的 `limit` 参数 3. **性能考虑**：查询大量消息时建议设置合理的 `limit` 参数
4. **消息格式**：返回的消息是字典格式，包含时间戳、发送者、内容等信息 4. **消息格式**：返回的消息是字典格式，包含时间戳、发送者、内容等信息
5. **用户ID**：`person_ids` 参数接受字符串列表，用于筛选特定用户的消息 5. **用户ID**：`person_ids` 参数接受字符串列表，用于筛选特定用户的消息

6
src/plugin_system/apis/message_api.py
View File

@@ -207,7 +207,7 @@ def get_random_chat_messages(
def get_messages_by_time_for_users( def get_messages_by_time_for_users(
start_time: float, end_time: float, person_ids: list, limit: int = 0, limit_mode: str = "latest" start_time: float, end_time: float, person_ids: List[str], limit: int = 0, limit_mode: str = "latest"
) -> List[Dict[str, Any]]: ) -> List[Dict[str, Any]]:
""" """
获取指定用户在所有聊天中指定时间范围内的消息 获取指定用户在所有聊天中指定时间范围内的消息
@@ -287,7 +287,7 @@ def get_messages_before_time_in_chat(
return get_raw_msg_before_timestamp_with_chat(chat_id, timestamp, limit) return get_raw_msg_before_timestamp_with_chat(chat_id, timestamp, limit)
def get_messages_before_time_for_users(timestamp: float, person_ids: list, limit: int = 0) -> List[Dict[str, Any]]: def get_messages_before_time_for_users(timestamp: float, person_ids: List[str], limit: int = 0) -> List[Dict[str, Any]]:
""" """
获取指定用户在指定时间戳之前的消息 获取指定用户在指定时间戳之前的消息
@@ -372,7 +372,7 @@ def count_new_messages(chat_id: str, start_time: float = 0.0, end_time: Optional
return num_new_messages_since(chat_id, start_time, end_time) return num_new_messages_since(chat_id, start_time, end_time)
def count_new_messages_for_users(chat_id: str, start_time: float, end_time: float, person_ids: list) -> int: def count_new_messages_for_users(chat_id: str, start_time: float, end_time: float, person_ids: List[str]) -> int:
""" """
计算指定聊天中指定用户从开始时间到结束时间的新消息数量 计算指定聊天中指定用户从开始时间到结束时间的新消息数量