feat(增量分析): 更新增量分析配置，增加安全上限并优化消息拉取逻辑

_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",

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(

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。
 
         返回:
             统一消息列表，按时间升序排列

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:
         """获取触发增量分析的最小消息数阈值"""

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 = []
 

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
 

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