nezumi/astrbot_plugin_qq_group_daily_analysis
1
0
Fork 0
mirror of https://github.com/Nezumi-2711/astrbot_plugin_qq_group_daily_analysis.git synced 2026-09-22 13:38:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(增量分析): 从按天存储改为滑动窗口批次架构

Browse Source 
- IncrementalBatch: 独立批次实体，每次增量分析产生一个，按批次独立存储到KV
- IncrementalState: 聚合视图，不再持久化，报告时由merge_batches合并产生
- IncrementalStore: 批次索引+数据KV持久化，支持按时间窗口查询和过期清理
- IncrementalMergeService: 新增merge_batches方法，负责批次合并和话题/金句去重
- AnalysisApplicationService: 增量分析存独立批次，最终报告按窗口查询合并
- AutoScheduler: 报告发送后清理2×窗口外的过期批次
- main.py: /增量状态命令改为滑动窗口查询展示
- 消除天然日期隔离问题，24h图表展示完整数据
This commit is contained in:
SXP-Simon
2026-02-10 18:15:46 +08:00
parent 9aa502f01a
commit 83b82de2bb
8 changed files with 920 additions and 532 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/INCREMENTAL_ANALYSIS.md
+84 -41
View File
@@ -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 |
main.py
+29 -10
View File
@@ -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:
src/application/services/analysis_application_service.py
+68 -52
View File
@@ -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] = {}
src/domain/entities/__init__.py
+4 -4
View File
@@ -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",
]
src/domain/entities/incremental_state.py
+221 -304
View File
@@ -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)
src/domain/services/incremental_merge_service.py
+129 -9
View File
@@ -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)}, "
src/infrastructure/persistence/incremental_store.py
+362 -108
View File
@@ -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.info(f"为群 {group_id} 创建了 {date_str} 的新增量状态")
return new_state
async def delete_state(
self, group_id: str, date_str: str | None = None
logger.debug(
f"窗口查询完成: 群 {group_id}, "
f"窗口 [{window_start:.0f}, {window_end:.0f}], "
f"匹配 {len(batches)}/{len(index)} 个批次"
)
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})")
return True
except Exception as e:
logger.error(f"删除增量状态失败 (Key: {key}): {e}", exc_info=True)
old_data = await self.plugin.get_kv_data(old_key, None)
if old_data is None:
return False
async def has_state(
self, group_id: str, date_str: str | None = None
) -> bool:
"""
判断指定群组在指定日期是否存在增量分析状态。
logger.info(
f"发现旧版增量状态 (群 {group_id}, 日期 {date_str})，开始迁移"
)
Args:
group_id: 群组 ID
date_str: 日期字符串 (YYYY-MM-DD)，缺省为当天
# 从旧数据中提取信息构建一个聚合批次
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", [])
),
)
Returns:
bool: 是否存在状态
"""
state = await self.get_state(group_id, date_str)
return state is not None
# 保存为新批次
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"旧版增量状态迁移失败 (群 {group_id}): {e}", exc_info=True
)
return False
src/infrastructure/scheduler/auto_scheduler.py
+22 -3
View File
@@ -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