From 867b78a11cf850d02e5570a55eade4042bd76a09 Mon Sep 17 00:00:00 2001 From: SXP-Simon Date: Sat, 21 Feb 2026 16:07:28 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=B0=9D=E8=AF=95=E5=88=86=E9=A1=B5?= =?UTF-8?q?=E6=8B=89=E5=8F=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../platform/adapters/onebot_adapter.py | 101 +++++++++++++----- 1 file changed, 77 insertions(+), 24 deletions(-) diff --git a/src/infrastructure/platform/adapters/onebot_adapter.py b/src/infrastructure/platform/adapters/onebot_adapter.py index 04cc4a9..2c699f8 100644 --- a/src/infrastructure/platform/adapters/onebot_adapter.py +++ b/src/infrastructure/platform/adapters/onebot_adapter.py @@ -4,6 +4,7 @@ OneBot v11 平台适配器 支持 NapCat、go-cqhttp、Lagrange 及其他 OneBot 实现。 """ +import asyncio import base64 from datetime import datetime, timedelta from typing import Any @@ -80,12 +81,13 @@ class OneBotAdapter(PlatformAdapter): ) -> list[UnifiedMessage]: """ 从 OneBot 后端拉取群组历史消息。 + 采用分页拉取策略(参考 portrayal 插件),减少 NapCat/go-cqhttp 单次请求的 CPU 和内存负担。 Args: group_id (str): 群号 days (int): 拉取过去几天的消息 max_count (int): 最大拉取条数 - before_id (str, optional): 锚点消息 ID(部分后端支持) + before_id (str, optional): 锚点消息 ID,用于分页回溯 Returns: list[UnifiedMessage]: 统一格式的消息列表 @@ -94,41 +96,92 @@ class OneBotAdapter(PlatformAdapter): return [] try: - # 调用 OneBot 标准 API: get_group_msg_history - result = await self.bot.call_action( - "get_group_msg_history", - group_id=int(group_id), - count=max_count, - ) - - if not result or "messages" not in result: - return [] + chunk_size = 100 # 每次拉取 100 条,较为稳健 + all_raw_messages = [] end_time = datetime.now() start_time = end_time - timedelta(days=days) + start_timestamp = int(start_time.timestamp()) - messages = [] - for raw_msg in result.get("messages", []): - msg_time = datetime.fromtimestamp(raw_msg.get("time", 0)) - # 时间范围过滤 - if not (start_time <= msg_time <= end_time): - continue + # 使用 message_seq (在 NapCat 中通常可用 message_id 作为 seq 参数) + # 进行分页回溯拉取 + current_anchor_id = before_id - # 身份过滤(排除机器人自己) - sender_id = str(raw_msg.get("sender", {}).get("user_id", "")) - if sender_id in self.bot_self_ids: + logger.info( + f"OneBot 开始分页回溯拉取消息: 群 {group_id}, 时间限制 {days}天, 数量限制 {max_count}" + ) + + while len(all_raw_messages) < max_count: + fetch_count = min(chunk_size, max_count - len(all_raw_messages)) + + params = { + "group_id": int(group_id), + "count": fetch_count, + "reverseOrder": True, # 关键:协助分页向上回退拉取历史 + } + + if current_anchor_id: + params["message_seq"] = current_anchor_id + + result = await self.bot.call_action("get_group_msg_history", **params) + + if not result or "messages" not in result: + break + + messages = result.get("messages", []) + if not messages: + break + + # 时间过滤与列表合并 + # 消息通常是按时间正序排列的,即 messages[0] 最旧,messages[-1] 最新 + chunk_earliest_msg = messages[0] + chunk_earliest_time = chunk_earliest_msg.get("time", 0) + + for raw_msg in messages: + msg_time = raw_msg.get("time", 0) + + # 身份过滤(排除机器人自己) + sender_id = str(raw_msg.get("sender", {}).get("user_id", "")) + if sender_id in self.bot_self_ids: + continue + + # 时间范围判定 + if start_timestamp <= msg_time <= int(end_time.timestamp()): + all_raw_messages.append(raw_msg) + + # 更新锚点为该批次最旧的消息 ID,用于下一次回溯请求 + current_anchor_id = chunk_earliest_msg.get("message_id") + + # 如果该批次最旧的消息已经早于 start_time,或者拉取的数量不足,说明已到达回溯终点 + if chunk_earliest_time < start_timestamp or len(messages) < fetch_count: + break + + # 稍微延迟,减缓服务端压力(避免 NapCat 瞬间 CPU 飙升) + await asyncio.sleep(0.05) + + # 统一转换为 UnifiedMessage 并在返回前去重排序 + unified_messages = [] + seen_ids = set() + for raw_msg in all_raw_messages: + mid = str(raw_msg.get("message_id", "")) + if not mid or mid in seen_ids: continue unified = self._convert_message(raw_msg, group_id) if unified: - messages.append(unified) + unified_messages.append(unified) + seen_ids.add(mid) - # 确保按时间顺序排列 - messages.sort(key=lambda m: m.timestamp) - return messages + # 确保最终结果符合时间顺序 + unified_messages.sort(key=lambda m: m.timestamp) + + logger.info( + f"OneBot 分页拉取完成: 共处理 {len(all_raw_messages)} 条原始消息, 最终有效 {len(unified_messages)} 条" + ) + return unified_messages except Exception as e: - logger.warning(f"OneBot 获取消息失败: {e}") + logger.warning(f"OneBot 分页获取消息失败: {e}") return [] def _convert_message(self, raw_msg: dict, group_id: str) -> UnifiedMessage | None: