diff --git a/_conf_schema.json b/_conf_schema.json index b158880..c1132f4 100644 --- a/_conf_schema.json +++ b/_conf_schema.json @@ -116,7 +116,7 @@ "max_concurrent_tasks": { "type": "int", "description": "自动分析最大并发数", - "default": 3, + "default": 1, "hint": "同时进行的群聊分析任务数量，建议根据机器性能和服务商情况调整，过高可能导致LLM API RPM 超出限制，卡顿或被风控" } } diff --git a/main.py b/main.py index 07b0a44..cd846ed 100644 --- a/main.py +++ b/main.py @@ -515,6 +515,8 @@ class GroupDailyAnalysis(Star): if not await adapter.send_text(group_id, text_report): yield event.plain_result(text_report) + except asyncio.CancelledError: + yield event.plain_result("📊 该群的分析任务正在执行中，请稍后再试哦~") except Exception as e: logger.error(f"群分析失败: {e}", exc_info=True) yield event.plain_result( @@ -787,6 +789,8 @@ class GroupDailyAnalysis(Star): try: await self.auto_scheduler._perform_auto_analysis_for_group(group_id) yield event.plain_result("✅ 自动分析测试完成，请查看群消息") + except asyncio.CancelledError: + yield event.plain_result("📊 该群的分析任务正在执行中，请稍后再试哦~") except Exception as e: yield event.plain_result(f"❌ 自动分析测试失败: {str(e)}") diff --git a/src/application/services/analysis_application_service.py b/src/application/services/analysis_application_service.py index ec45ae2..54fe2bc 100644 --- a/src/application/services/analysis_application_service.py +++ b/src/application/services/analysis_application_service.py @@ -7,7 +7,9 @@ import asyncio import datetime as dt import time as time_mod +import weakref from collections import defaultdict +from contextlib import asynccontextmanager from typing import Any from ...domain.entities.incremental_state import IncrementalBatch @@ -46,6 +48,37 @@ class AnalysisApplicationService: self.analysis_domain_service = analysis_domain_service self.incremental_store = incremental_store self.incremental_merge_service = incremental_merge_service + self._locks = weakref.WeakValueDictionary() + # 全局 LLM 分析信号量，控制对外 API 的并发压力 + max_concurrent = self.config_manager.get_max_concurrent_tasks() + self.llm_semaphore = asyncio.Semaphore(max_concurrent) + + @asynccontextmanager + async def group_lock(self, group_id: str, task_type: str = "analysis"): + """ + 同一时间、同一个群、同一种任务只能有一个在执行 + 锁将在退出上下文时自动释放。 + """ + lock_key = f"{task_type}:{group_id}" + + # 获取或创建该群专属的异步锁 + lock = self._locks.get(lock_key) + if lock is None: + lock = asyncio.Lock() + self._locks[lock_key] = lock + + # 检查是否已经锁定（防止并发） + if lock.locked(): + logger.warning(f"群 {group_id} 的 {task_type} 任务已在运行，跳过本次请求") + # 这里抛出异常以便上层识别并优雅跳过 + raise asyncio.CancelledError(f"Duplicate task for {lock_key}") + + async with lock: + logger.debug(f"[Lock] 已获取群 {group_id} 的 {task_type} 排他锁") + try: + yield + finally: + logger.debug(f"[Lock] 已释放群 {group_id} 的 {task_type} 排他锁") async def execute_daily_analysis( self, group_id: str, platform_id: str | None = None, manual: bool = False @@ -63,121 +96,129 @@ class AnalysisApplicationService: 7. 持久化摘要 (Persistence) 8. 返回结果 """ - logger.info(f"开始执行分析用例: 群 {group_id}, 平台 {platform_id or '默认'}") - # 1. 获取适配器 - adapter = self.bot_manager.get_adapter(platform_id) - if not adapter: - raise ValueError(f"未找到平台 {platform_id} 的适配器") - - # 2. 拉取消息 - days = self.config_manager.get_analysis_days() - max_count = self.config_manager.get_max_messages() - - raw_messages = await adapter.fetch_messages( - group_id=group_id, days=days, max_count=max_count - ) - - if not raw_messages: - logger.warning(f"群 {group_id} 在最近 {days} 天内无消息或无法获取") - return {"success": False, "reason": "no_messages"} - - # 3. 清理消息 (Filter commands, bot messages, noise) - from ...domain.services.message_cleaner_service import MessageCleanerService - - cleaner = MessageCleanerService() - bot_self_ids = self.config_manager.get_bot_self_ids() - - # 对于自动任务，强制过滤指令；对于手动任务，也建议过滤以保持报告纯净 - unified_messages = cleaner.clean_messages( - raw_messages, bot_self_ids=bot_self_ids, filter_commands=True - ) - - # 4. 检查最小消息阈值 (在清理后进行) - threshold = self.config_manager.get_min_messages_threshold() - if len(unified_messages) < threshold and not manual: + async with self.group_lock(group_id, "daily"): logger.info( - f"群 {group_id} 有效消息数 ({len(unified_messages)}) 未达到自动分析阈值 ({threshold})" - ) - return {"success": False, "reason": "below_threshold"} - - # 5. 基础统计 (Domain Service) - statistics = await asyncio.to_thread( - self.statistics_service.calculate_group_statistics, unified_messages - ) - - # 4. 用户分析 (Domain Service) - bot_self_ids = self.config_manager.get_bot_self_ids() - user_activity = await asyncio.to_thread( - self.analysis_domain_service.analyze_user_activity, - unified_messages, - bot_self_ids, - ) - - max_user_titles = self.config_manager.get_max_user_titles() - top_users = self.analysis_domain_service.get_top_users( - user_activity, limit=max_user_titles - ) - - # 5. LLM 语义分析 (为了保持兼容，目前直接传 UnifiedMessage，后续如需传 raw dict 再加转换) - # LLMAnalyzer 内部可能已经处理了转换（见之前代码） - topic_enabled = self.config_manager.get_topic_analysis_enabled() - user_title_enabled = self.config_manager.get_user_title_analysis_enabled() - golden_quote_enabled = self.config_manager.get_golden_quote_analysis_enabled() - - topics = [] - user_titles = [] - golden_quotes = [] - total_token_usage = TokenUsage() - - # Note: LLMAnalyzer 目前可能只接收 legacy 格式或特定的 UnifiedMessage 适配 - # 暂时转换回 legacy 格式以确保稳定性，直到 LLMAnalyzer 被重构 - legacy_messages = self.statistics_service._convert_to_legacy_dict( - unified_messages - ) - - unified_msg_origin = ( - f"{platform_id}:GroupMessage:{group_id}" if platform_id else group_id - ) - - if topic_enabled or user_title_enabled or golden_quote_enabled: - ( - topics, - user_titles, - golden_quotes, - total_token_usage, - ) = await self.llm_analyzer.analyze_all_concurrent( - legacy_messages, - user_activity, - umo=unified_msg_origin, - top_users=top_users, - topic_enabled=topic_enabled, - user_title_enabled=user_title_enabled, - golden_quote_enabled=golden_quote_enabled, + f"开始执行分析用例: 群 {group_id}, platform_id={platform_id or '默认'}" ) - # 回填结果 - statistics.golden_quotes = golden_quotes - statistics.token_usage = total_token_usage + # 1. 获取适配器 + adapter = self.bot_manager.get_adapter(platform_id) + if not adapter: + raise ValueError(f"未找到平台 {platform_id} 的适配器") - analysis_result = { - "statistics": statistics, - "topics": topics, - "user_titles": user_titles, - "user_analysis": user_activity, - } + # 2. 拉取消息 + days = self.config_manager.get_analysis_days() + max_count = self.config_manager.get_max_messages() - # 6. 持久化摘要 (Persistence) - await self.history_manager.save_analysis(group_id, analysis_result) + raw_messages = await adapter.fetch_messages( + group_id=group_id, days=days, max_count=max_count + ) - # 7. 生成报告并发送 (应用层编排发送动作) - # 这里由调用方处理发送，本服务只返回分析结果和可能的视觉产物 - return { - "success": True, - "analysis_result": analysis_result, - "messages_count": len(unified_messages), - "adapter": adapter, - } + if not raw_messages: + logger.warning(f"群 {group_id} 在最近 {days} 天内无消息或无法获取") + return {"success": False, "reason": "no_messages"} + + # 3. 清理消息 (Filter commands, bot messages, noise) + from ...domain.services.message_cleaner_service import MessageCleanerService + + cleaner = MessageCleanerService() + bot_self_ids = self.config_manager.get_bot_self_ids() + + # 对于自动任务，强制过滤指令；对于手动任务，也建议过滤以保持报告纯净 + unified_messages = cleaner.clean_messages( + raw_messages, bot_self_ids=bot_self_ids, filter_commands=True + ) + + # 4. 检查最小消息阈值 (在清理后进行) + threshold = self.config_manager.get_min_messages_threshold() + if len(unified_messages) < threshold and not manual: + logger.info( + f"群 {group_id} 有效消息数 ({len(unified_messages)}) 未达到自动分析阈值 ({threshold})" + ) + return {"success": False, "reason": "below_threshold"} + + # 5. 基础统计 (Domain Service) + statistics = await asyncio.to_thread( + self.statistics_service.calculate_group_statistics, unified_messages + ) + + # 4. 用户分析 (Domain Service) + bot_self_ids = self.config_manager.get_bot_self_ids() + user_activity = await asyncio.to_thread( + self.analysis_domain_service.analyze_user_activity, + unified_messages, + bot_self_ids, + ) + + max_user_titles = self.config_manager.get_max_user_titles() + top_users = self.analysis_domain_service.get_top_users( + user_activity, limit=max_user_titles + ) + + # 5. LLM 语义分析 (为了保持兼容，目前直接传 UnifiedMessage，后续如需传 raw dict 再加转换) + # LLMAnalyzer 内部可能已经处理了转换（见之前代码） + topic_enabled = self.config_manager.get_topic_analysis_enabled() + user_title_enabled = self.config_manager.get_user_title_analysis_enabled() + golden_quote_enabled = ( + self.config_manager.get_golden_quote_analysis_enabled() + ) + + topics = [] + user_titles = [] + golden_quotes = [] + total_token_usage = TokenUsage() + + # Note: LLMAnalyzer 目前可能只接收 legacy 格式或特定的 UnifiedMessage 适配 + # 暂时转换回 legacy 格式以确保稳定性，直到 LLMAnalyzer 被重构 + legacy_messages = self.statistics_service._convert_to_legacy_dict( + unified_messages + ) + + unified_msg_origin = ( + f"{platform_id}:GroupMessage:{group_id}" if platform_id else group_id + ) + + if topic_enabled or user_title_enabled or golden_quote_enabled: + async with self.llm_semaphore: + logger.debug(f"[LLM] 已进入分析队列 (群: {group_id})") + ( + topics, + user_titles, + golden_quotes, + total_token_usage, + ) = await self.llm_analyzer.analyze_all_concurrent( + legacy_messages, + user_activity, + umo=unified_msg_origin, + top_users=top_users, + topic_enabled=topic_enabled, + user_title_enabled=user_title_enabled, + golden_quote_enabled=golden_quote_enabled, + ) + + # 回填结果 + statistics.golden_quotes = golden_quotes + statistics.token_usage = total_token_usage + + analysis_result = { + "statistics": statistics, + "topics": topics, + "user_titles": user_titles, + "user_analysis": user_activity, + } + + # 6. 持久化摘要 (Persistence) + await self.history_manager.save_analysis(group_id, analysis_result) + + # 7. 生成报告并发送 (应用层编排发送动作) + # 这里由调用方处理发送，本服务只返回分析结果和可能的视觉产物 + return { + "success": True, + "analysis_result": analysis_result, + "messages_count": len(unified_messages), + "adapter": adapter, + } # ---------------------------------------------------------------- # 增量分析用例 @@ -212,188 +253,195 @@ class AnalysisApplicationService: Returns: dict: 包含 success、batch_summary 等信息 """ - if not self.incremental_store: - raise RuntimeError("增量分析未初始化：缺少 IncrementalStore") + async with self.group_lock(group_id, "incremental"): + if not self.incremental_store: + raise RuntimeError("增量分析未初始化：缺少 IncrementalStore") - logger.info(f"开始增量分析用例: 群 {group_id}, 平台 {platform_id or '默认'}") + logger.info( + f"开始增量分析用例: 群 {group_id}, 平台 {platform_id or '默认'}" + ) - # 1. 获取适配器 - adapter = self.bot_manager.get_adapter(platform_id) - if not adapter: - raise ValueError(f"未找到平台 {platform_id} 的适配器") + # 1. 获取适配器 + adapter = self.bot_manager.get_adapter(platform_id) + if not adapter: + raise ValueError(f"未找到平台 {platform_id} 的适配器") - # 2. 拉取消息（使用增量配置的消息数量上限） - days = self.config_manager.get_analysis_days() - max_count = self.config_manager.get_incremental_max_messages() + # 2. 拉取消息（使用增量配置的消息数量上限） + days = self.config_manager.get_analysis_days() + max_count = self.config_manager.get_incremental_max_messages() - raw_messages = await adapter.fetch_messages( - group_id=group_id, days=days, max_count=max_count - ) + raw_messages = await adapter.fetch_messages( + group_id=group_id, days=days, max_count=max_count + ) - if not raw_messages: - logger.warning(f"群 {group_id} 增量分析：无法获取消息") - return {"success": False, "reason": "no_messages"} + if not raw_messages: + logger.warning(f"群 {group_id} 增量分析：无法获取消息") + return {"success": False, "reason": "no_messages"} - # 3. 清理消息 - from ...domain.services.message_cleaner_service import MessageCleanerService + # 3. 清理消息 + from ...domain.services.message_cleaner_service import MessageCleanerService - cleaner = MessageCleanerService() - bot_self_ids = self.config_manager.get_bot_self_ids() - unified_messages = cleaner.clean_messages( - raw_messages, bot_self_ids=bot_self_ids, filter_commands=True - ) + cleaner = MessageCleanerService() + bot_self_ids = self.config_manager.get_bot_self_ids() + unified_messages = cleaner.clean_messages( + raw_messages, bot_self_ids=bot_self_ids, filter_commands=True + ) - # 4. 按时间戳去重：获取最后分析消息时间戳 - last_analyzed_ts = await self.incremental_store.get_last_analyzed_timestamp( - group_id - ) + # 4. 按时间戳去重：获取最后分析消息时间戳 + last_analyzed_ts = await self.incremental_store.get_last_analyzed_timestamp( + group_id + ) - if last_analyzed_ts > 0: - unified_messages = [ - msg for msg in unified_messages if msg.timestamp > last_analyzed_ts + if last_analyzed_ts > 0: + unified_messages = [ + msg for msg in unified_messages if msg.timestamp > last_analyzed_ts + ] + + # 5. 检查最小消息阈值 + min_messages = self.config_manager.get_incremental_min_messages() + if len(unified_messages) < min_messages: + logger.info( + f"群 {group_id} 增量分析：新消息数 ({len(unified_messages)}) " + f"未达到阈值 ({min_messages})，跳过本次分析" + ) + return {"success": False, "reason": "below_threshold"} + + # 6. 计算基础统计 + statistics = await asyncio.to_thread( + self.statistics_service.calculate_group_statistics, unified_messages + ) + user_activity = await asyncio.to_thread( + self.analysis_domain_service.analyze_user_activity, + unified_messages, + bot_self_ids, + ) + + # 计算本批次的小时分布 + hourly_msg_counts, hourly_char_counts = self._compute_hourly_counts( + unified_messages + ) + + # 7. LLM 增量分析（仅话题 + 金句） + topics_per_batch = self.config_manager.get_incremental_topics_per_batch() + quotes_per_batch = self.config_manager.get_incremental_quotes_per_batch() + + # 获取功能开关状态 + topic_enabled = self.config_manager.get_topic_analysis_enabled() + golden_quote_enabled = ( + self.config_manager.get_golden_quote_analysis_enabled() + ) + + # 需要将 UnifiedMessage 转换为 legacy 格式供 LLM 分析器使用 + legacy_messages = self.statistics_service._convert_to_legacy_dict( + unified_messages + ) + unified_msg_origin = ( + f"{platform_id}:GroupMessage:{group_id}" if platform_id else group_id + ) + + async with self.llm_semaphore: + logger.debug(f"[LLM] 已进入增量分析队列 (群: {group_id})") + ( + topics, + golden_quotes, + token_usage, + ) = await self.llm_analyzer.analyze_incremental_concurrent( + legacy_messages, + umo=unified_msg_origin, + topics_per_batch=topics_per_batch, + quotes_per_batch=quotes_per_batch, + topic_enabled=topic_enabled, + golden_quote_enabled=golden_quote_enabled, + ) + + # 8. 构建 IncrementalBatch + # 8a. 转换话题: SummaryTopic -> dict + new_topics = [ + { + "topic": t.topic, + "contributors": t.contributors, + "detail": t.detail, + "contributor_ids": t.contributor_ids, + } + for t in topics ] - # 5. 检查最小消息阈值 - min_messages = self.config_manager.get_incremental_min_messages() - if len(unified_messages) < min_messages: - logger.info( - f"群 {group_id} 增量分析：新消息数 ({len(unified_messages)}) " - f"未达到阈值 ({min_messages})，跳过本次分析" + # 8b. 转换金句: GoldenQuote -> dict + new_quotes = [ + { + "content": q.content, + "sender": q.sender, + "reason": q.reason, + "user_id": q.user_id, + } + for q in golden_quotes + ] + + # 8c. 转换 token 消耗: TokenUsage -> dict + token_usage_dict = { + "prompt_tokens": token_usage.prompt_tokens, + "completion_tokens": token_usage.completion_tokens, + "total_tokens": token_usage.total_tokens, + } + + # 8d. 转换用户统计: AnalysisDomainService 格式 -> IncrementalBatch 格式 + user_stats = self._convert_user_activity_for_merge( + user_activity, unified_messages ) - return {"success": False, "reason": "below_threshold"} - # 6. 计算基础统计 - statistics = await asyncio.to_thread( - self.statistics_service.calculate_group_statistics, unified_messages - ) - user_activity = await asyncio.to_thread( - self.analysis_domain_service.analyze_user_activity, - unified_messages, - bot_self_ids, - ) - - # 计算本批次的小时分布 - hourly_msg_counts, hourly_char_counts = self._compute_hourly_counts( - unified_messages - ) - - # 7. LLM 增量分析（仅话题 + 金句） - topics_per_batch = self.config_manager.get_incremental_topics_per_batch() - quotes_per_batch = self.config_manager.get_incremental_quotes_per_batch() - - # 获取功能开关状态 - topic_enabled = self.config_manager.get_topic_analysis_enabled() - golden_quote_enabled = self.config_manager.get_golden_quote_analysis_enabled() - - # 需要将 UnifiedMessage 转换为 legacy 格式供 LLM 分析器使用 - legacy_messages = self.statistics_service._convert_to_legacy_dict( - unified_messages - ) - unified_msg_origin = ( - f"{platform_id}:GroupMessage:{group_id}" if platform_id else group_id - ) - - ( - topics, - golden_quotes, - token_usage, - ) = await self.llm_analyzer.analyze_incremental_concurrent( - legacy_messages, - umo=unified_msg_origin, - topics_per_batch=topics_per_batch, - quotes_per_batch=quotes_per_batch, - topic_enabled=topic_enabled, - golden_quote_enabled=golden_quote_enabled, - ) - - # 8. 构建 IncrementalBatch - # 8a. 转换话题: SummaryTopic -> dict - new_topics = [ - { - "topic": t.topic, - "contributors": t.contributors, - "detail": t.detail, - "contributor_ids": t.contributor_ids, + # 8e. 转换表情统计: EmojiStatistics -> dict + emoji_stats = { + "face_count": statistics.emoji_statistics.face_count, + "mface_count": statistics.emoji_statistics.mface_count, + "bface_count": statistics.emoji_statistics.bface_count, + "sface_count": statistics.emoji_statistics.sface_count, + "other_emoji_count": statistics.emoji_statistics.other_emoji_count, + "face_details": statistics.emoji_statistics.face_details, } - for t in topics - ] - # 8b. 转换金句: GoldenQuote -> dict - new_quotes = [ - { - "content": q.content, - "sender": q.sender, - "reason": q.reason, - "user_id": q.user_id, + # 8f. 获取参与者 ID 和最后消息时间戳 + participant_ids = list({msg.sender_id for msg in unified_messages}) + last_message_timestamp = max( + (msg.timestamp for msg in unified_messages), default=0 + ) + + # 8g. 计算本批次总字符数 + characters_count = sum(msg.get_text_length() for msg in unified_messages) + + # 构建批次对象 + batch = IncrementalBatch( + group_id=group_id, + timestamp=time_mod.time(), + messages_count=len(unified_messages), + characters_count=characters_count, + 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, + topics=new_topics, + golden_quotes=new_quotes, + token_usage=token_usage_dict, + last_message_timestamp=last_message_timestamp, + participant_ids=participant_ids, + ) + + # 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)}" + ) + + return { + "success": True, + "batch_summary": batch.get_summary(), + "messages_count": len(unified_messages), } - for q in golden_quotes - ] - - # 8c. 转换 token 消耗: TokenUsage -> dict - token_usage_dict = { - "prompt_tokens": token_usage.prompt_tokens, - "completion_tokens": token_usage.completion_tokens, - "total_tokens": token_usage.total_tokens, - } - - # 8d. 转换用户统计: AnalysisDomainService 格式 -> IncrementalBatch 格式 - user_stats = self._convert_user_activity_for_merge( - user_activity, unified_messages - ) - - # 8e. 转换表情统计: EmojiStatistics -> dict - emoji_stats = { - "face_count": statistics.emoji_statistics.face_count, - "mface_count": statistics.emoji_statistics.mface_count, - "bface_count": statistics.emoji_statistics.bface_count, - "sface_count": statistics.emoji_statistics.sface_count, - "other_emoji_count": statistics.emoji_statistics.other_emoji_count, - "face_details": statistics.emoji_statistics.face_details, - } - - # 8f. 获取参与者 ID 和最后消息时间戳 - participant_ids = list({msg.sender_id for msg in unified_messages}) - last_message_timestamp = max( - (msg.timestamp for msg in unified_messages), default=0 - ) - - # 8g. 计算本批次总字符数 - characters_count = sum(msg.get_text_length() for msg in unified_messages) - - # 构建批次对象 - batch = IncrementalBatch( - group_id=group_id, - timestamp=time_mod.time(), - messages_count=len(unified_messages), - characters_count=characters_count, - 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, - topics=new_topics, - golden_quotes=new_quotes, - token_usage=token_usage_dict, - last_message_timestamp=last_message_timestamp, - participant_ids=participant_ids, - ) - - # 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)}" - ) - - return { - "success": True, - "batch_summary": batch.get_summary(), - "messages_count": len(unified_messages), - } async def execute_incremental_final_report( self, group_id: str, platform_id: str | None = None @@ -422,101 +470,110 @@ class AnalysisApplicationService: Returns: dict: 包含 success、analysis_result、adapter 等信息 """ - if not self.incremental_store or not self.incremental_merge_service: - raise RuntimeError( - "增量分析未初始化：缺少 IncrementalStore 或 IncrementalMergeService" + async with self.group_lock(group_id, "final"): + if not self.incremental_store or not self.incremental_merge_service: + raise RuntimeError( + "增量分析未初始化：缺少 IncrementalStore 或 IncrementalMergeService" + ) + + logger.info( + f"开始增量最终报告: 群 {group_id}, 平台 {platform_id or '默认'}" ) - logger.info(f"开始增量最终报告: 群 {group_id}, 平台 {platform_id or '默认'}") + # 1. 计算滑动窗口范围 + analysis_days = self.config_manager.get_analysis_days() + window_end = time_mod.time() + window_start = window_end - (analysis_days * 24 * 3600) - # 1. 计算滑动窗口范围 - analysis_days = self.config_manager.get_analysis_days() - window_end = time_mod.time() - window_start = window_end - (analysis_days * 24 * 3600) - - # 2. 查询窗口内的所有批次 - batches = await self.incremental_store.query_batches( - group_id, window_start, window_end - ) - - # 3. 检查批次有效性 - if not batches: - logger.warning(f"群 {group_id} 滑动窗口内无增量分析数据，无法生成最终报告") - return {"success": False, "reason": "no_incremental_data"} - - # 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} 的适配器") - - # 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 用户 - top_users = state.get_user_activity_ranking(max_user_titles) - - unified_msg_origin = ( - f"{platform_id}:GroupMessage:{group_id}" if platform_id else group_id + # 2. 查询窗口内的所有批次 + batches = await self.incremental_store.query_batches( + group_id, window_start, window_end ) - try: - ( - user_titles_result, - title_token_usage, - ) = await self.llm_analyzer.analyze_user_titles( - messages=[], # 增量模式下不传原始消息 - user_activity=state.user_activities, - umo=unified_msg_origin, - top_users=top_users, + # 3. 检查批次有效性 + if not batches: + logger.warning( + f"群 {group_id} 滑动窗口内无增量分析数据，无法生成最终报告" ) - user_titles = user_titles_result + return {"success": False, "reason": "no_incremental_data"} - # 将称号分析的 token 消耗追加到状态中 - state.total_token_usage["prompt_tokens"] = ( - state.total_token_usage.get("prompt_tokens", 0) - + title_token_usage.prompt_tokens + # 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} 的适配器") + + # 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 用户 + top_users = state.get_user_activity_ranking(max_user_titles) + + unified_msg_origin = ( + f"{platform_id}:GroupMessage:{group_id}" + if platform_id + else group_id ) - state.total_token_usage["completion_tokens"] = ( - state.total_token_usage.get("completion_tokens", 0) - + title_token_usage.completion_tokens - ) - state.total_token_usage["total_tokens"] = ( - state.total_token_usage.get("total_tokens", 0) - + title_token_usage.total_tokens - ) - except Exception as e: - logger.error(f"增量最终报告用户称号分析失败: {e}", exc_info=True) - # 7. 构建 analysis_result - analysis_result = self.incremental_merge_service.build_analysis_result( - state, user_titles - ) + try: + async with self.llm_semaphore: + logger.debug(f"[LLM] 已进入称号分析队列 (群: {group_id})") + ( + user_titles_result, + title_token_usage, + ) = await self.llm_analyzer.analyze_user_titles( + messages=[], # 增量模式下不传原始消息 + user_activity=state.user_activities, + umo=unified_msg_origin, + top_users=top_users, + ) + user_titles = user_titles_result - # 8. 持久化到 history_manager - await self.history_manager.save_analysis(group_id, analysis_result) + # 将称号分析的 token 消耗追加到状态中 + state.total_token_usage["prompt_tokens"] = ( + state.total_token_usage.get("prompt_tokens", 0) + + title_token_usage.prompt_tokens + ) + state.total_token_usage["completion_tokens"] = ( + state.total_token_usage.get("completion_tokens", 0) + + title_token_usage.completion_tokens + ) + state.total_token_usage["total_tokens"] = ( + state.total_token_usage.get("total_tokens", 0) + + title_token_usage.total_tokens + ) + except Exception as e: + logger.error(f"增量最终报告用户称号分析失败: {e}", exc_info=True) - 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}" - ) + # 7. 构建 analysis_result + analysis_result = self.incremental_merge_service.build_analysis_result( + state, user_titles + ) - return { - "success": True, - "analysis_result": analysis_result, - "messages_count": state.total_message_count, - "adapter": adapter, - } + # 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}" + ) + + return { + "success": True, + "analysis_result": analysis_result, + "messages_count": state.total_message_count, + "adapter": adapter, + } # ---------------------------------------------------------------- # 辅助方法 diff --git a/src/infrastructure/platform/adapters/onebot_adapter.py b/src/infrastructure/platform/adapters/onebot_adapter.py index 5413d84..2eca828 100644 --- a/src/infrastructure/platform/adapters/onebot_adapter.py +++ b/src/infrastructure/platform/adapters/onebot_adapter.py @@ -533,21 +533,28 @@ class OneBotAdapter(PlatformAdapter): logger.error(f"OneBot 图片发送最终失败: {e}") return False - async def was_image_sent_recently(self, group_id: str, seconds: int = 60) -> bool: + async def was_image_sent_recently( + self, group_id: str, seconds: int = 60, token: str | None = None + ) -> bool: """ [真相检查] 检查最近 X 秒内，机器人是否已经向该群发送过图片。 用于判断之前的“超时/1200”错误是否其实已经在后台发送成功。 """ try: # 1. 获取最近的消息历史 (OneBot 标准 API) - history = await self.bot.call_action( - "get_group_msg_history", - group_id=int(group_id), - count=100, # [针对重复检查优化] 提高扫描深度，覆盖大群高频刷屏的情况 - ) + try: + history = await self.bot.call_action( + "get_group_msg_history", + group_id=int(group_id), + count=50, # 适度缩减扫描深度以提高成功率 + ) + except Exception as e: + logger.warning( + f"[OneBot] was_image_sent_recently: get_group_msg_history 失败 (可能 API 繁忙): {e}" + ) + return False # API 失败时，我们保持谨慎，但不阻止重试 if not history or "messages" not in history: - # 某些 OneBot 实现返回值结构不同 messages = history if isinstance(history, list) else [] else: messages = history["messages"] @@ -559,16 +566,16 @@ class OneBotAdapter(PlatformAdapter): # 1. 优先从内存缓存中获取机器人 ID self_id = self.bot_self_ids[0] if self.bot_self_ids else "" - # 2. 如果列表为空，尝试反射实例属性 if not self_id: + # 尝试从 bot 实例中获取多个可能的 ID 属性 self_id = ( str(getattr(self.bot, "self_id", "")) or str(getattr(self.bot, "uin", "")) or str(getattr(self.bot, "user_id", "")) ) - # 3. [兜底方案] 仍未获取到，通过 API 向 OneBot 服务端请求 if not self_id: + # 最后的 API 兜底：尝试从 login_info 获取 try: login_info = await self.bot.call_action("get_login_info") if login_info and "user_id" in login_info: @@ -587,6 +594,15 @@ class OneBotAdapter(PlatformAdapter): "[OneBot] was_image_sent_recently: 无法确定机器人 ID，历史回显校验可能不准确" ) + # [优化] 如果提供了 token，我们也尝试从 caption 中提取 ID 部分进行更精准匹配 + search_token = None + if token and "[ID: " in token: + import re + + match = re.search(r"\[ID: ([^\]]+)\]", token) + if match: + search_token = match.group(0) # 例如 "[ID: report_XXXX]" + for msg in reversed(messages): msg_time = msg.get("time", 0) # 只检查约定时间范围内的消息 @@ -604,15 +620,27 @@ class OneBotAdapter(PlatformAdapter): raw_message = msg.get("message", []) # 适配字符串形式或列表形式的消息 msg_str = str(raw_message) - if "[CQ:image" in msg_str or '"type": "image"' in msg_str: - logger.debug( - f"自检发现群 {group_id} 已有成功发送的图片回显，无需重试。" - ) - return True + + has_image = "[CQ:image" in msg_str or '"type": "image"' in msg_str + + if has_image: + if search_token: + # 精确匹配 TraceID + if search_token in msg_str: + logger.debug( + f"自检发现群 {group_id} 已有匹配 ID ({search_token}) 的图片回显，拦截重复发送。" + ) + return True + else: + # 广义匹配（回退模式） + logger.debug( + f"自检发现群 {group_id} 已有成功发送的图片回显，无需重试。" + ) + return True return False except Exception as e: - logger.debug(f"回显自检失败 (可能不支持 get_group_msg_history): {e}") + logger.debug(f"回显自检失败: {e}") return False async def send_file( diff --git a/src/infrastructure/reporting/dispatcher.py b/src/infrastructure/reporting/dispatcher.py index c4db150..a1142c3 100644 --- a/src/infrastructure/reporting/dispatcher.py +++ b/src/infrastructure/reporting/dispatcher.py @@ -94,8 +94,9 @@ class ReportDispatcher: # 3. 发送图片 if image_url: + caption = f"📊 每日群聊分析报告已生成：\n[ID: {trace_id}]" sent = await self.message_sender.send_image_smart( - group_id, image_url, "📊 每日群聊分析报告已生成：", platform_id + group_id, image_url, caption, platform_id ) if sent: # 4. 发送成功后，尝试上传到群文件/群相册（静默处理） @@ -119,7 +120,7 @@ class ReportDispatcher: analysis_result, group_id, platform_id, - caption="📊 每日群聊分析报告已生成：", + caption=f"📊 每日群聊分析报告已生成：\n[ID: {trace_id}]", ) return True # 已加入队列视作处理成功 (不在此处报错) else: diff --git a/src/infrastructure/reporting/generators.py b/src/infrastructure/reporting/generators.py index c906a4b..1ada24f 100644 --- a/src/infrastructure/reporting/generators.py +++ b/src/infrastructure/reporting/generators.py @@ -25,6 +25,9 @@ class ReportGenerator(IReportGenerator): self.config_manager = config_manager self.activity_visualizer = ActivityVisualizer() self.html_templates = HTMLTemplates(config_manager) # 实例化HTML模板管理器 + # 全局 T2I 渲染信号量，保护本地资源 + max_concurrent = self.config_manager.get_max_concurrent_tasks() + self._render_semaphore = asyncio.Semaphore(max_concurrent) async def generate_image_report( self, @@ -67,106 +70,112 @@ class ReportGenerator(IReportGenerator): logger.info(f"图片报告HTML渲染完成，长度: {len(html_content)} 字符") - # 定义渲染策略 - render_strategies = [ - # 1. 第一策略: PNG, Ultra quality, Device scale - { - "full_page": True, - "type": "png", - "scale": "device", - "device_scale_factor_level": "ultra", - }, - # 2. 第二策略: JPEG, ultra, quality 100%, Device scale - { - "full_page": True, - "type": "jpeg", - "quality": 100, - "scale": "device", - "device_scale_factor_level": "ultra", - }, - # 3. 第三策略: JPEG, high, quality 80%, Device scale - { - "full_page": True, - "type": "jpeg", - "quality": 95, - "scale": "device", - "device_scale_factor_level": "high", # 尝试高分辨率 - }, - # 4. 第四策略: JPEG, normal quality, Device scale (后备) - { - "full_page": True, - "type": "jpeg", - "quality": 80, - "scale": "device", - # normal quality - }, - ] + # 使用信号量控制并发进入渲染引擎 + async with self._render_semaphore: + logger.debug(f"[T2I] 已进入渲染队列 (群: {group_id})") - last_exception = None + # 定义渲染策略 + render_strategies = [ + # 1. 第一策略: PNG, Ultra quality, Device scale + { + "full_page": True, + "type": "png", + "scale": "device", + "device_scale_factor_level": "ultra", + }, + # 2. 第二策略: JPEG, ultra, quality 100%, Device scale + { + "full_page": True, + "type": "jpeg", + "quality": 100, + "scale": "device", + "device_scale_factor_level": "ultra", + }, + # 3. 第三策略: JPEG, high, quality 80%, Device scale + { + "full_page": True, + "type": "jpeg", + "quality": 95, + "scale": "device", + "device_scale_factor_level": "high", # 尝试高分辨率 + }, + # 4. 第四策略: JPEG, normal quality, Device scale (后备) + { + "full_page": True, + "type": "jpeg", + "quality": 80, + "scale": "device", + # normal quality + }, + ] - for image_options in render_strategies: - try: - # Cleanse options - if image_options.get("type") == "png": - image_options["quality"] = None + last_exception = None - logger.info(f"正在尝试渲染策略: {image_options}") - # 改为获取 bytes 数据，避免 OneBot 无法访问内部 URL - image_data = await html_render_func( - html_content, # 渲染后的HTML内容 - {}, # 空数据字典，因为数据已包含在HTML中 - False, # return_url=False，直接获取图片数据 - image_options, - ) + for image_options in render_strategies: + try: + # Cleanse options + if image_options.get("type") == "png": + image_options["quality"] = None - if image_data: - # 校验是否为合法图片（防止 T2I 返回 500 错误 HTML 字符流） - is_valid = False - actual_data_head = None + logger.info(f"正在尝试渲染策略: {image_options}") + # 改为获取 bytes 数据，避免 OneBot 无法访问内部 URL + image_data = await html_render_func( + html_content, # 渲染后的HTML内容 + {}, # 空数据字典，因为数据已包含在HTML中 + False, # return_url=False，直接获取图片数据 + image_options, + ) - if isinstance(image_data, bytes): - actual_data_head = image_data[:10] - elif isinstance(image_data, str) and os.path.exists(image_data): - try: - with open(image_data, "rb") as f: - actual_data_head = f.read(10) - except Exception as e: - logger.warning(f"读取图片临时文件失败: {e}") + if image_data: + # 校验是否为合法图片（防止 T2I 返回 500 错误 HTML 字符流） + is_valid = False + actual_data_head = None - if actual_data_head: - # 检查 magic numbers (JPEG: FF D8, PNG: 89 50 4E 47) - if actual_data_head.startswith( - b"\xff\xd8" - ) or actual_data_head.startswith(b"\x89PNG"): - is_valid = True - else: - logger.warning( - f"渲染结果似乎不是有效的图片数据 (头部: {actual_data_head.hex()})" - ) - - if is_valid: if isinstance(image_data, bytes): - b64 = base64.b64encode(image_data).decode("utf-8") - image_url = f"base64://{b64}" - logger.info( - f"图片生成成功 ({image_options}): [Base64 Data {len(image_data)} bytes]" - ) - return image_url, html_content - elif isinstance(image_data, str): - logger.info(f"图片生成成功 (String): {image_data}") - return image_data, html_content + actual_data_head = image_data[:10] + elif isinstance(image_data, str) and os.path.exists( + image_data + ): + try: + with open(image_data, "rb") as f: + actual_data_head = f.read(10) + except Exception as e: + logger.warning(f"读取图片临时文件失败: {e}") - logger.warning(f"渲染策略 {image_options} 返回了无效或空数据") + if actual_data_head: + # 检查 magic numbers (JPEG: FF D8, PNG: 89 50 4E 47) + if actual_data_head.startswith( + b"\xff\xd8" + ) or actual_data_head.startswith(b"\x89PNG"): + is_valid = True + else: + logger.warning( + f"渲染结果似乎不是有效的图片数据 (头部: {actual_data_head.hex()})" + ) - except Exception as e: - logger.warning(f"渲染策略 {image_options} 失败: {e}") - last_exception = e - logger.warning("尝试下一个策略") - continue + if is_valid: + if isinstance(image_data, bytes): + b64 = base64.b64encode(image_data).decode("utf-8") + image_url = f"base64://{b64}" + logger.info( + f"图片生成成功 ({image_options}): [Base64 Data {len(image_data)} bytes]" + ) + return image_url, html_content + elif isinstance(image_data, str): + logger.info(f"图片生成成功 (String): {image_data}") + return image_data, html_content - # 如果所有策略都失败 - logger.error(f"所有渲染策略都失败。最后一个错误: {last_exception}") - return None, html_content + logger.warning(f"渲染策略 {image_options} 返回了无效或空数据") + + except Exception as e: + logger.warning(f"渲染策略 {image_options} 失败: {e}") + last_exception = e + logger.warning("尝试下一个策略") + continue + + # 如果所有策略都失败 + logger.error(f"所有渲染策略都失败。最后一个错误: {last_exception}") + return None, html_content except Exception as e: logger.error(f"生成图片报告过程发生严重错误: {e}", exc_info=True) diff --git a/src/infrastructure/scheduler/auto_scheduler.py b/src/infrastructure/scheduler/auto_scheduler.py index 90f7115..6646547 100644 --- a/src/infrastructure/scheduler/auto_scheduler.py +++ b/src/infrastructure/scheduler/auto_scheduler.py @@ -5,7 +5,6 @@ import asyncio import time as time_mod -import weakref from typing import Any from apscheduler.triggers.cron import CronTrigger @@ -319,21 +318,10 @@ class AutoScheduler: logger.info(f"将为 {len(target_list)} 个群聊并发执行分析") - # 创建并发任务，限制最大并发数 - max_concurrent = self.config_manager.get_max_concurrent_tasks() - logger.info(f"自动分析并发数限制: {max_concurrent}") - sem = asyncio.Semaphore(max_concurrent) - - async def safe_perform_analysis(gid, pid): - async with sem: - return await self._perform_auto_analysis_for_group_with_timeout( - gid, pid - ) - analysis_tasks = [] for gid, pid in target_list: task = asyncio.create_task( - safe_perform_analysis(gid, pid), + self._perform_auto_analysis_for_group_with_timeout(gid, pid), name=f"analysis_group_{gid}", ) analysis_tasks.append(task) @@ -343,18 +331,22 @@ class AutoScheduler: # 统计执行结果 success_count = 0 + skip_count = 0 error_count = 0 for i, result in enumerate(results): gid, _ = target_list[i] - if isinstance(result, Exception): + if isinstance(result, asyncio.CancelledError): + # 锁冲突导致的跳过 + skip_count += 1 + elif isinstance(result, Exception): logger.error(f"群 {gid} 分析任务异常: {result}") error_count += 1 else: success_count += 1 logger.info( - f"并发分析完成 - 成功: {success_count}, 失败: {error_count}, 总计: {len(target_list)}" + f"并发分析完成 - 成功: {success_count}, 跳过: {skip_count}, 失败: {error_count}, 总计: {len(target_list)}" ) except Exception as e: @@ -379,60 +371,54 @@ class AutoScheduler: self, group_id: str, target_platform_id: str | None = None ): """为指定群执行自动分析（业务逻辑委派给 AnalysisApplicationService）""" - # 为每个群聊使用独立的锁 - group_lock_key = f"analysis_{group_id}" - if not hasattr(self, "_group_locks"): - self._group_locks = weakref.WeakValueDictionary() + try: + # 设置 TraceID + trace_id = TraceContext.generate(prefix=f"group_{group_id}") + TraceContext.set(trace_id) - lock = self._group_locks.get(group_lock_key) - if lock is None: - lock = asyncio.Lock() - self._group_locks[group_lock_key] = lock + logger.info( + f"开始为群 {group_id} 执行自动分析 (Platform: {target_platform_id or 'Auto'})" + ) - async with lock: - try: - # 设置 TraceID - trace_id = TraceContext.generate(prefix=f"group_{group_id}") - TraceContext.set(trace_id) + # 检查平台状态 (BotManager 为基础设施层，用于获取平台就绪状态) + if not self.bot_manager.is_ready_for_auto_analysis(): + logger.warning(f"群 {group_id} 自动分析跳过：bot管理器未就绪") + return - logger.info( - f"开始为群 {group_id} 执行自动分析 (Platform: {target_platform_id or 'Auto'})" - ) + # 委派给应用层服务执行核心用例 + # AnalysisApplicationService 内部已处理群锁 (group_lock) + result = await self.analysis_service.execute_daily_analysis( + group_id=group_id, platform_id=target_platform_id, manual=False + ) - # 检查平台状态 (BotManager 为基础设施层，用于获取平台就绪状态) - if not self.bot_manager.is_ready_for_auto_analysis(): - logger.warning(f"群 {group_id} 自动分析跳过：bot管理器未就绪") - return + if not result.get("success"): + reason = result.get("reason") + logger.info(f"群 {group_id} 自动分析跳过: {reason}") + return - # 委派给应用层服务执行核心用例 - result = await self.analysis_service.execute_daily_analysis( - group_id=group_id, platform_id=target_platform_id, manual=False - ) + # 获取分析结果及适配器 + analysis_result = result["analysis_result"] + adapter = result["adapter"] - if not result.get("success"): - reason = result.get("reason") - logger.info(f"群 {group_id} 自动分析跳过: {reason}") - return + # 调度导出并发送报告 + await self.report_dispatcher.dispatch( + group_id, + analysis_result, + adapter.platform_id + if hasattr(adapter, "platform_id") + else target_platform_id, + ) - # 获取分析结果及适配器 - analysis_result = result["analysis_result"] - adapter = result["adapter"] + logger.info(f"群 {group_id} 自动分析任务执行成功") - # 调度导出并发送报告（由 ReportDispatcher 协调） - await self.report_dispatcher.dispatch( - group_id, - analysis_result, - adapter.platform_id - if hasattr(adapter, "platform_id") - else target_platform_id, - ) - - logger.info(f"群 {group_id} 自动分析任务执行成功") - - except Exception as e: - logger.error(f"群 {group_id} 自动分析执行失败: {e}", exc_info=True) - finally: - logger.debug(f"群 {group_id} 自动分析流程结束") + except asyncio.CancelledError: + # group_lock 抛出的 CancelledError 表示任务正在运行，优雅跳过 + logger.debug(f"群 {group_id} 任务因并发锁冲突而跳过（已在运行）") + raise # 重新抛出，让上层知道任务并没真正执行而是跳过了 + except Exception as e: + logger.error(f"群 {group_id} 自动分析执行失败: {e}", exc_info=True) + finally: + logger.debug(f"群 {group_id} 自动分析流程结束") # ================================================================ # 增量模式：增量分析 @@ -458,32 +444,31 @@ class AutoScheduler: f"(并发限制: {max_concurrent}, 交错间隔: {stagger}秒)" ) - sem = asyncio.Semaphore(max_concurrent) + # 资源限制现在由 Application Service 全局控制，此处仅保留交错逻辑 async def staggered_incremental(idx, gid, pid): - async with sem: - # 按索引交错延迟，均匀分散 API 压力 - if idx > 0 and stagger > 0: - await asyncio.sleep(stagger * idx) + # 按索引交错延迟，均匀分散 API 压力 + if idx > 0 and stagger > 0: + await asyncio.sleep(stagger * idx) - result = ( - await self._perform_incremental_analysis_for_group_with_timeout( + result = ( + await self._perform_incremental_analysis_for_group_with_timeout( + gid, pid + ) + ) + + # 检查是否需要立即发送报告（调试模式） + if self.config_manager.get_incremental_report_immediately(): + if isinstance(result, dict) and result.get("success"): + logger.info( + f"增量分析立即报告模式生效，正在为群 {gid} 生成报告..." + ) + # 立即生成最终报告 + await self._perform_incremental_final_report_for_group_with_timeout( gid, pid ) - ) - # 检查是否需要立即发送报告（调试模式） - if self.config_manager.get_incremental_report_immediately(): - if isinstance(result, dict) and result.get("success"): - logger.info( - f"增量分析立即报告模式生效，正在为群 {gid} 生成报告..." - ) - # 立即生成最终报告 - await self._perform_incremental_final_report_for_group_with_timeout( - gid, pid - ) - - return result + return result analysis_tasks = [] for idx, (gid, pid) in enumerate(target_list): @@ -542,57 +527,51 @@ class AutoScheduler: self, group_id: str, target_platform_id: str | None = None ): """为指定群执行增量分析（业务逻辑委派给 AnalysisApplicationService）""" - # 为每个群聊使用独立的锁 - group_lock_key = f"incremental_{group_id}" - if not hasattr(self, "_group_locks"): - self._group_locks = weakref.WeakValueDictionary() + try: + # 设置 TraceID + trace_id = TraceContext.generate(prefix=f"incr_{group_id}") + TraceContext.set(trace_id) - lock = self._group_locks.get(group_lock_key) - if lock is None: - lock = asyncio.Lock() - self._group_locks[group_lock_key] = lock + logger.info( + f"开始为群 {group_id} 执行增量分析 " + f"(Platform: {target_platform_id or 'Auto'})" + ) - async with lock: - try: - # 设置 TraceID - trace_id = TraceContext.generate(prefix=f"incr_{group_id}") - TraceContext.set(trace_id) + # 检查平台状态 + if not self.bot_manager.is_ready_for_auto_analysis(): + logger.warning(f"群 {group_id} 增量分析跳过：bot管理器未就绪") + return {"success": False, "reason": "bot_not_ready"} - logger.info( - f"开始为群 {group_id} 执行增量分析 " - f"(Platform: {target_platform_id or 'Auto'})" - ) + # 委派给应用层服务执行增量分析用例 + # AnalysisApplicationService 内部已处理群锁 (group_lock) + result = await self.analysis_service.execute_incremental_analysis( + group_id=group_id, platform_id=target_platform_id + ) - # 检查平台状态 - if not self.bot_manager.is_ready_for_auto_analysis(): - logger.warning(f"群 {group_id} 增量分析跳过：bot管理器未就绪") - return {"success": False, "reason": "bot_not_ready"} - - # 委派给应用层服务执行增量分析用例 - result = await self.analysis_service.execute_incremental_analysis( - group_id=group_id, platform_id=target_platform_id - ) - - if not result.get("success"): - reason = result.get("reason", "unknown") - logger.info(f"群 {group_id} 增量分析跳过: {reason}") - return result - - # 增量分析只累积数据，不发送报告 - batch_summary = result.get("batch_summary", {}) - logger.info( - f"群 {group_id} 增量分析完成: " - f"消息数={result.get('messages_count', 0)}, " - f"话题={batch_summary.get('topics_count', 0)}, " - f"金句={batch_summary.get('quotes_count', 0)}" - ) + if not result.get("success"): + reason = result.get("reason", "unknown") + logger.info(f"群 {group_id} 增量分析跳过: {reason}") return result - except Exception as e: - logger.error(f"群 {group_id} 增量分析执行失败: {e}", exc_info=True) - return {"success": False, "reason": str(e)} - finally: - logger.debug(f"群 {group_id} 增量分析流程结束") + # 增量分析只累积数据，不发送报告 + batch_summary = result.get("batch_summary", {}) + logger.info( + f"群 {group_id} 增量分析完成: " + f"消息数={result.get('messages_count', 0)}, " + f"话题={batch_summary.get('topics_count', 0)}, " + f"金句={batch_summary.get('quotes_count', 0)}" + ) + return result + + except asyncio.CancelledError: + # group_lock 抛出的 CancelledError 表示任务正在运行，优雅跳过 + logger.debug(f"群 {group_id} 增量分析因并发锁冲突而跳过（已在运行）") + return {"success": False, "reason": "already_running"} + except Exception as e: + logger.error(f"群 {group_id} 增量分析执行失败: {e}", exc_info=True) + return {"success": False, "reason": str(e)} + finally: + logger.debug(f"群 {group_id} 增量分析流程结束") # ================================================================ # 增量模式：最终报告生成 @@ -618,15 +597,14 @@ class AutoScheduler: f"(并发限制: {max_concurrent}, 交错间隔: {stagger}秒)" ) - sem = asyncio.Semaphore(max_concurrent) - async def staggered_final_report(idx, gid, pid): - async with sem: - if idx > 0 and stagger > 0: - await asyncio.sleep(stagger * idx) - return await self._perform_incremental_final_report_for_group_with_timeout( + if idx > 0 and stagger > 0: + await asyncio.sleep(stagger * idx) + return ( + await self._perform_incremental_final_report_for_group_with_timeout( gid, pid ) + ) report_tasks = [] for idx, (gid, pid) in enumerate(target_list): @@ -685,80 +663,74 @@ class AutoScheduler: self, group_id: str, target_platform_id: str | None = None ): """为指定群生成增量最终报告（业务逻辑委派给 AnalysisApplicationService）""" - # 为每个群聊使用独立的锁 - group_lock_key = f"final_report_{group_id}" - if not hasattr(self, "_group_locks"): - self._group_locks = weakref.WeakValueDictionary() + try: + # 设置 TraceID + trace_id = TraceContext.generate(prefix=f"report_{group_id}") + TraceContext.set(trace_id) - lock = self._group_locks.get(group_lock_key) - if lock is None: - lock = asyncio.Lock() - self._group_locks[group_lock_key] = lock + logger.info( + f"开始为群 {group_id} 生成增量最终报告 " + f"(Platform: {target_platform_id or 'Auto'})" + ) - async with lock: - try: - # 设置 TraceID - trace_id = TraceContext.generate(prefix=f"report_{group_id}") - TraceContext.set(trace_id) + # 检查平台状态 + if not self.bot_manager.is_ready_for_auto_analysis(): + logger.warning(f"群 {group_id} 最终报告跳过：bot管理器未就绪") + return {"success": False, "reason": "bot_not_ready"} - logger.info( - f"开始为群 {group_id} 生成增量最终报告 " - f"(Platform: {target_platform_id or 'Auto'})" - ) + # 委派给应用层服务执行最终报告用例 + # AnalysisApplicationService 内部已处理群锁 (group_lock) + result = await self.analysis_service.execute_incremental_final_report( + group_id=group_id, platform_id=target_platform_id + ) - # 检查平台状态 - if not self.bot_manager.is_ready_for_auto_analysis(): - logger.warning(f"群 {group_id} 最终报告跳过：bot管理器未就绪") - return {"success": False, "reason": "bot_not_ready"} - - # 委派给应用层服务执行最终报告用例 - result = await self.analysis_service.execute_incremental_final_report( - group_id=group_id, platform_id=target_platform_id - ) - - if not result.get("success"): - reason = result.get("reason", "unknown") - logger.info(f"群 {group_id} 最终报告跳过: {reason}") - return result - - # 获取分析结果及适配器，分发报告 - analysis_result = result["analysis_result"] - adapter = result["adapter"] - - await self.report_dispatcher.dispatch( - group_id, - analysis_result, - adapter.platform_id - if hasattr(adapter, "platform_id") - 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} 增量最终报告发送成功") + if not result.get("success"): + reason = result.get("reason", "unknown") + logger.info(f"群 {group_id} 最终报告跳过: {reason}") return result - except Exception as e: - logger.error(f"群 {group_id} 最终报告执行失败: {e}", exc_info=True) - return {"success": False, "reason": str(e)} - finally: - logger.debug(f"群 {group_id} 最终报告流程结束") + # 获取分析结果及适配器，分发报告 + analysis_result = result["analysis_result"] + adapter = result["adapter"] + + await self.report_dispatcher.dispatch( + group_id, + analysis_result, + adapter.platform_id + if hasattr(adapter, "platform_id") + 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 + + except asyncio.CancelledError: + # group_lock 抛出的 CancelledError 表示任务正在运行，优雅跳过 + logger.debug(f"群 {group_id} 最终报告因并发锁冲突而跳过（已在运行）") + return {"success": False, "reason": "already_running"} + except Exception as e: + logger.error(f"群 {group_id} 最终报告执行失败: {e}", exc_info=True) + return {"success": False, "reason": str(e)} + finally: + logger.debug(f"群 {group_id} 最终报告流程结束") # ================================================================ # 群列表获取（基础设施层） diff --git a/src/infrastructure/scheduler/retry.py b/src/infrastructure/scheduler/retry.py index 38e54c8..f78562b 100644 --- a/src/infrastructure/scheduler/retry.py +++ b/src/infrastructure/scheduler/retry.py @@ -157,7 +157,9 @@ class RetryManager: adapter = self.bot_manager.get_adapter(task.platform_id) if adapter and hasattr(adapter, "was_image_sent_recently"): # 检查过去 5 分钟内的消息回显 (覆盖初发和之前的重试) - if await adapter.was_image_sent_recently(task.group_id, seconds=300): + if await adapter.was_image_sent_recently( + task.group_id, seconds=300, token=task.caption + ): logger.info( f"[RetryManager] [拦截] 根据历史回显，群 {task.group_id} 的图片已成功送达。取消本次重试。" ) @@ -287,7 +289,9 @@ class RetryManager: # 3. 【临界检查 2】发送图片前最后一次复核 (针对渲染耗时极长产生的盲窗) # 例如渲染 10s 期间图片出来了，这里可以最后贴身拦截一次 if adapter and hasattr(adapter, "was_image_sent_recently"): - if await adapter.was_image_sent_recently(task.group_id, seconds=120): + if await adapter.was_image_sent_recently( + task.group_id, seconds=120, token=task.caption + ): logger.info( f"[RetryManager] [临界拦截] 渲染完成后检测到群 {task.group_id} 已有报告。拦截重复发送。" )