From cd6a6294e1ae4440321cfcd2ab16669e22311917 Mon Sep 17 00:00:00 2001 From: SXP-Simon Date: Fri, 20 Mar 2026 23:59:24 +0800 Subject: [PATCH] =?UTF-8?q?feat(=E5=A2=9E=E9=87=8F=E5=88=86=E6=9E=90):=20?= =?UTF-8?q?=E6=9B=B4=E6=96=B0=E5=A2=9E=E9=87=8F=E5=88=86=E6=9E=90=E9=85=8D?= =?UTF-8?q?=E7=BD=AE=EF=BC=8C=E5=A2=9E=E5=8A=A0=E5=AE=89=E5=85=A8=E4=B8=8A?= =?UTF-8?q?=E9=99=90=E5=B9=B6=E4=BC=98=E5=8C=96=E6=B6=88=E6=81=AF=E6=8B=89?= =?UTF-8?q?=E5=8F=96=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- _conf_schema.json | 12 +++--- .../services/analysis_application_service.py | 31 +++++++++----- src/domain/repositories/message_repository.py | 2 + src/infrastructure/config/config_manager.py | 6 +-- .../platform/adapters/discord_adapter.py | 8 +++- .../platform/adapters/onebot_adapter.py | 41 +++++++++++++------ .../platform/adapters/telegram_adapter.py | 7 +++- 7 files changed, 72 insertions(+), 35 deletions(-) diff --git a/_conf_schema.json b/_conf_schema.json index afe7f57..188981b 100644 --- a/_conf_schema.json +++ b/_conf_schema.json @@ -258,17 +258,17 @@ "default": 8, "hint": "一天内最多执行多少次增量分析。达到上限后当天不再触发新的增量分析，但仍会在报告时间生成最终报告。建议4-12次。" }, - "incremental_max_messages": { + "incremental_safe_limit": { "type": "int", - "description": "单次增量分析最大消息数", - "default": 300, - "hint": "每次增量分析最多处理的消息条数。数值越大单次分析越全面但 Token 消耗越高。因为插件的逻辑是记录旧一次拉取获取到的最晚消息，只会处理后续的新消息，所以建议允许根据群聊实际情况设定的越大越好。" + "description": "增量分析单次安全上限 (Safe Count)", + "default": 2000, + "hint": "增量分析模式下，当群聊极其活跃导致大量消息堆积时，插件单次回溯历史的最大消息条数。这同时也是为了防止特殊情况下无限回溯。即使没追到上次分析的断点，拉满此数值也会强制停止。建议 1000-5000。" }, "incremental_min_messages": { "type": "int", "description": "增量分析最小消息数阈值", - "default": 100, - "hint": "当自上次分析以来的新消息数低于此阈值时，跳过本次增量分析以节省 Token。可能导致少量消息的群聊分析结果不完整，但是依旧建议设置较大值，因为当获取到的消息不够时，最晚消息的游标不会更新，可以等消息足够后的增量分析进行分析。" + "default": 300, + "hint": "当自上次分析以来的新消息数低于此阈值时，跳过本次增量分析以节省 Token。建议设置较大值，因为当获取到的消息不够时，最晚消息的游标不会更新，等消息累计足够后的增量分析会进行分析。" }, "incremental_topics_per_batch": { "type": "int", diff --git a/src/application/services/analysis_application_service.py b/src/application/services/analysis_application_service.py index 94a19fa..e7dbfbc 100644 --- a/src/application/services/analysis_application_service.py +++ b/src/application/services/analysis_application_service.py @@ -287,16 +287,24 @@ class AnalysisApplicationService: if not adapter: raise ValueError(f"未找到平台 {platform_id} 的适配器") - # 2. 拉取消息（使用增量配置的消息数量上限） + # 2. 拉取消息，获取进度并确定拉取量 + last_analyzed_ts = await self.incremental_store.get_last_analyzed_timestamp( + group_id + ) days = self.config_manager.get_analysis_days() - max_count = self.config_manager.get_incremental_max_messages() + # 在增量模式下，拉取上限由安全限制 (Safe Count) 统一控制，确保能追平进度且不溢出 + max_count = self.config_manager.get_incremental_safe_limit() + # 3. 拉取消息（优先从上次进度点开始回溯，确保不遗漏高活跃期间的 Gap） raw_messages = await adapter.fetch_messages( - group_id=group_id, days=days, max_count=max_count + group_id=group_id, + days=days, + max_count=max_count, + since_ts=last_analyzed_ts, ) if not raw_messages: - logger.warning(f"群 {group_id} 增量分析：无法获取消息") + logger.warning(f"群 {group_id} 在最近 {days} 天内无消息或无法获取") return {"success": False, "reason": "no_messages"} # 3. 清理消息 @@ -308,11 +316,7 @@ class AnalysisApplicationService: raw_messages, bot_self_ids=bot_self_ids, filter_commands=True ) - # 4. 按时间戳去重：获取最后分析消息时间戳 - last_analyzed_ts = await self.incremental_store.get_last_analyzed_timestamp( - group_id - ) - + # 5. 二次去重，确保只保留断点之后的真正新消息 if last_analyzed_ts > 0: unified_messages = [ msg for msg in unified_messages if msg.timestamp > last_analyzed_ts @@ -448,8 +452,15 @@ class AnalysisApplicationService: # 9. 保存批次并更新最后分析时间戳 await self.incremental_store.save_batch(batch) + + # 安全更新水位线：取消息最大时间戳，但不能超过当前时间+1分钟，防止未来时间戳毒化导致后续分析死锁 + import time + + safe_now = int(time.time()) + 60 + safe_ts = min(last_message_timestamp, safe_now) + await self.incremental_store.update_last_analyzed_timestamp( - group_id, last_message_timestamp + group_id, safe_ts ) logger.info( diff --git a/src/domain/repositories/message_repository.py b/src/domain/repositories/message_repository.py index c7b7019..a430495 100644 --- a/src/domain/repositories/message_repository.py +++ b/src/domain/repositories/message_repository.py @@ -24,6 +24,7 @@ class IMessageRepository(ABC): days: int = 1, max_count: int = 1000, before_id: str | None = None, + since_ts: int | None = None, ) -> list[UnifiedMessage]: """ 获取群组消息历史 @@ -33,6 +34,7 @@ class IMessageRepository(ABC): days: 获取最近 N 天的消息 max_count: 最大消息数量 before_id: 获取此 ID 之前的消息（用于分页） + since_ts: 从指定时间戳开始拉取消息（Unix timestamp），优先级高于 days。 返回: 统一消息列表，按时间升序排列 diff --git a/src/infrastructure/config/config_manager.py b/src/infrastructure/config/config_manager.py index bcd4539..2cb55ca 100644 --- a/src/infrastructure/config/config_manager.py +++ b/src/infrastructure/config/config_manager.py @@ -473,9 +473,9 @@ class ConfigManager: """获取每天最大增量分析次数""" return self._get_group("incremental").get("incremental_max_daily_analyses", 8) - def get_incremental_max_messages(self) -> int: - """获取单次增量分析的最大消息数""" - return self._get_group("incremental").get("incremental_max_messages", 300) + def get_incremental_safe_limit(self) -> int: + """获取单次增量分析的安全分析/同步上限 (Safe Count)""" + return self._get_group("incremental").get("incremental_safe_limit", 2000) def get_incremental_min_messages(self) -> int: """获取触发增量分析的最小消息数阈值""" diff --git a/src/infrastructure/platform/adapters/discord_adapter.py b/src/infrastructure/platform/adapters/discord_adapter.py index 039d5c0..7f81133 100644 --- a/src/infrastructure/platform/adapters/discord_adapter.py +++ b/src/infrastructure/platform/adapters/discord_adapter.py @@ -109,6 +109,7 @@ class DiscordAdapter(PlatformAdapter): days: int = 1, max_count: int = 100, before_id: str | None = None, + since_ts: int | None = None, ) -> list[UnifiedMessage]: """ 从 Discord 频道异步拉取历史消息记录。 @@ -143,8 +144,11 @@ class DiscordAdapter(PlatformAdapter): logger.warning(f"频道 {group_id} 不支持历史消息访问。") return [] - end_time = datetime.now() - start_time = end_time - timedelta(days=days) + if since_ts and since_ts > 0: + start_time = datetime.fromtimestamp(since_ts) + else: + end_time = datetime.now() + start_time = end_time - timedelta(days=days) messages = [] diff --git a/src/infrastructure/platform/adapters/onebot_adapter.py b/src/infrastructure/platform/adapters/onebot_adapter.py index f010b59..2c4603a 100644 --- a/src/infrastructure/platform/adapters/onebot_adapter.py +++ b/src/infrastructure/platform/adapters/onebot_adapter.py @@ -79,6 +79,7 @@ class OneBotAdapter(PlatformAdapter): days: int = 1, max_count: int = 1000, before_id: str | None = None, + since_ts: int | None = None, ) -> list[UnifiedMessage]: """ 从 OneBot 后端拉取群组历史消息。 @@ -89,6 +90,7 @@ class OneBotAdapter(PlatformAdapter): days (int): 拉取过去几天的消息 max_count (int): 最大拉取条数 before_id (str, optional): 锚点消息 ID，用于分页回溯 + since_ts (int, optional): 从指定时间戳开始拉取消息（Unix timestamp），优先级高于 days。 Returns: list[UnifiedMessage]: 统一格式的消息列表 @@ -100,16 +102,21 @@ class OneBotAdapter(PlatformAdapter): chunk_size = 100 # 每次拉取 100 条，较为稳健 all_raw_messages = [] - end_time = datetime.now() - start_time = end_time - timedelta(days=days) - start_timestamp = int(start_time.timestamp()) + # 确定回溯的起始时间点 + if since_ts and since_ts > 0: + start_timestamp = since_ts + else: + end_time_dt = datetime.now() + start_time_dt = end_time_dt - timedelta(days=days) + start_timestamp = int(start_time_dt.timestamp()) - # 使用 message_seq (在 NapCat 中通常可用 message_id 作为 seq 参数) - # 进行分页回溯拉取 + # 使用 message_seq 或 message_id 进行分页回溯拉取 current_anchor_id = before_id logger.info( - f"OneBot 开始分页回溯拉取消息: 群 {group_id}, 时间限制 {days}天, 数量限制 {max_count}" + f"OneBot 开始分页回溯消息: 群 {group_id}, " + f"起始时间 {datetime.fromtimestamp(start_timestamp).strftime('%Y-%m-%d %H:%M:%S')}, " + f"上限 {max_count} 条" ) while len(all_raw_messages) < max_count: @@ -118,7 +125,7 @@ class OneBotAdapter(PlatformAdapter): params = { "group_id": int(group_id), "count": fetch_count, - "reverseOrder": True, # 关键：协助分页向上回退拉取历史 + "reverseOrder": False, # 关键修复：关闭此属性以启用标准的回滚分页逻辑 } if current_anchor_id: @@ -169,7 +176,7 @@ class OneBotAdapter(PlatformAdapter): continue # 时间范围判定 - if start_timestamp <= msg_time <= int(end_time.timestamp()): + if start_timestamp <= msg_time <= int(datetime.now().timestamp()): all_raw_messages.append(raw_msg) # 提取锚点。 @@ -185,18 +192,26 @@ class OneBotAdapter(PlatformAdapter): mid_val = chunk_earliest_msg.get("message_id") # 优先使用 seq_val (针对 LLBot)，如果没有则回退回 ID - new_anchor_id = seq_val if seq_val is not None else mid_val + # 优先使用 seq_val 进行精准的分页位移控制 + if seq_val is not None: + try: + # 通过 -1 克服 API 的 inclusive (包含) 限制，防止翻页死循环 + new_anchor_id = int(seq_val) - 1 + except (ValueError, TypeError): + new_anchor_id = seq_val + else: + new_anchor_id = mid_val - # 如果时间已经超过限制，或者锚点没有变化（说明已经到底），则停止 - if chunk_earliest_time < start_timestamp: + # 如果消息时间已到达起始点，或者锚点无法继续往前位移，则停止 + if chunk_earliest_time <= start_timestamp: logger.debug( - f"OneBot 分页拉取：消息时间 ({chunk_earliest_time}) 早于起始时间 ({start_timestamp})，回溯完成。" + f"OneBot 分页拉取：已到达起始时间 ({start_timestamp})，回溯同步完成。" ) break if current_anchor_id and str(new_anchor_id) == str(current_anchor_id): logger.debug( - "OneBot 分页拉取：消息锚点没有变化，可能已到达历史尽头。" + "OneBot 分页拉取：消息锚点未发生有效位移，可能已到达历史尽头。" ) break diff --git a/src/infrastructure/platform/adapters/telegram_adapter.py b/src/infrastructure/platform/adapters/telegram_adapter.py index 13a00e6..31c003a 100644 --- a/src/infrastructure/platform/adapters/telegram_adapter.py +++ b/src/infrastructure/platform/adapters/telegram_adapter.py @@ -169,6 +169,7 @@ class TelegramAdapter(PlatformAdapter): days: int = 1, max_count: int = 100, before_id: str | None = None, + since_ts: int | None = None, ) -> list[UnifiedMessage]: """ 获取历史消息。 @@ -190,7 +191,11 @@ class TelegramAdapter(PlatformAdapter): except (TypeError, ValueError): logger.warning(f"[Telegram] before_id invalid: {before_id}") - cutoff_time = datetime.now(timezone.utc) - timedelta(days=days) + if since_ts and since_ts > 0: + # 统一使用 UTC 以兼容数据库记录的时间存储 + cutoff_time = datetime.fromtimestamp(since_ts, timezone.utc) + else: + cutoff_time = datetime.now(timezone.utc) - timedelta(days=days) target_count = max(1, int(max_count)) page_size = target_count current_page = 1