entari-plugin-chronicle

Entari 聊天记录持久化插件

安装

pip install entari-plugin-chronicle
# or use pdm
pdm add entari-plugin-chronicle
# or use uv
uv add entari-plugin-chronicle

配置

插件提供以下配置选项：

配置项	必填	默认值
record_send	否	False
to_me_only	否	False

示例

在 entari.yml 配置文件中启用插件：

plugins:
  chronicle:
    record_send: true
    to_me_only: false

消息记录模型

插件公开 MessageRecord 模型：

字段	类型	说明
id	int	数据库记录 ID
user_id	int	用户 ID
channel_id	str | None	频道 ID
platform	str	消息平台
message_id	str	平台消息 ID
message	dict	完整消息 JSON
plain_text	str	消息的纯文本内容
type	"message" | "message_sent"	接收消息或 Bot 发送消息
time	datetime	消息时间

可以通过 get_message() 恢复消息链：

record = await get_latest_message_record(user_ids=[user_id])
if record is not None:
    message = record.get_message()

查询接口

可以直接从插件根模块导入以下接口：

from entari_plugin_chronicle import (
    count_message_records,
    get_latest_message_record,
    get_message_records,
    get_messages,
    get_messages_plain_text,
)

通用筛选参数

参数	类型	说明
user_ids	Iterable[int] | None	统一用户 ID；不是平台账号 ID
channel_ids	Iterable[str] | None	频道 ID
platforms	Iterable[str] | None	平台名称
time_start	datetime | None	只查询该时间及之后的记录
time_end	datetime | None	只查询该时间及之前的记录
types	Iterable["message" | "message_sent"] | None	消息记录类型

排序与分页参数

get_message_records、get_messages 和 get_messages_plain_text 还支持：

参数	类型	默认值	说明
order	"asc" | "desc"	"asc"	asc 从旧到新，desc 从新到旧
limit	int | None	None	最多返回多少条；0 返回空结果
offset	int	0	排序后跳过多少条记录

记录会先按筛选条件过滤，再按 time 和 id 排序，最后应用 offset 和 limit

获取消息记录

get_message_records 返回完整的 MessageRecord 对象。

获取某个用户最近一天在指定频道发送或触发的最新 50 条消息记录：

from datetime import datetime, timedelta

from entari_plugin_chronicle import get_message_records

records = await get_message_records(
    user_ids=[user_id],
    channel_ids=[channel_id],
    time_start=datetime.now() - timedelta(days=1),
    order="desc",
    limit=50,
)

for record in records:
    print(record.time, record.type, record.plain_text)

只查询接收消息：

records = await get_message_records(
    user_ids=[user_id],
    types=["message"],
)

获取消息链

get_messages 返回反序列化后的 MessageChain 列表：

from entari_plugin_chronicle import get_messages

messages = await get_messages(
    user_ids=[user_id],
    channel_ids=[channel_id],
    order="desc",
    limit=20,
)

for message in messages:
    print(message)

如果只需要文本内容，优先使用 get_messages_plain_text。该函数只查询数据库中的 纯文本列，不会读取完整消息 JSON：

from entari_plugin_chronicle import get_messages_plain_text

texts = await get_messages_plain_text(
    user_ids=[user_id],
    order="desc",
    limit=20,
)

统计消息记录

from entari_plugin_chronicle import count_message_records

message_count = await count_message_records(
    user_ids=[user_id],
    channel_ids=[channel_id],
    types=["message"],
)

count_message_records 只执行计数查询，不会加载消息记录。

获取最新记录

from entari_plugin_chronicle import get_latest_message_record

latest = await get_latest_message_record(
    user_ids=[user_id],
    channel_ids=[channel_id],
)

if latest is not None:
    print(latest.plain_text)

没有符合条件的记录时返回 None。

分页

页码从 1 开始时，可以这样计算 offset：

page = 2
page_size = 20

records = await get_message_records(
    user_ids=[user_id],
    order="desc",
    limit=page_size,
    offset=(page - 1) * page_size,
)

上面的查询会按时间从新到旧排序，跳过前 20 条，再返回最多 20 条。

与 entari-plugin-user 配合

查询接口中的 user_ids 使用 entari-plugin-user 提供的统一用户 ID。使用 UserSession 时可以直接传入 session.user_id：

from arclet.entari import command
from entari_plugin_user import UserSession

from entari_plugin_chronicle import get_messages_plain_text


@command.on("history")
async def history(session: UserSession):
    texts = await get_messages_plain_text(
        user_ids=[session.user_id],
        channel_ids=[session.channel.id],
        order="desc",
        limit=10,
    )
    await session.send("\n".join(texts) if texts else "暂无聊天记录")

账号绑定到同一个统一用户后，新记录会使用绑定后的统一用户 ID

许可证

MIT License