diff --git a/docs/INCREMENTAL_ANALYSIS.md b/docs/INCREMENTAL_ANALYSIS.md index b4a68d6..67e6b55 100644 --- a/docs/INCREMENTAL_ANALYSIS.md +++ b/docs/INCREMENTAL_ANALYSIS.md @@ -2,36 +2,47 @@ ## 概述 -增量分析是对传统"一天一次完整分析"模式的改进。核心思路是在一天内多次执行小批量分析，将结果累积合并，最终在配置的报告时间点生成完整的日报。 +增量分析是对传统"一天一次完整分析"模式的改进。核心思路是在一天内多次执行小批量分析，将结果作为独立批次存储，最终在配置的报告时间点按滑动窗口查询并合并所有批次，生成完整的日报。 ### 解决的问题 1. **消息量过大时分析效果差**：单次拉取的消息量有限，无法覆盖全天聊天内容 2. **24小时活跃图表形同虚设**：单次分析只能捕捉到部分时段的数据 3. **API端点短期压力暴增**：所有群聊在同一时间点执行分析，LLM API 瞬时负载极高 -4. **空闲时段浪费分析次数**：固定间隔调度无法适应群聊活跃度的波动 +4. **天然日期隔离问题**：旧版按天存储（key含日期），跨天分析数据断裂，多次发送报告时窗口内容相同 -## 架构设计 +## 架构设计（v2 — 滑动窗口批次架构） 增量分析遵循项目现有的 DDD 分层架构： ``` 应用层 (Application) └── AnalysisApplicationService - ├── execute_incremental_analysis() # 单次增量批次 - └── execute_incremental_final_report() # 最终报告生成 + ├── execute_incremental_analysis() # 单次增量 → 存储独立批次 + └── execute_incremental_final_report() # 滑动窗口查询 → 合并 → 报告 领域层 (Domain) -├── IncrementalState # 增量状态实体（累积数据） -├── BatchRecord # 批次记录值对象 -└── IncrementalMergeService # 合并服务（去重、统计构建） +├── IncrementalBatch # 独立批次实体（持久化单元） +├── IncrementalState # 聚合视图（不持久化，报告时合并产生） +└── IncrementalMergeService # 合并服务（merge_batches + 去重 + 统计构建） 基础设施层 (Infrastructure) -├── IncrementalStore # 持久化（KV存储） -├── AutoScheduler # 调度器（传统/增量双模式） +├── IncrementalStore # 批次持久化（KV索引 + 批次数据） +├── AutoScheduler # 调度器（传统/增量双模式 + 过期清理） └── LLMAnalyzer # LLM分析（增量并发方法） ``` +### 核心设计变更（v1 → v2） + +| 维度 | v1（旧版） | v2（当前版本） | +|------|-----------|--------------| +| 存储单元 | 按天的 `IncrementalState` | 独立的 `IncrementalBatch` | +| KV Key | `incremental_state_{group_id}_{date}` | `incr_batch_{group_id}_{batch_id}` | +| 合并时机 | 每次增量分析时合并 | 报告生成时按窗口查询后合并 | +| 窗口范围 | 自然日（0:00-24:00） | 滑动窗口（now - analysis_days×24h ~ now） | +| 日期隔离 | 有（跨天数据断裂） | 无（窗口连续覆盖） | +| 多次发送 | 相同数据 | 窗口随时间滑动，数据不同 | + ## 数据流 ### 增量分析批次流程 @@ -41,13 +52,14 @@ → 获取启用的群聊目标 → 交错并发执行（控制API压力） → AnalysisApplicationService.execute_incremental_analysis() - → 加载/创建当天 IncrementalState + → 获取 last_analyzed_timestamp（跨批次去重） → 拉取自上次分析以来的新消息 → 检查最小消息数阈值（不足则跳过） → LLM 并发分析（话题 + 金句，限制数量） → 统计小时级消息分布、用户活跃度 - → IncrementalState.merge_batch() 合并（去重） - → 持久化保存 + → 构建 IncrementalBatch 对象 + → save_batch() 保存批次 + 更新索引 + → update_last_analyzed_timestamp() ``` ### 最终报告生成流程 @@ -57,19 +69,55 @@ → 获取启用的群聊目标 → 交错并发执行 → AnalysisApplicationService.execute_incremental_final_report() - → 加载当天 IncrementalState - → 检查是否有分析数据 - → IncrementalMergeService.build_final_statistics() → GroupStatistics - → IncrementalMergeService.build_topics_for_report() → [SummaryTopic] - → IncrementalMergeService.build_quotes_for_report() → [GoldenQuote] - → 用户画像分析（使用累积的全天数据） - → 组装 analysis_result + → 计算滑动窗口: [now - analysis_days×24h, now] + → query_batches() 按窗口查询批次列表 + → IncrementalMergeService.merge_batches() → IncrementalState + → 用户画像分析（使用合并后的全窗口数据） + → build_analysis_result() → analysis_result → ReportDispatcher 分发报告 + → cleanup_old_batches() 清理 2×窗口外的过期批次 +``` + +## 持久化（KV 键设计） + +``` +批次索引: incr_batch_index_{group_id} + 值: [{"batch_id": "uuid", "timestamp": 1234567890.0}, ...] + +批次数据: incr_batch_{group_id}_{batch_id} + 值: IncrementalBatch.to_dict() + +去重时间戳: incr_last_ts_{group_id} + 值: int (最后分析消息的 epoch 时间戳) +``` + +### 滑动窗口查询 + +```python +# 报告生成时 +window_end = time.time() +window_start = window_end - (analysis_days * 24 * 3600) +batches = await store.query_batches(group_id, window_start, window_end) +state = merge_service.merge_batches(batches, window_start, window_end) +``` + +### 过期清理 + +报告发送成功后，清理 2×窗口范围之前的旧批次： + +```python +before_ts = time.time() - (analysis_days * 2 * 24 * 3600) +await store.cleanup_old_batches(group_id, before_ts) ``` ## 去重机制 -### 话题去重 +### 消息去重（跨批次） + +使用全局的 `incr_last_ts_{group_id}` 记录最后分析消息时间戳， +每次增量分析只处理时间戳大于该值的新消息。 + +### 话题去重（合并时） 使用 Jaccard 字符级相似度，阈值 0.6： @@ -77,9 +125,9 @@ similarity = len(chars_a & chars_b) / len(chars_a | chars_b) ``` -比较维度：`keyword` + `summary` 文本拼接后的字符集合。 +比较维度：`topic` 文本的字符集合。 -### 金句去重 +### 金句去重（合并时） 同样使用 Jaccard 字符级相似度，阈值 0.7（更严格，避免误去重）。 @@ -112,37 +160,32 @@ similarity = len(chars_a & chars_b) / len(chars_a | chars_b) - 在活跃时段内按间隔执行小批量分析（如每2小时一次） - 每次只分析上次以来的新消息，提取少量话题和金句 -- 在配置的报告时间点汇总全天数据生成最终报告 +- 在配置的报告时间点按滑动窗口合并所有批次生成最终报告 - 适合消息量大、需要全天覆盖的群聊 +- 支持同一天多次发送报告（窗口随时间滑动） ## 命令 | 命令 | 说明 | |------|------| -| `/增量状态` | 查看当前群今日的增量分析累积情况 | +| `/增量状态` | 查看当前滑动窗口内的增量分析累积情况 | | `/分析设置 status` | 查看完整设置状态（含增量分析配置） | -## 持久化 +## 旧版兼容 -增量状态使用 AstrBot 的 KV 存储： - -- Key 格式：`incremental_state_{group_id}_{date_str}` -- Value：JSON 序列化的 `IncrementalState` -- 每日自然过期（下一天生成新的 key） +`IncrementalStore.migrate_legacy_state()` 支持将旧版 `incremental_state_{group_id}_{date}` 格式的数据迁移到新批次架构。迁移后旧键会被删除。 ## 文件清单 | 文件 | 层 | 说明 | |------|-----|------| -| `src/domain/entities/incremental_state.py` | 领域 | 增量状态实体 + 批次记录 | -| `src/domain/services/incremental_merge_service.py` | 领域 | 合并服务（构建统计、话题、金句） | -| `src/infrastructure/persistence/incremental_store.py` | 基础设施 | KV 持久化仓储 | -| `src/infrastructure/analysis/llm_analyzer.py` | 基础设施 | 新增 `analyze_incremental_concurrent()` | -| `src/infrastructure/analysis/analyzers/base_analyzer.py` | 基础设施 | 新增 `_incremental_max_count` 属性 | -| `src/infrastructure/analysis/analyzers/topic_analyzer.py` | 基础设施 | 覆盖 `get_max_count()` | -| `src/infrastructure/analysis/analyzers/golden_quote_analyzer.py` | 基础设施 | 覆盖 `get_max_count()` | -| `src/infrastructure/scheduler/auto_scheduler.py` | 基础设施 | 双模式调度（传统+增量） | +| `src/domain/entities/incremental_state.py` | 领域 | IncrementalBatch（批次实体）+ IncrementalState（聚合视图） | +| `src/domain/services/incremental_merge_service.py` | 领域 | merge_batches() + 构建统计、话题、金句 | +| `src/infrastructure/persistence/incremental_store.py` | 基础设施 | 批次索引/数据 KV 持久化、窗口查询、过期清理 | +| `src/infrastructure/analysis/llm_analyzer.py` | 基础设施 | `analyze_incremental_concurrent()` | +| `src/infrastructure/analysis/analyzers/base_analyzer.py` | 基础设施 | `_incremental_max_count` 属性 | +| `src/infrastructure/scheduler/auto_scheduler.py` | 基础设施 | 双模式调度（传统+增量）+ 报告后过期批次清理 | | `src/infrastructure/config/config_manager.py` | 基础设施 | 10个增量配置 getter | -| `src/application/services/analysis_application_service.py` | 应用 | 增量分析 + 最终报告用例 | -| `main.py` | 入口 | 接线 + `/增量状态` 命令 | +| `src/application/services/analysis_application_service.py` | 应用 | 增量分析（存批次）+ 最终报告（窗口查询+合并）| +| `main.py` | 入口 | 接线 + `/增量状态` 命令（滑动窗口查询）| | `_conf_schema.json` | 配置 | 增量分析配置 Schema | diff --git a/main.py b/main.py index 2ac6b15..e580626 100644 --- a/main.py +++ b/main.py @@ -739,7 +739,7 @@ class QQGroupDailyAnalysis(Star): @filter.command("增量状态", alias={"incremental_status"}) @filter.permission_type(PermissionType.ADMIN) async def incremental_status(self, event: AstrMessageEvent): - """查看当前增量分析状态""" + """查看当前增量分析状态（滑动窗口）""" group_id = self._get_group_id_from_event(event) if not group_id: yield event.plain_result("❌ 请在群聊中使用此命令") @@ -749,23 +749,42 @@ class QQGroupDailyAnalysis(Star): yield event.plain_result("ℹ️ 增量分析模式未启用，请在插件配置中开启") return - import datetime as dt_mod + import time as time_mod - today_str = dt_mod.datetime.now().strftime("%Y-%m-%d") - state = await self.incremental_store.get_state(group_id, today_str) + # 计算滑动窗口范围 + analysis_days = self.config_manager.get_analysis_days() + window_end = time_mod.time() + window_start = window_end - (analysis_days * 24 * 3600) - if not state or state.total_analysis_count == 0: - yield event.plain_result(f"📊 今日 ({today_str}) 尚无增量分析数据") + # 查询窗口内的批次 + batches = await self.incremental_store.query_batches( + group_id, window_start, window_end + ) + + if not batches: + from datetime import datetime + + start_str = datetime.fromtimestamp(window_start).strftime("%m-%d %H:%M") + end_str = datetime.fromtimestamp(window_end).strftime("%m-%d %H:%M") + yield event.plain_result( + f"📊 滑动窗口 ({start_str} ~ {end_str}) 内尚无增量分析数据" + ) return + # 合并批次获取聚合视图 + state = self.incremental_merge_service.merge_batches( + batches, window_start, window_end + ) summary = state.get_summary() + yield event.plain_result( - f"📊 增量分析状态 ({today_str})\n" - f"• 分析次数: {summary['total_analysis_count']}\n" - f"• 累计消息: {summary['total_message_count']}\n" + f"📊 增量分析状态 (窗口: {summary['window']})\n" + f"• 分析次数: {summary['total_analyses']}\n" + f"• 累计消息: {summary['total_messages']}\n" f"• 话题数: {summary['topics_count']}\n" f"• 金句数: {summary['quotes_count']}\n" - f"• 参与者: {summary['participant_count']}" + f"• 参与者: {summary['participants']}\n" + f"• 高峰时段: {summary['peak_hours']}" ) def _get_group_id_from_event(self, event: AstrMessageEvent) -> str | None: diff --git a/src/application/services/analysis_application_service.py b/src/application/services/analysis_application_service.py index 98e133c..26a2de1 100644 --- a/src/application/services/analysis_application_service.py +++ b/src/application/services/analysis_application_service.py @@ -6,9 +6,11 @@ import asyncio import datetime as dt +import time as time_mod from collections import defaultdict from typing import Any +from ...domain.entities.incremental_state import IncrementalBatch from ...domain.models.data_models import TokenUsage from ...domain.repositories.analysis_repository import IAnalysisProvider from ...domain.repositories.report_repository import IReportGenerator @@ -185,10 +187,10 @@ class AnalysisApplicationService: self, group_id: str, platform_id: str | None = None ) -> dict[str, Any]: """ - 执行一次增量分析用例。 + 执行一次增量分析用例（滑动窗口批次架构）。 与每日分析不同，增量分析每次仅处理最近一段时间的消息， - 提取少量话题和金句，将结果合并到当天的累积状态中。 + 提取少量话题和金句，将结果作为独立批次存储到 KV。 不生成用户称号（留到最终报告时再做），不生成报告。 流程： @@ -199,8 +201,8 @@ class AnalysisApplicationService: 5. 检查最小消息阈值 6. 计算基础统计（小时分布、用户活跃、表情） 7. LLM 增量分析（仅话题 + 金句） - 8. 构建合并参数并合并到 IncrementalState - 9. 持久化状态 + 8. 构建 IncrementalBatch 并保存 + 9. 更新最后分析消息时间戳 10. 返回批次结果 Args: @@ -208,7 +210,7 @@ class AnalysisApplicationService: platform_id: 平台标识，缺省为默认 Returns: - dict: 包含 success、batch_record、state_summary 等信息 + dict: 包含 success、batch_summary 等信息 """ if not self.incremental_store: raise RuntimeError("增量分析未初始化：缺少 IncrementalStore") @@ -241,15 +243,16 @@ class AnalysisApplicationService: raw_messages, bot_self_ids=bot_self_ids, filter_commands=True ) - # 4. 获取当天增量状态并按时间戳去重 - today_str = dt.datetime.now().strftime("%Y-%m-%d") - state = await self.incremental_store.get_or_create_state(group_id, today_str) + # 4. 按时间戳去重：获取最后分析消息时间戳 + last_analyzed_ts = await self.incremental_store.get_last_analyzed_timestamp( + group_id + ) - if state.last_analyzed_message_timestamp > 0: + if last_analyzed_ts > 0: unified_messages = [ msg for msg in unified_messages - if msg.timestamp > state.last_analyzed_message_timestamp + if msg.timestamp > last_analyzed_ts ] # 5. 检查最小消息阈值 @@ -297,7 +300,7 @@ class AnalysisApplicationService: ) ) - # 8. 构建合并参数 + # 8. 构建 IncrementalBatch # 8a. 转换话题: SummaryTopic -> dict new_topics = [ {"topic": t.topic, "contributors": t.contributors, "detail": t.detail} @@ -322,7 +325,7 @@ class AnalysisApplicationService: "total_tokens": token_usage.total_tokens, } - # 8d. 转换用户统计: AnalysisDomainService 格式 -> IncrementalState 格式 + # 8d. 转换用户统计: AnalysisDomainService 格式 -> IncrementalBatch 格式 user_stats = self._convert_user_activity_for_merge( user_activity, unified_messages ) @@ -338,7 +341,7 @@ class AnalysisApplicationService: } # 8f. 获取参与者 ID 和最后消息时间戳 - participant_ids = {msg.sender_id for msg in unified_messages} + participant_ids = list({msg.sender_id for msg in unified_messages}) last_message_timestamp = max( (msg.timestamp for msg in unified_messages), default=0 ) @@ -346,35 +349,38 @@ class AnalysisApplicationService: # 8g. 计算本批次总字符数 characters_count = sum(msg.get_text_length() for msg in unified_messages) - # 9. 合并到增量状态 - batch_record = state.merge_batch( + # 构建批次对象 + batch = IncrementalBatch( + group_id=group_id, + timestamp=time_mod.time(), messages_count=len(unified_messages), characters_count=characters_count, - hourly_msg_counts=hourly_msg_counts, - hourly_char_counts=hourly_char_counts, + hourly_msg_counts={str(k): v for k, v in hourly_msg_counts.items()}, + hourly_char_counts={str(k): v for k, v in hourly_char_counts.items()}, user_stats=user_stats, emoji_stats=emoji_stats, - new_topics=new_topics, - new_quotes=new_quotes, + topics=new_topics, + golden_quotes=new_quotes, token_usage=token_usage_dict, last_message_timestamp=last_message_timestamp, participant_ids=participant_ids, ) - # 10. 持久化状态 - await self.incremental_store.save_state(state) + # 9. 保存批次并更新最后分析时间戳 + await self.incremental_store.save_batch(batch) + await self.incremental_store.update_last_analyzed_timestamp( + group_id, last_message_timestamp + ) logger.info( f"群 {group_id} 增量分析完成: " f"本批次消息={len(unified_messages)}, " - f"新话题={len(new_topics)}, 新金句={len(new_quotes)}, " - f"累计分析次数={state.total_analysis_count}" + f"新话题={len(new_topics)}, 新金句={len(new_quotes)}" ) return { "success": True, - "batch_record": batch_record.to_dict(), - "state_summary": state.get_summary(), + "batch_summary": batch.get_summary(), "messages_count": len(unified_messages), } @@ -382,19 +388,21 @@ class AnalysisApplicationService: self, group_id: str, platform_id: str | None = None ) -> dict[str, Any]: """ - 基于当天增量累积状态生成最终报告。 + 基于滑动窗口内的增量批次生成最终报告。 - 将一天内多次增量分析积累的话题、金句、统计数据汇总， - 额外执行用户称号分析（需要完整的累积数据），然后生成 - 与传统每日分析格式完全一致的 analysis_result。 + 按 analysis_days × 24h 的时间窗口查询所有批次， + 合并为 IncrementalState，额外执行用户称号分析， + 然后生成与传统每日分析格式完全一致的 analysis_result。 流程： - 1. 加载当天增量状态 - 2. 检查状态有效性 - 3. 执行用户称号 LLM 分析（基于累积数据） - 4. 使用 IncrementalMergeService 构建 analysis_result - 5. 持久化到 history_manager - 6. 返回结果 + 1. 计算滑动窗口范围 + 2. 查询窗口内的所有批次 + 3. 检查批次有效性 + 4. 合并批次为 IncrementalState + 5. 执行用户称号 LLM 分析（基于合并后的累积数据） + 6. 使用 IncrementalMergeService 构建 analysis_result + 7. 持久化到 history_manager + 8. 返回结果 Args: group_id: 群组 ID @@ -410,34 +418,42 @@ class AnalysisApplicationService: logger.info(f"开始增量最终报告: 群 {group_id}, 平台 {platform_id or '默认'}") - # 1. 加载当天增量状态 - today_str = dt.datetime.now().strftime("%Y-%m-%d") - state = await self.incremental_store.get_state(group_id, today_str) + # 1. 计算滑动窗口范围 + analysis_days = self.config_manager.get_analysis_days() + window_end = time_mod.time() + window_start = window_end - (analysis_days * 24 * 3600) - # 2. 检查状态有效性 - if not state or state.total_analysis_count == 0: + # 2. 查询窗口内的所有批次 + batches = await self.incremental_store.query_batches( + group_id, window_start, window_end + ) + + # 3. 检查批次有效性 + if not batches: logger.warning( - f"群 {group_id} 无当天增量分析数据，无法生成最终报告" + f"群 {group_id} 滑动窗口内无增量分析数据，无法生成最终报告" ) return {"success": False, "reason": "no_incremental_data"} - # 3. 获取适配器（报告发送需要） + # 4. 合并批次为 IncrementalState + state = self.incremental_merge_service.merge_batches( + batches, window_start, window_end + ) + + # 5. 获取适配器（报告发送需要） adapter = self.bot_manager.get_adapter(platform_id) if not adapter: raise ValueError(f"未找到平台 {platform_id} 的适配器") - # 4. 执行用户称号 LLM 分析 + # 6. 执行用户称号 LLM 分析 user_titles = [] user_title_enabled = self.config_manager.get_user_title_analysis_enabled() if user_title_enabled and state.user_activities: max_user_titles = self.config_manager.get_max_user_titles() - # 从累积的 user_activities 中取出 top 用户 + # 从合并后的 user_activities 中取出 top 用户 top_users = state.get_user_activity_ranking(max_user_titles) - # 准备用户称号分析所需的 legacy 消息格式 - # 因为增量模式不保存原始消息，这里用空列表 - # 用户称号分析器主要依赖 user_analysis 和 top_users，消息内容非必需 unified_msg_origin = ( f"{platform_id}:GroupMessage:{group_id}" if platform_id @@ -468,20 +484,20 @@ class AnalysisApplicationService: state.total_token_usage.get("total_tokens", 0) + title_token_usage.total_tokens ) - await self.incremental_store.save_state(state) except Exception as e: logger.error(f"增量最终报告用户称号分析失败: {e}", exc_info=True) - # 5. 构建 analysis_result + # 7. 构建 analysis_result analysis_result = self.incremental_merge_service.build_analysis_result( state, user_titles ) - # 6. 持久化到 history_manager + # 8. 持久化到 history_manager await self.history_manager.save_analysis(group_id, analysis_result) logger.info( f"群 {group_id} 增量最终报告完成: " + f"窗口={state.get_window_date_str()}, " f"累计消息={state.total_message_count}, " f"话题={len(state.topics)}, 金句={len(state.golden_quotes)}, " f"批次={state.total_analysis_count}" @@ -528,7 +544,7 @@ class AnalysisApplicationService: ) -> dict[str, dict]: """ 将 AnalysisDomainService.analyze_user_activity() 的返回格式 - 转换为 IncrementalState.merge_batch() 所需的 user_stats 格式。 + 转换为 IncrementalBatch 所需的 user_stats 格式。 转换映射： - nickname -> name @@ -540,7 +556,7 @@ class AnalysisApplicationService: messages: 本批次的消息列表（用于提取每个用户的最后发言时间） Returns: - dict: IncrementalState.merge_batch() 所需的 user_stats 格式 + dict: IncrementalBatch 所需的 user_stats 格式 """ # 预先计算每个用户的最后消息时间戳 user_last_time: dict[str, int] = {} diff --git a/src/domain/entities/__init__.py b/src/domain/entities/__init__.py index 9c14c4a..755018f 100644 --- a/src/domain/entities/__init__.py +++ b/src/domain/entities/__init__.py @@ -4,8 +4,8 @@ 该模块导出所有领域实体类，包括: - AnalysisTask: 分析任务聚合根 - GroupAnalysisResult: 群聊分析结果实体 -- IncrementalState: 增量分析状态实体 -- BatchRecord: 增量分析批次记录 +- IncrementalBatch: 增量分析独立批次实体 +- IncrementalState: 增量分析聚合视图（报告时使用） """ from .analysis_result import ( @@ -19,7 +19,7 @@ from .analysis_result import ( UserTitle, ) from .analysis_task import AnalysisTask, TaskStatus -from .incremental_state import BatchRecord, IncrementalState +from .incremental_state import IncrementalBatch, IncrementalState # 别名，保持向后兼容 AnalysisResult = GroupAnalysisResult @@ -36,6 +36,6 @@ __all__ = [ "EmojiStatistics", "ActivityVisualization", "GroupStatistics", + "IncrementalBatch", "IncrementalState", - "BatchRecord", ] diff --git a/src/domain/entities/incremental_state.py b/src/domain/entities/incremental_state.py index ffda472..87aabbf 100644 --- a/src/domain/entities/incremental_state.py +++ b/src/domain/entities/incremental_state.py @@ -1,88 +1,171 @@ """ -增量分析状态实体 +增量分析实体 — 滑动窗口批次存储架构 -存储单个群聊在一天内累积的增量分析数据。 -每次增量分析产生一个批次(batch)，批次结果合并到此状态中。 -最终报告时从此状态中提取完整的统计数据和分析内容。 +核心概念： +- IncrementalBatch: 单次增量分析产生的独立批次数据，按批次独立存储 +- IncrementalState: 报告生成时由多个批次合并而成的聚合视图（不再持久化） + +滑动窗口设计： +- 每次增量分析产生一个 IncrementalBatch，独立存储到 KV +- 最终报告时按 analysis_days × 24h 的时间窗口查询批次并合并 +- 支持同一天多次发送报告，每次都基于当前时间窗口内的所有批次 """ import time +import uuid from dataclasses import dataclass, field from datetime import datetime @dataclass -class BatchRecord: - """单次增量分析批次记录""" +class IncrementalBatch: + """ + 单次增量分析批次数据 - batch_id: int = 0 - timestamp: float = 0.0 - message_count: int = 0 - new_topics_count: int = 0 - new_quotes_count: int = 0 - token_usage: dict = field(default_factory=dict) + 每次增量分析执行完毕后产生一个 IncrementalBatch， + 包含该批次的所有统计数据和 LLM 分析结果，独立存储到 KV。 + + Attributes: + group_id: 群组 ID + batch_id: 批次唯一标识（UUID） + timestamp: 批次创建时间戳（epoch） + messages_count: 本批次分析的消息数量 + characters_count: 本批次的总字符数 + hourly_msg_counts: 按小时的消息计数 {hour_str: count} + hourly_char_counts: 按小时的字符计数 {hour_str: count} + user_stats: 用户统计 {user_id: {name, message_count, char_count, ...}} + emoji_stats: 表情统计 {emoji_type: count} + topics: 本批次提取的话题列表 + golden_quotes: 本批次提取的金句列表 + token_usage: 本批次 token 消耗 {prompt_tokens, completion_tokens, total_tokens} + last_message_timestamp: 本批次最后一条消息的时间戳 + participant_ids: 本批次参与者 ID 列表 + """ + + group_id: str = "" + batch_id: str = field(default_factory=lambda: str(uuid.uuid4())) + timestamp: float = field(default_factory=time.time) + + # 统计数据 + messages_count: int = 0 + characters_count: int = 0 + hourly_msg_counts: dict[str, int] = field(default_factory=dict) + hourly_char_counts: dict[str, int] = field(default_factory=dict) + + # 用户活跃数据 + user_stats: dict[str, dict] = field(default_factory=dict) + + # 表情统计 + emoji_stats: dict[str, int] = field(default_factory=dict) + + # LLM 分析结果 + topics: list[dict] = field(default_factory=list) + golden_quotes: list[dict] = field(default_factory=list) + + # Token 消耗 + token_usage: dict = field(default_factory=lambda: { + "prompt_tokens": 0, + "completion_tokens": 0, + "total_tokens": 0, + }) + + # 增量追踪 + last_message_timestamp: int = 0 + participant_ids: list[str] = field(default_factory=list) def to_dict(self) -> dict: - """序列化为字典""" + """序列化为字典，用于 KV 存储""" return { + "group_id": self.group_id, "batch_id": self.batch_id, "timestamp": self.timestamp, - "message_count": self.message_count, - "new_topics_count": self.new_topics_count, - "new_quotes_count": self.new_quotes_count, + "messages_count": self.messages_count, + "characters_count": self.characters_count, + "hourly_msg_counts": self.hourly_msg_counts, + "hourly_char_counts": self.hourly_char_counts, + "user_stats": self.user_stats, + "emoji_stats": self.emoji_stats, + "topics": self.topics, + "golden_quotes": self.golden_quotes, "token_usage": self.token_usage, + "last_message_timestamp": self.last_message_timestamp, + "participant_ids": self.participant_ids, } @classmethod - def from_dict(cls, data: dict) -> "BatchRecord": + def from_dict(cls, data: dict) -> "IncrementalBatch": """从字典反序列化""" return cls( - batch_id=data.get("batch_id", 0), + group_id=data.get("group_id", ""), + batch_id=data.get("batch_id", ""), timestamp=data.get("timestamp", 0.0), - message_count=data.get("message_count", 0), - new_topics_count=data.get("new_topics_count", 0), - new_quotes_count=data.get("new_quotes_count", 0), - token_usage=data.get("token_usage", {}), + messages_count=data.get("messages_count", 0), + characters_count=data.get("characters_count", 0), + hourly_msg_counts=data.get("hourly_msg_counts", {}), + hourly_char_counts=data.get("hourly_char_counts", {}), + user_stats=data.get("user_stats", {}), + emoji_stats=data.get("emoji_stats", {}), + topics=data.get("topics", []), + golden_quotes=data.get("golden_quotes", []), + token_usage=data.get("token_usage", { + "prompt_tokens": 0, + "completion_tokens": 0, + "total_tokens": 0, + }), + last_message_timestamp=data.get("last_message_timestamp", 0), + participant_ids=data.get("participant_ids", []), ) + def get_summary(self) -> dict: + """获取批次摘要信息""" + return { + "batch_id": self.batch_id[:8], + "timestamp": datetime.fromtimestamp(self.timestamp).strftime( + "%Y-%m-%d %H:%M:%S" + ), + "messages_count": self.messages_count, + "topics_count": len(self.topics), + "quotes_count": len(self.golden_quotes), + "participants": len(self.participant_ids), + } + @dataclass class IncrementalState: """ - 增量分析状态聚合实体 + 增量分析聚合视图（报告时使用） - 该实体代表一个群聊在一天内的增量分析累积状态。 - 随着当天多次增量分析的执行，话题、金句、统计数据会不断合并更新。 + 由多个 IncrementalBatch 合并而成，不直接持久化。 + IncrementalMergeService.merge_batches() 负责从批次列表构建此对象。 Attributes: group_id: 群组 ID - date_str: 日期字符串 (YYYY-MM-DD) - topics: 累积的话题列表（每个元素为 dict，包含 topic/contributors/detail） - golden_quotes: 累积的金句列表（每个元素为 dict，包含 content/sender/reason） - hourly_message_counts: 每小时消息计数 {hour_int: count} - hourly_character_counts: 每小时字符计数 {hour_int: count} - user_activities: 用户活跃数据 {user_id: {name, message_count, char_count, ...}} - emoji_counts: 表情统计 {emoji_type: count} - batch_records: 已完成的增量分析批次记录 - total_message_count: 当天总消息数 - total_character_count: 当天总字符数 - total_analysis_count: 当天已执行的增量分析次数 + window_start: 滑动窗口起始时间戳 + window_end: 滑动窗口结束时间戳 + topics: 合并去重后的话题列表 + golden_quotes: 合并去重后的金句列表 + hourly_message_counts: 合并后的每小时消息计数 {hour_str: count} + hourly_character_counts: 合并后的每小时字符计数 {hour_str: count} + user_activities: 合并后的用户活跃数据 + emoji_counts: 合并后的表情统计 + total_message_count: 窗口内总消息数 + total_character_count: 窗口内总字符数 + total_analysis_count: 窗口内批次数量 total_token_usage: 累计 token 消耗 - last_analyzed_message_timestamp: 上次分析的最后一条消息时间戳（用于去重） + last_analyzed_message_timestamp: 最后分析消息时间戳 all_participant_ids: 所有参与者 ID 集合 - created_at: 状态创建时间 - updated_at: 状态最后更新时间 """ # 标识信息 group_id: str = "" - date_str: str = "" + window_start: float = 0.0 + window_end: float = 0.0 - # 累积的 LLM 分析结果 + # 合并后的 LLM 分析结果 topics: list[dict] = field(default_factory=list) golden_quotes: list[dict] = field(default_factory=list) - # 累积的统计数据（按小时） + # 合并后的统计数据（按小时） hourly_message_counts: dict[str, int] = field(default_factory=dict) hourly_character_counts: dict[str, int] = field(default_factory=dict) @@ -92,9 +175,6 @@ class IncrementalState: # 表情统计 emoji_counts: dict[str, int] = field(default_factory=dict) - # 批次记录 - batch_records: list[BatchRecord] = field(default_factory=list) - # 汇总统计 total_message_count: int = 0 total_character_count: int = 0 @@ -113,205 +193,6 @@ class IncrementalState: created_at: float = field(default_factory=time.time) updated_at: float = field(default_factory=time.time) - def merge_batch( - self, - messages_count: int, - characters_count: int, - hourly_msg_counts: dict[int, int], - hourly_char_counts: dict[int, int], - user_stats: dict[str, dict], - emoji_stats: dict[str, int], - new_topics: list[dict], - new_quotes: list[dict], - token_usage: dict, - last_message_timestamp: int, - participant_ids: set[str], - ) -> "BatchRecord": - """ - 合并一次增量分析的结果到当前状态中。 - - Args: - messages_count: 本批次分析的消息数量 - characters_count: 本批次的总字符数 - hourly_msg_counts: 本批次按小时的消息计数 {hour: count} - hourly_char_counts: 本批次按小时的字符计数 {hour: count} - user_stats: 本批次用户统计 {user_id: {name, message_count, char_count, ...}} - emoji_stats: 本批次表情统计 {emoji_type: count} - new_topics: 本批次提取的新话题 - new_quotes: 本批次提取的新金句 - token_usage: 本批次 token 消耗 {prompt_tokens, completion_tokens, total_tokens} - last_message_timestamp: 本批次最后一条消息的时间戳 - participant_ids: 本批次参与者 ID 集合 - - Returns: - BatchRecord: 本次批次的记录 - """ - # 更新统计汇总 - self.total_message_count += messages_count - self.total_character_count += characters_count - self.total_analysis_count += 1 - - # 合并小时统计 - for hour, count in hourly_msg_counts.items(): - hour_key = str(hour) - self.hourly_message_counts[hour_key] = ( - self.hourly_message_counts.get(hour_key, 0) + count - ) - for hour, count in hourly_char_counts.items(): - hour_key = str(hour) - self.hourly_character_counts[hour_key] = ( - self.hourly_character_counts.get(hour_key, 0) + count - ) - - # 合并用户活跃数据 - for user_id, stats in user_stats.items(): - if user_id in self.user_activities: - existing = self.user_activities[user_id] - existing["message_count"] = ( - existing.get("message_count", 0) + stats.get("message_count", 0) - ) - existing["char_count"] = ( - existing.get("char_count", 0) + stats.get("char_count", 0) - ) - existing["emoji_count"] = ( - existing.get("emoji_count", 0) + stats.get("emoji_count", 0) - ) - # 合并活跃小时集合 - existing_hours = set(existing.get("active_hours", [])) - new_hours = set(stats.get("active_hours", [])) - existing["active_hours"] = list(existing_hours | new_hours) - # 更新最后发言时间 - if stats.get("last_message_time", 0) > existing.get("last_message_time", 0): - existing["last_message_time"] = stats["last_message_time"] - else: - self.user_activities[user_id] = dict(stats) - - # 合并表情统计 - for emoji_type, count in emoji_stats.items(): - self.emoji_counts[emoji_type] = ( - self.emoji_counts.get(emoji_type, 0) + count - ) - - # 合并话题（带去重） - for new_topic in new_topics: - if not self._is_duplicate_topic(new_topic): - self.topics.append(new_topic) - - # 合并金句（带去重） - for new_quote in new_quotes: - if not self._is_duplicate_quote(new_quote): - self.golden_quotes.append(new_quote) - - # 更新 token 消耗 - self.total_token_usage["prompt_tokens"] = ( - self.total_token_usage.get("prompt_tokens", 0) - + token_usage.get("prompt_tokens", 0) - ) - self.total_token_usage["completion_tokens"] = ( - self.total_token_usage.get("completion_tokens", 0) - + token_usage.get("completion_tokens", 0) - ) - self.total_token_usage["total_tokens"] = ( - self.total_token_usage.get("total_tokens", 0) - + token_usage.get("total_tokens", 0) - ) - - # 更新增量追踪 - if last_message_timestamp > self.last_analyzed_message_timestamp: - self.last_analyzed_message_timestamp = last_message_timestamp - self.all_participant_ids.update(participant_ids) - - # 更新时间戳 - self.updated_at = time.time() - - # 创建批次记录 - batch = BatchRecord( - batch_id=self.total_analysis_count, - timestamp=time.time(), - message_count=messages_count, - new_topics_count=len(new_topics), - new_quotes_count=len(new_quotes), - token_usage=dict(token_usage), - ) - self.batch_records.append(batch) - - return batch - - def _is_duplicate_topic(self, new_topic: dict, threshold: float = 0.6) -> bool: - """ - 检测话题是否与已有话题重复。 - - 使用简单的字符重叠相似度判断。 - 当新话题的名称与已有话题名称相似度超过阈值时，认为是重复话题。 - - Args: - new_topic: 待检测的新话题 - threshold: 相似度阈值（0-1），默认 0.6 - - Returns: - bool: 是否重复 - """ - new_name = new_topic.get("topic", "") - if not new_name: - return False - - for existing in self.topics: - existing_name = existing.get("topic", "") - if not existing_name: - continue - similarity = self._char_overlap_similarity(new_name, existing_name) - if similarity >= threshold: - return True - return False - - def _is_duplicate_quote(self, new_quote: dict, threshold: float = 0.7) -> bool: - """ - 检测金句是否与已有金句重复。 - - Args: - new_quote: 待检测的新金句 - threshold: 相似度阈值（0-1），默认 0.7 - - Returns: - bool: 是否重复 - """ - new_content = new_quote.get("content", "") - if not new_content: - return False - - for existing in self.golden_quotes: - existing_content = existing.get("content", "") - if not existing_content: - continue - similarity = self._char_overlap_similarity(new_content, existing_content) - if similarity >= threshold: - return True - return False - - @staticmethod - def _char_overlap_similarity(s1: str, s2: str) -> float: - """ - 计算两个字符串的字符重叠相似度。 - - 使用 Jaccard 相似系数：交集大小 / 并集大小。 - - Args: - s1: 第一个字符串 - s2: 第二个字符串 - - Returns: - float: 相似度值（0-1） - """ - if not s1 or not s2: - return 0.0 - set1 = set(s1) - set2 = set(s2) - intersection = set1 & set2 - union = set1 | set2 - if not union: - return 0.0 - return len(intersection) / len(union) - def get_peak_hours(self, top_n: int = 3) -> list[int]: """ 获取消息最活跃的时段。 @@ -365,71 +246,22 @@ class IncrementalState: users.sort(key=lambda x: x["message_count"], reverse=True) return users[:top_n] - def to_dict(self) -> dict: + def get_window_date_str(self) -> str: """ - 序列化为字典，用于 KV 存储持久化。 + 获取窗口的日期范围字符串，用于报告显示。 Returns: - dict: 可 JSON 序列化的字典 + str: 如 "2024-01-15" 或 "2024-01-14 ~ 2024-01-15" """ - return { - "group_id": self.group_id, - "date_str": self.date_str, - "topics": self.topics, - "golden_quotes": self.golden_quotes, - "hourly_message_counts": self.hourly_message_counts, - "hourly_character_counts": self.hourly_character_counts, - "user_activities": self.user_activities, - "emoji_counts": self.emoji_counts, - "batch_records": [b.to_dict() for b in self.batch_records], - "total_message_count": self.total_message_count, - "total_character_count": self.total_character_count, - "total_analysis_count": self.total_analysis_count, - "total_token_usage": self.total_token_usage, - "last_analyzed_message_timestamp": self.last_analyzed_message_timestamp, - "all_participant_ids": list(self.all_participant_ids), - "created_at": self.created_at, - "updated_at": self.updated_at, - } + if self.window_start <= 0 or self.window_end <= 0: + return datetime.now().strftime("%Y-%m-%d") - @classmethod - def from_dict(cls, data: dict) -> "IncrementalState": - """ - 从字典反序列化。 + start_date = datetime.fromtimestamp(self.window_start).strftime("%Y-%m-%d") + end_date = datetime.fromtimestamp(self.window_end).strftime("%Y-%m-%d") - Args: - data: 从 KV 存储读取的字典数据 - - Returns: - IncrementalState: 重建的状态实例 - """ - state = cls( - group_id=data.get("group_id", ""), - date_str=data.get("date_str", ""), - topics=data.get("topics", []), - golden_quotes=data.get("golden_quotes", []), - hourly_message_counts=data.get("hourly_message_counts", {}), - hourly_character_counts=data.get("hourly_character_counts", {}), - user_activities=data.get("user_activities", {}), - emoji_counts=data.get("emoji_counts", {}), - batch_records=[ - BatchRecord.from_dict(b) - for b in data.get("batch_records", []) - ], - total_message_count=data.get("total_message_count", 0), - total_character_count=data.get("total_character_count", 0), - total_analysis_count=data.get("total_analysis_count", 0), - total_token_usage=data.get("total_token_usage", { - "prompt_tokens": 0, - "completion_tokens": 0, - "total_tokens": 0, - }), - last_analyzed_message_timestamp=data.get("last_analyzed_message_timestamp", 0), - all_participant_ids=set(data.get("all_participant_ids", [])), - created_at=data.get("created_at", time.time()), - updated_at=data.get("updated_at", time.time()), - ) - return state + if start_date == end_date: + return end_date + return f"{start_date} ~ {end_date}" def get_summary(self) -> dict: """ @@ -440,7 +272,7 @@ class IncrementalState: """ return { "group_id": self.group_id, - "date": self.date_str, + "window": self.get_window_date_str(), "total_messages": self.total_message_count, "total_characters": self.total_character_count, "total_analyses": self.total_analysis_count, @@ -455,3 +287,88 @@ class IncrementalState: ), "peak_hours": self.get_peak_hours(3), } + + @staticmethod + def is_duplicate_topic( + new_topic: dict, existing_topics: list[dict], threshold: float = 0.6 + ) -> bool: + """ + 检测话题是否与已有话题重复。 + + 使用简单的字符重叠相似度判断。 + 当新话题的名称与已有话题名称相似度超过阈值时，认为是重复话题。 + + Args: + new_topic: 待检测的新话题 + existing_topics: 已有话题列表 + threshold: 相似度阈值（0-1），默认 0.6 + + Returns: + bool: 是否重复 + """ + new_name = new_topic.get("topic", "") + if not new_name: + return False + + for existing in existing_topics: + existing_name = existing.get("topic", "") + if not existing_name: + continue + similarity = IncrementalState.char_overlap_similarity( + new_name, existing_name + ) + if similarity >= threshold: + return True + return False + + @staticmethod + def is_duplicate_quote( + new_quote: dict, existing_quotes: list[dict], threshold: float = 0.7 + ) -> bool: + """ + 检测金句是否与已有金句重复。 + + Args: + new_quote: 待检测的新金句 + existing_quotes: 已有金句列表 + threshold: 相似度阈值（0-1），默认 0.7 + + Returns: + bool: 是否重复 + """ + new_content = new_quote.get("content", "") + if not new_content: + return False + + for existing in existing_quotes: + existing_content = existing.get("content", "") + if not existing_content: + continue + similarity = IncrementalState.char_overlap_similarity( + new_content, existing_content + ) + if similarity >= threshold: + return True + return False + + @staticmethod + def char_overlap_similarity(s1: str, s2: str) -> float: + """ + 计算两个字符串的字符重叠相似度（Jaccard 相似系数）。 + + Args: + s1: 第一个字符串 + s2: 第二个字符串 + + Returns: + float: 相似度值（0-1） + """ + if not s1 or not s2: + return 0.0 + set1 = set(s1) + set2 = set(s2) + intersection = set1 & set2 + union = set1 | set2 + if not union: + return 0.0 + return len(intersection) / len(union) diff --git a/src/domain/services/incremental_merge_service.py b/src/domain/services/incremental_merge_service.py index 3cae380..92ddee0 100644 --- a/src/domain/services/incremental_merge_service.py +++ b/src/domain/services/incremental_merge_service.py @@ -1,16 +1,20 @@ """ 增量合并领域服务 -负责将 IncrementalState 累积数据转换为现有实体类型， +负责将 IncrementalBatch 列表合并为 IncrementalState， +以及将 IncrementalState 累积数据转换为现有实体类型， 以便复用现有的报告生成器和分发器。 核心职责： +- merge_batches: 将多个 IncrementalBatch 合并为一个 IncrementalState（滑动窗口聚合） - IncrementalState → GroupStatistics（含 ActivityVisualization、EmojiStatistics） - IncrementalState → list[SummaryTopic] - IncrementalState → list[GoldenQuote] """ -from ...domain.entities.incremental_state import IncrementalState +import time + +from ...domain.entities.incremental_state import IncrementalBatch, IncrementalState from ...domain.models.data_models import ( ActivityVisualization, EmojiStatistics, @@ -26,10 +30,126 @@ class IncrementalMergeService: """ 增量合并服务 - 将一天内累积的增量分析状态转换为现有报告系统所需的数据结构， + 将滑动窗口内的多个批次数据合并为报告所需的数据结构， 确保增量模式下生成的最终报告与传统单次分析报告格式完全一致。 """ + def merge_batches( + self, + batches: list[IncrementalBatch], + window_start: float, + window_end: float, + ) -> IncrementalState: + """ + 从批次列表合并构建 IncrementalState。 + + 遍历所有批次，累加统计数据并对话题和金句执行去重， + 生成可用于报告的聚合视图。 + + Args: + batches: 时间窗口内的批次列表（按时间升序） + window_start: 窗口起始时间戳（epoch） + window_end: 窗口结束时间戳（epoch） + + Returns: + IncrementalState: 合并后的聚合视图 + """ + state = IncrementalState( + group_id=batches[0].group_id if batches else "", + window_start=window_start, + window_end=window_end, + total_analysis_count=len(batches), + created_at=window_start, + updated_at=time.time(), + ) + + for batch in batches: + # 累加消息和字符计数 + state.total_message_count += batch.messages_count + state.total_character_count += batch.characters_count + + # 合并每小时消息分布（按键累加） + for hour_key, count in batch.hourly_msg_counts.items(): + hour_str = str(hour_key) + state.hourly_message_counts[hour_str] = ( + state.hourly_message_counts.get(hour_str, 0) + count + ) + + # 合并每小时字符分布 + for hour_key, count in batch.hourly_char_counts.items(): + hour_str = str(hour_key) + state.hourly_character_counts[hour_str] = ( + state.hourly_character_counts.get(hour_str, 0) + count + ) + + # 合并用户统计（按用户累加消息数、字符数等） + for user_id, stats in batch.user_stats.items(): + if user_id not in state.user_activities: + state.user_activities[user_id] = { + "name": stats.get("name", user_id), + "message_count": 0, + "char_count": 0, + "emoji_count": 0, + "active_hours": [], + "last_message_time": 0, + } + existing = state.user_activities[user_id] + existing["message_count"] += stats.get("message_count", 0) + existing["char_count"] += stats.get("char_count", 0) + existing["emoji_count"] += stats.get("emoji_count", 0) + # 合并活跃小时（去重） + existing_hours = set(existing.get("active_hours", [])) + existing_hours.update(stats.get("active_hours", [])) + existing["active_hours"] = list(existing_hours) + # 取最后消息时间的较大值 + batch_last = stats.get("last_message_time", 0) + if batch_last > existing.get("last_message_time", 0): + existing["last_message_time"] = batch_last + # 更新昵称（使用最新批次的昵称） + name = stats.get("name", "") + if name: + existing["name"] = name + + # 合并表情统计（按键累加） + for emoji_key, count in batch.emoji_stats.items(): + state.emoji_counts[emoji_key] = ( + state.emoji_counts.get(emoji_key, 0) + count + ) + + # 合并话题（去重） + for topic in batch.topics: + if not IncrementalState.is_duplicate_topic(topic, state.topics): + state.topics.append(topic) + + # 合并金句（去重） + for quote in batch.golden_quotes: + if not IncrementalState.is_duplicate_quote(quote, state.golden_quotes): + state.golden_quotes.append(quote) + + # 累加 token 消耗 + for token_key in ("prompt_tokens", "completion_tokens", "total_tokens"): + state.total_token_usage[token_key] = ( + state.total_token_usage.get(token_key, 0) + + batch.token_usage.get(token_key, 0) + ) + + # 合并参与者 ID（取并集） + state.all_participant_ids.update(batch.participant_ids) + + # 记录最后分析消息时间戳（取最大值） + if batch.last_message_timestamp > state.last_analyzed_message_timestamp: + state.last_analyzed_message_timestamp = batch.last_message_timestamp + + logger.info( + f"合并批次完成: 群={state.group_id}, " + f"窗口={state.get_window_date_str()}, " + f"批次数={len(batches)}, " + f"总消息={state.total_message_count}, " + f"话题={len(state.topics)}, 金句={len(state.golden_quotes)}" + ) + + return state + def build_final_statistics(self, state: IncrementalState) -> GroupStatistics: """ 从增量状态构建最终的群组统计数据。 @@ -38,7 +158,7 @@ class IncrementalMergeService: 包含完整的 24 小时活跃度分布、表情统计和 token 消耗。 Args: - state: 当天的增量分析状态 + state: 由 merge_batches 合并生成的增量分析状态 Returns: GroupStatistics: 与传统分析格式一致的统计数据 @@ -58,7 +178,7 @@ class IncrementalMergeService: # 构建活跃度可视化数据 activity_visualization = ActivityVisualization( hourly_activity=hourly_activity, - daily_activity={state.date_str: state.total_message_count}, + daily_activity={state.get_window_date_str(): state.total_message_count}, user_activity_ranking=user_ranking, peak_hours=peak_hours, activity_heatmap_data={}, @@ -106,7 +226,7 @@ class IncrementalMergeService: 将 IncrementalState 中累积的话题字典转换为 SummaryTopic 实例列表。 Args: - state: 当天的增量分析状态 + state: 由 merge_batches 合并生成的增量分析状态 Returns: list[SummaryTopic]: 话题列表，格式与传统分析结果一致 @@ -130,7 +250,7 @@ class IncrementalMergeService: 将 IncrementalState 中累积的金句字典转换为 GoldenQuote 实例列表。 Args: - state: 当天的增量分析状态 + state: 由 merge_batches 合并生成的增量分析状态 Returns: list[GoldenQuote]: 金句列表，格式与传统分析结果一致 @@ -160,7 +280,7 @@ class IncrementalMergeService: 返回的 analysis_result 完全一致，可直接传入 ReportDispatcher。 Args: - state: 当天的增量分析状态 + state: 由 merge_batches 合并生成的增量分析状态 user_titles: 用户称号列表（由最终报告时 LLM 分析生成） Returns: @@ -182,7 +302,7 @@ class IncrementalMergeService: logger.info( f"从增量状态构建完整分析结果: " - f"群={state.group_id}, 日期={state.date_str}, " + f"群={state.group_id}, 窗口={state.get_window_date_str()}, " f"消息={state.total_message_count}, " f"话题={len(topics)}, " f"金句={len(golden_quotes)}, " diff --git a/src/infrastructure/persistence/incremental_store.py b/src/infrastructure/persistence/incremental_store.py index e255db4..07ed819 100644 --- a/src/infrastructure/persistence/incremental_store.py +++ b/src/infrastructure/persistence/incremental_store.py @@ -1,186 +1,440 @@ """ -增量分析状态持久化存储 - 基础设施持久化层 +增量分析批次持久化存储 — 滑动窗口架构 -负责增量分析状态的存储和读取。 -使用 AstrBot 的 put_kv_data/get_kv_data 实现， -每个群聊每天对应一个独立的状态键。 +基于 AstrBot 的 put_kv_data/get_kv_data 实现按批次独立存储， +支持按时间窗口查询批次、批次索引管理和过期批次清理。 -键格式: incremental_state_{group_id}_{date_str} +KV 键设计： +- 批次索引: incr_batch_index_{group_id} + 值: [{"batch_id": "xxx", "timestamp": 1234567890.0}, ...] +- 批次数据: incr_batch_{group_id}_{batch_id} + 值: IncrementalBatch.to_dict() +- 最后分析消息时间戳: incr_last_ts_{group_id} + 值: int (epoch timestamp) """ -import datetime +import time from typing import Any -from ...domain.entities.incremental_state import IncrementalState +from ...domain.entities.incremental_state import IncrementalBatch from ...utils.logger import logger class IncrementalStore: """ - 增量分析状态持久化仓储 + 增量分析批次持久化仓储 - 该类封装了增量分析状态在 KV 存储中的读写操作。 - 每个群组每天的增量状态独立存储，支持创建、读取、更新和删除。 - - 使用方式与 HistoryManager 一致，依赖 star_instance 提供的 - put_kv_data / get_kv_data 异步接口。 + 核心职责： + - save_batch: 保存单个批次数据并更新索引 + - query_batches: 按时间窗口查询批次列表 + - get_last_analyzed_timestamp / update_last_analyzed_timestamp: 跨批次去重 + - cleanup_old_batches: 清理过期批次 + - get_batch_count: 获取当前批次总数（状态查询用） """ - # KV 存储键前缀 - KEY_PREFIX = "incremental_state" + # KV 键前缀 + INDEX_PREFIX = "incr_batch_index" + BATCH_PREFIX = "incr_batch" + LAST_TS_PREFIX = "incr_last_ts" def __init__(self, star_instance: Any): """ - 初始化增量状态仓储。 + 初始化批次持久化仓储。 Args: star_instance: Star 插件实例，用于访问底层 KV 存储引擎 """ self.plugin = star_instance - def _build_key(self, group_id: str, date_str: str | None = None) -> str: + # ================================================================ + # 键构建 + # ================================================================ + + def _index_key(self, group_id: str) -> str: + """构建批次索引键""" + return f"{self.INDEX_PREFIX}_{group_id}" + + def _batch_key(self, group_id: str, batch_id: str) -> str: + """构建单个批次数据键""" + return f"{self.BATCH_PREFIX}_{group_id}_{batch_id}" + + def _last_ts_key(self, group_id: str) -> str: + """构建最后分析消息时间戳键""" + return f"{self.LAST_TS_PREFIX}_{group_id}" + + # ================================================================ + # 批次索引操作 + # ================================================================ + + async def _get_index(self, group_id: str) -> list[dict]: """ - 构建 KV 存储键。 + 获取指定群的批次索引列表。 Args: group_id: 群组 ID - date_str: 日期字符串 (YYYY-MM-DD)，缺省为当天 Returns: - str: 格式为 "incremental_state_{group_id}_{date_str}" 的键 + list[dict]: 索引条目列表，每项包含 batch_id 和 timestamp """ - if not date_str: - date_str = datetime.datetime.now().strftime("%Y-%m-%d") - return f"{self.KEY_PREFIX}_{group_id}_{date_str}" - - async def get_state( - self, group_id: str, date_str: str | None = None - ) -> IncrementalState | None: - """ - 读取指定群组在指定日期的增量分析状态。 - - Args: - group_id: 群组 ID - date_str: 日期字符串 (YYYY-MM-DD)，缺省为当天 - - Returns: - IncrementalState | None: 状态实例，不存在则返回 None - """ - if not date_str: - date_str = datetime.datetime.now().strftime("%Y-%m-%d") - - key = self._build_key(group_id, date_str) - + key = self._index_key(group_id) try: data = await self.plugin.get_kv_data(key, None) if data is None: - return None - - state = IncrementalState.from_dict(data) - logger.debug(f"已读取群 {group_id} 在 {date_str} 的增量状态 (Key: {key})") - return state + return [] + if isinstance(data, list): + return data + logger.warning(f"批次索引数据格式异常 (Key: {key}): {type(data)}") + return [] except Exception as e: - logger.error(f"读取增量状态失败 (Key: {key}): {e}", exc_info=True) - return None + logger.error(f"读取批次索引失败 (Key: {key}): {e}", exc_info=True) + return [] - async def save_state(self, state: IncrementalState) -> bool: + async def _save_index(self, group_id: str, index: list[dict]) -> None: """ - 持久化增量分析状态。 - - 将状态序列化为字典后写入 KV 存储。 - 如果已存在同键数据则覆盖更新。 + 保存批次索引列表。 Args: - state: 要保存的增量分析状态实例 + group_id: 群组 ID + index: 索引条目列表 + """ + key = self._index_key(group_id) + try: + await self.plugin.put_kv_data(key, index) + except Exception as e: + logger.error(f"保存批次索引失败 (Key: {key}): {e}", exc_info=True) + raise + + # ================================================================ + # 批次数据操作 + # ================================================================ + + async def save_batch(self, batch: IncrementalBatch) -> bool: + """ + 保存单个批次数据并更新索引。 + + 流程： + 1. 将批次数据写入独立 KV 键 + 2. 将批次元数据（batch_id + timestamp）追加到索引 + + Args: + batch: 要保存的增量分析批次 Returns: bool: 保存是否成功 """ - key = self._build_key(state.group_id, state.date_str) + group_id = batch.group_id + batch_key = self._batch_key(group_id, batch.batch_id) try: - data = state.to_dict() - await self.plugin.put_kv_data(key, data) + # 1. 保存批次数据 + await self.plugin.put_kv_data(batch_key, batch.to_dict()) + + # 2. 更新索引 + index = await self._get_index(group_id) + index.append({ + "batch_id": batch.batch_id, + "timestamp": batch.timestamp, + }) + await self._save_index(group_id, index) + logger.debug( - f"已保存群 {state.group_id} 在 {state.date_str} 的增量状态 " - f"(Key: {key}, 批次数: {state.total_analysis_count})" + f"已保存批次 {batch.batch_id[:8]}... " + f"(群 {group_id}, 消息数={batch.messages_count})" ) return True except Exception as e: - logger.error(f"保存增量状态失败 (Key: {key}): {e}", exc_info=True) + logger.error( + f"保存批次失败 (群 {group_id}, 批次 {batch.batch_id[:8]}...): {e}", + exc_info=True, + ) return False - async def get_or_create_state( - self, group_id: str, date_str: str | None = None - ) -> IncrementalState: + async def query_batches( + self, + group_id: str, + window_start: float, + window_end: float, + ) -> list[IncrementalBatch]: """ - 获取或创建增量分析状态。 + 按时间窗口查询批次列表。 - 如果指定群组在指定日期已有状态则返回现有状态， - 否则创建一个新的空白状态实例（不自动持久化）。 + 从索引中筛选时间戳落在 [window_start, window_end] 范围内的批次， + 逐个加载完整批次数据。 Args: group_id: 群组 ID - date_str: 日期字符串 (YYYY-MM-DD)，缺省为当天 + window_start: 窗口起始时间戳（epoch） + window_end: 窗口结束时间戳（epoch） Returns: - IncrementalState: 现有或新创建的状态实例 + list[IncrementalBatch]: 符合窗口范围的批次列表，按时间戳升序 """ - if not date_str: - date_str = datetime.datetime.now().strftime("%Y-%m-%d") + index = await self._get_index(group_id) - existing = await self.get_state(group_id, date_str) - if existing is not None: - return existing + # 筛选在窗口范围内的批次 + matching_entries = [ + entry for entry in index + if window_start <= entry.get("timestamp", 0) <= window_end + ] - # 创建新的空白状态 - new_state = IncrementalState( - group_id=group_id, - date_str=date_str, + # 按时间戳升序排列 + matching_entries.sort(key=lambda x: x.get("timestamp", 0)) + + batches: list[IncrementalBatch] = [] + for entry in matching_entries: + batch_id = entry.get("batch_id", "") + if not batch_id: + continue + + batch_key = self._batch_key(group_id, batch_id) + try: + data = await self.plugin.get_kv_data(batch_key, None) + if data is not None: + batch = IncrementalBatch.from_dict(data) + batches.append(batch) + else: + logger.warning( + f"批次数据缺失 (群 {group_id}, 批次 {batch_id[:8]}...)" + ) + except Exception as e: + logger.error( + f"加载批次数据失败 (群 {group_id}, 批次 {batch_id[:8]}...): {e}", + exc_info=True, + ) + + logger.debug( + f"窗口查询完成: 群 {group_id}, " + f"窗口 [{window_start:.0f}, {window_end:.0f}], " + f"匹配 {len(batches)}/{len(index)} 个批次" ) - logger.info(f"为群 {group_id} 创建了 {date_str} 的新增量状态") - return new_state - async def delete_state( - self, group_id: str, date_str: str | None = None + return batches + + # ================================================================ + # 最后分析消息时间戳（跨批次去重用） + # ================================================================ + + async def get_last_analyzed_timestamp(self, group_id: str) -> int: + """ + 获取指定群的最后分析消息时间戳。 + + 用于增量分析时过滤已分析过的消息。 + + Args: + group_id: 群组 ID + + Returns: + int: 最后分析消息的 epoch 时间戳，不存在则返回 0 + """ + key = self._last_ts_key(group_id) + try: + data = await self.plugin.get_kv_data(key, 0) + return int(data) if data else 0 + except Exception as e: + logger.error(f"读取最后分析时间戳失败 (Key: {key}): {e}", exc_info=True) + return 0 + + async def update_last_analyzed_timestamp( + self, group_id: str, timestamp: int + ) -> None: + """ + 更新指定群的最后分析消息时间戳。 + + Args: + group_id: 群组 ID + timestamp: 最后分析消息的 epoch 时间戳 + """ + key = self._last_ts_key(group_id) + try: + await self.plugin.put_kv_data(key, timestamp) + logger.debug( + f"更新最后分析时间戳: 群 {group_id}, ts={timestamp}" + ) + except Exception as e: + logger.error( + f"更新最后分析时间戳失败 (Key: {key}): {e}", exc_info=True + ) + raise + + # ================================================================ + # 过期批次清理 + # ================================================================ + + async def cleanup_old_batches( + self, group_id: str, before_timestamp: float + ) -> int: + """ + 清理指定群中早于给定时间戳的所有批次。 + + 流程： + 1. 从索引中分离出过期条目和保留条目 + 2. 逐个删除过期批次的 KV 数据 + 3. 用保留条目覆盖索引 + + Args: + group_id: 群组 ID + before_timestamp: 清理此时间戳之前的所有批次 + + Returns: + int: 已清理的批次数量 + """ + index = await self._get_index(group_id) + if not index: + return 0 + + # 分离过期和保留 + expired = [] + retained = [] + for entry in index: + if entry.get("timestamp", 0) < before_timestamp: + expired.append(entry) + else: + retained.append(entry) + + if not expired: + return 0 + + # 删除过期批次数据 + deleted_count = 0 + for entry in expired: + batch_id = entry.get("batch_id", "") + if not batch_id: + continue + batch_key = self._batch_key(group_id, batch_id) + try: + await self.plugin.put_kv_data(batch_key, None) + deleted_count += 1 + except Exception as e: + logger.error( + f"删除过期批次失败 (群 {group_id}, 批次 {batch_id[:8]}...): {e}", + exc_info=True, + ) + + # 更新索引（仅保留未过期条目） + await self._save_index(group_id, retained) + + logger.info( + f"清理过期批次: 群 {group_id}, " + f"删除 {deleted_count} 个, 保留 {len(retained)} 个" + ) + + return deleted_count + + # ================================================================ + # 状态查询 + # ================================================================ + + async def get_batch_count(self, group_id: str) -> int: + """ + 获取指定群的当前批次总数。 + + Args: + group_id: 群组 ID + + Returns: + int: 批次总数 + """ + index = await self._get_index(group_id) + return len(index) + + async def get_all_batch_summaries( + self, group_id: str + ) -> list[dict]: + """ + 获取指定群所有批次的摘要信息（不加载完整数据）。 + + 用于状态查询命令展示批次概览。 + + Args: + group_id: 群组 ID + + Returns: + list[dict]: 批次摘要列表，按时间升序 + """ + index = await self._get_index(group_id) + # 按时间戳升序排列 + index.sort(key=lambda x: x.get("timestamp", 0)) + return index + + # ================================================================ + # 旧版兼容（迁移期间使用） + # ================================================================ + + async def migrate_legacy_state( + self, group_id: str, date_str: str ) -> bool: """ - 删除指定群组在指定日期的增量分析状态。 + 尝试迁移旧版按天存储的 IncrementalState 到新批次架构。 - 通过将键值设为 None 来实现删除效果。 + 检查旧键 incremental_state_{group_id}_{date_str} 是否存在， + 如果存在则将其数据转换为一个 IncrementalBatch 并保存， + 然后删除旧键。 Args: group_id: 群组 ID - date_str: 日期字符串 (YYYY-MM-DD)，缺省为当天 + date_str: 日期字符串 (YYYY-MM-DD) Returns: - bool: 删除是否成功 + bool: 是否成功迁移（True=迁移了数据，False=无需迁移或失败） """ - if not date_str: - date_str = datetime.datetime.now().strftime("%Y-%m-%d") - - key = self._build_key(group_id, date_str) + old_key = f"incremental_state_{group_id}_{date_str}" try: - await self.plugin.put_kv_data(key, None) - logger.info(f"已删除群 {group_id} 在 {date_str} 的增量状态 (Key: {key})") + old_data = await self.plugin.get_kv_data(old_key, None) + if old_data is None: + return False + + logger.info( + f"发现旧版增量状态 (群 {group_id}, 日期 {date_str})，开始迁移" + ) + + # 从旧数据中提取信息构建一个聚合批次 + batch = IncrementalBatch( + group_id=group_id, + timestamp=old_data.get("created_at", time.time()), + messages_count=old_data.get("total_message_count", 0), + characters_count=old_data.get("total_character_count", 0), + hourly_msg_counts=old_data.get("hourly_message_counts", {}), + hourly_char_counts=old_data.get("hourly_character_counts", {}), + user_stats=old_data.get("user_activities", {}), + emoji_stats=old_data.get("emoji_counts", {}), + topics=old_data.get("topics", []), + golden_quotes=old_data.get("golden_quotes", []), + token_usage=old_data.get("total_token_usage", { + "prompt_tokens": 0, + "completion_tokens": 0, + "total_tokens": 0, + }), + last_message_timestamp=old_data.get( + "last_analyzed_message_timestamp", 0 + ), + participant_ids=list( + old_data.get("all_participant_ids", []) + ), + ) + + # 保存为新批次 + saved = await self.save_batch(batch) + if not saved: + logger.error(f"旧版数据迁移保存失败 (群 {group_id})") + return False + + # 迁移最后分析时间戳 + last_ts = old_data.get("last_analyzed_message_timestamp", 0) + if last_ts > 0: + await self.update_last_analyzed_timestamp(group_id, last_ts) + + # 删除旧键 + await self.plugin.put_kv_data(old_key, None) + + logger.info( + f"旧版增量状态迁移完成 (群 {group_id}, 日期 {date_str}), " + f"消息数={batch.messages_count}" + ) return True + except Exception as e: - logger.error(f"删除增量状态失败 (Key: {key}): {e}", exc_info=True) + logger.error( + f"旧版增量状态迁移失败 (群 {group_id}): {e}", exc_info=True + ) return False - - async def has_state( - self, group_id: str, date_str: str | None = None - ) -> bool: - """ - 判断指定群组在指定日期是否存在增量分析状态。 - - Args: - group_id: 群组 ID - date_str: 日期字符串 (YYYY-MM-DD)，缺省为当天 - - Returns: - bool: 是否存在状态 - """ - state = await self.get_state(group_id, date_str) - return state is not None diff --git a/src/infrastructure/scheduler/auto_scheduler.py b/src/infrastructure/scheduler/auto_scheduler.py index b5f00d2..cf09795 100644 --- a/src/infrastructure/scheduler/auto_scheduler.py +++ b/src/infrastructure/scheduler/auto_scheduler.py @@ -4,6 +4,7 @@ """ import asyncio +import time as time_mod import weakref from apscheduler.triggers.cron import CronTrigger @@ -575,12 +576,12 @@ class AutoScheduler: return result # 增量分析只累积数据，不发送报告 - batch = result.get("batch_record", {}) + batch_summary = result.get("batch_summary", {}) logger.info( f"群 {group_id} 增量分析完成: " f"消息数={result.get('messages_count', 0)}, " - f"新话题={batch.get('topics_added', 0)}, " - f"新金句={batch.get('quotes_added', 0)}" + f"话题={batch_summary.get('topics_count', 0)}, " + f"金句={batch_summary.get('quotes_count', 0)}" ) return result @@ -733,6 +734,24 @@ class AutoScheduler: else target_platform_id, ) + # 清理过期批次（保留 2 倍窗口范围的数据作为缓冲） + try: + analysis_days = self.config_manager.get_analysis_days() + before_ts = time_mod.time() - (analysis_days * 2 * 24 * 3600) + incremental_store = self.analysis_service.incremental_store + if incremental_store: + cleaned = await incremental_store.cleanup_old_batches( + group_id, before_ts + ) + if cleaned > 0: + logger.info( + f"群 {group_id} 报告发送后清理了 {cleaned} 个过期批次" + ) + except Exception as cleanup_err: + logger.warning( + f"群 {group_id} 过期批次清理失败（不影响报告）: {cleanup_err}" + ) + logger.info(f"群 {group_id} 增量最终报告发送成功") return result