From e627ac99f32dea467934cc7fa427178a39b33032 Mon Sep 17 00:00:00 2001 From: Codex <242516109+Codex@users.noreply.github.com> Date: Fri, 3 Apr 2026 01:17:31 +0800 Subject: [PATCH 1/3] feat: Add UMO Group aggregation support and routing Co-authored-by: lekoOwO <20151124+lekoOwO@users.noreply.github.com> Co-authored-by: openai-code-agent[bot] <242516109+Codex@users.noreply.github.com> --- README.md | 7 + _conf_schema.json | 33 ++ .../services/analysis_application_service.py | 513 +++++++++++++++--- src/infrastructure/config/config_manager.py | 110 ++++ src/infrastructure/reporting/dispatcher.py | 38 +- .../scheduler/auto_scheduler.py | 405 ++++++++++++-- 6 files changed, 981 insertions(+), 125 deletions(-) diff --git a/README.md b/README.md index 461f2ce4..d6690593 100644 --- a/README.md +++ b/README.md @@ -160,6 +160,13 @@ _✨ 一个基于 AstrBot 的智能群聊分析插件，支持 **QQ (OneBot)** | HTML 格式 (自建) | 配置 `html_base_url` 后，机器人会发送可直接点击的报告外链。 | 输出格式需设为 html | | 自定义 LLM 服务 | 用户可自行选取个人提供的服务商。 | 留空则回退到默认服务商 | +### UMO Group 聚合配置 +- 在面板的 **UMO 分组** 中配置多条分组，每条包含 `id`、`source_umos`（来源群 UMO 列表）和 `output_umos`（报告输出 UMO 列表）。 +- 在基础白/黑名单、定时分析群列表、增量分析群列表里，可用 `umoGroup:` 引用某个分组。来源群既可以继续单独产出报告，也会参与所属分组的聚合报告。 +- 每个分组会聚合其来源 UMO 的消息生成一份报告，并发送到所有 `output_umos`；同一个来源可属于多个分组，互不影响。 +- 组 ID 仅允许字母、数字、下划线、短横线，不得包含空格、点号、分号等特殊字符。 +- 若分组出现在增量名单，会使用增量模式按来源累积批次并合并生成报告，立即上报和定时报告均使用同一套增量数据。 + ### 分析黑白名单配置说明（小白能懂） 下面只讲“在面板里怎么点”。 diff --git a/_conf_schema.json b/_conf_schema.json index 0a47a117..75d1112b 100644 --- a/_conf_schema.json +++ b/_conf_schema.json @@ -98,6 +98,39 @@ } } }, + "umo_groups": { + "description": "UMO 分组", + "type": "list", + "hint": "将多个群的 UMO 组合为一个 UMO Group，用于聚合分析和指定输出 UMO。UMO Group ID 会以 umoGroup:ID 的形式在其他配置中引用。", + "default": [], + "items": { + "type": "object", + "items": { + "id": { + "type": "string", + "description": "UMO Group ID", + "hint": "用于标识此 UMO Group，引用时以 umoGroup:ID 的形式填写。仅允许字母、数字、下划线、短横线，禁止包含空格、点号、分号等特殊字符。" + }, + "source_umos": { + "type": "list", + "description": "来源 UMO 列表", + "hint": "参与聚合分析的来源 UMO，格式为 platform:GroupMessage:session_id。", + "items": { + "type": "string" + } + }, + "output_umos": { + "type": "list", + "description": "输出 UMO 列表", + "hint": "聚合报告要发送到的目标 UMO，支持填写多个 UMO。", + "default": [], + "items": { + "type": "string" + } + } + } + } + }, "auto_analysis": { "description": "定时分析设置", "type": "object", diff --git a/src/application/services/analysis_application_service.py b/src/application/services/analysis_application_service.py index c860a4b4..572c97e8 100644 --- a/src/application/services/analysis_application_service.py +++ b/src/application/services/analysis_application_service.py @@ -127,7 +127,7 @@ async def execute_daily_analysis( ) # 1. 获取适配器 - adapter = self.bot_manager.get_adapter(platform_id) + adapter = self.bot_manager.get_adapter(platform_id) if platform_id else None if not adapter: raise ValueError(f"未找到平台 {platform_id} 的适配器") @@ -199,101 +199,176 @@ async def execute_daily_analysis( ) 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, + return await self._analyze_messages( unified_messages, - bot_self_ids, + group_id, + platform_id=platform_id, + adapter=adapter, ) - 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 - ) + async def execute_daily_analysis_for_sources( + self, group_id: str, source_umos: list[str] + ) -> dict[str, Any]: + """ + 聚合多个来源 UMO 的消息执行一次全量分析。 - # 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() - ) - chat_quality_enabled = ( - self.config_manager.get_chat_quality_analysis_enabled() - ) + Args: + group_id: 虚拟 UMO Group ID（用于上下文与记录） + source_umos: 来源 UMO 列表 + """ + async with self.group_lock(group_id, "daily"): + from ...domain.services.message_cleaner_service import MessageCleanerService - topics = [] - user_titles = [] - golden_quotes = [] - chat_quality_review = None - total_token_usage = TokenUsage() + days = self.config_manager.get_analysis_days() + max_count = self.config_manager.get_max_messages() - # Note: LLMAnalyzer 目前可能只接收 legacy 格式或特定的 UnifiedMessage 适配 - # 暂时转换回 legacy 格式以确保稳定性，直到 LLMAnalyzer 被重构 - legacy_messages = self.statistics_service._convert_to_legacy_dict( - unified_messages + raw_messages = [] + first_platform: str | None = None + + for umo in source_umos: + platform_id, session_id = self.config_manager.parse_umo_string(umo) + if not platform_id or not session_id: + logger.warning(f"[UMOGroup] 无效的 UMO 格式，跳过: {umo}") + continue + + adapter = self.bot_manager.get_adapter(platform_id) + if not adapter: + logger.warning( + f"[UMOGroup] 未找到平台 {platform_id} 的适配器，跳过 {umo}" + ) + continue + + try: + msgs = await adapter.fetch_messages( + group_id=session_id, days=days, max_count=max_count + ) + if msgs: + raw_messages.extend(msgs) + if not first_platform: + first_platform = platform_id + except Exception as e: + logger.warning(f"[UMOGroup] 拉取 {umo} 消息失败: {e}") + + if not raw_messages: + logger.warning(f"[UMOGroup] 组 {group_id} 无可用消息，跳过分析") + return {"success": False, "reason": "no_messages"} + + 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 ) - unified_msg_origin = ( - f"{platform_id}:GroupMessage:{group_id}" if platform_id else group_id + threshold = self.config_manager.get_min_messages_threshold() + if len(unified_messages) < threshold: + logger.info( + f"[UMOGroup] 组 {group_id} 有效消息数 " + f"({len(unified_messages)}) 低于阈值 ({threshold})，跳过分析" + ) + return {"success": False, "reason": "below_threshold"} + + return await self._analyze_messages( + unified_messages, + group_id, + platform_id=first_platform, + adapter=None, + umo_override=group_id, ) - if ( - topic_enabled - or user_title_enabled - or golden_quote_enabled - or chat_quality_enabled - ): - async with self.llm_semaphore: - logger.debug(f"[LLM] 已进入分析队列 (群: {group_id})") - ( - topics, - user_titles, - golden_quotes, - total_token_usage, - chat_quality_review, - ) = 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, - chat_quality_enabled=chat_quality_enabled, - ) + async def _analyze_messages( + self, + unified_messages: list[UnifiedMessage], + group_id: str, + platform_id: str | None = None, + adapter=None, + umo_override: str | None = None, + ) -> dict[str, Any]: + """复用的分析流水线，输入已清洗的消息。""" + # 1. 基础统计 + statistics = await asyncio.to_thread( + self.statistics_service.calculate_group_statistics, unified_messages + ) + + # 2. 用户分析 + 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 + ) + + 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() + chat_quality_enabled = self.config_manager.get_chat_quality_analysis_enabled() + + topics = [] + user_titles = [] + golden_quotes = [] + chat_quality_review = None + total_token_usage = TokenUsage() + + legacy_messages = self.statistics_service._convert_to_legacy_dict( + unified_messages + ) + + unified_msg_origin = ( + umo_override + if umo_override + else (f"{platform_id}:GroupMessage:{group_id}" if platform_id else group_id) + ) + + if ( + topic_enabled + or user_title_enabled + or golden_quote_enabled + or chat_quality_enabled + ): + async with self.llm_semaphore: + logger.debug(f"[LLM] 已进入分析队列 (群/UMO: {group_id})") + ( + topics, + user_titles, + golden_quotes, + total_token_usage, + chat_quality_review, + ) = 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, + chat_quality_enabled=chat_quality_enabled, + ) - # 回填结果 - statistics.golden_quotes = golden_quotes - statistics.token_usage = total_token_usage + 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, - "chat_quality_review": chat_quality_review, - } + analysis_result = { + "statistics": statistics, + "topics": topics, + "user_titles": user_titles, + "user_analysis": user_activity, + "chat_quality_review": chat_quality_review, + } - # 6. 持久化摘要 (Persistence) - await self.history_manager.save_analysis(group_id, analysis_result) + 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, - "group_id": group_id, - "platform_id": getattr(adapter, "platform_id", platform_id), - } + return { + "success": True, + "analysis_result": analysis_result, + "messages_count": len(unified_messages), + "adapter": adapter, + "group_id": group_id, + "platform_id": platform_id, + } # ---------------------------------------------------------------- # 增量分析用例 @@ -561,8 +636,276 @@ async def execute_incremental_analysis( "platform_id": getattr(adapter, "platform_id", platform_id), } + async def execute_incremental_analysis_for_sources( + self, group_id: str, source_umos: list[str] + ) -> dict[str, Any]: + """ + 聚合多个来源的增量消息，生成单个批次并存储到组级增量存储。 + + 与单群增量分析类似，但会为每个来源维护独立的水位线， + 批次数据统一归档到 group_id 名下，供后续最终报告合并。 + """ + async with self.group_lock(group_id, "incremental"): + if not self.incremental_store: + raise RuntimeError("增量分析未初始化：缺少 IncrementalStore") + + from ...domain.services.message_cleaner_service import MessageCleanerService + + logger.info( + f"[UMOGroup] 开始增量分析: 组 {group_id}, 来源={len(source_umos)}" + ) + + cleaner = MessageCleanerService() + bot_self_ids = self.config_manager.get_bot_self_ids() + + days = self.config_manager.get_analysis_days() + max_count = self.config_manager.get_incremental_safe_limit() + min_messages = self.config_manager.get_incremental_min_messages() + + combined_messages: list[UnifiedMessage] = [] + source_watermarks: dict[str, int] = {} + platform_hint: str | None = None + + for umo in source_umos: + platform_id, session_id = self.config_manager.parse_umo_string(umo) + if not platform_id or not session_id: + logger.warning(f"[UMOGroup] 无效的来源 UMO，跳过: {umo}") + continue + + adapter = self.bot_manager.get_adapter(platform_id) + if not adapter: + logger.warning( + f"[UMOGroup] 未找到平台 {platform_id} 的适配器，跳过 {umo}" + ) + continue + + progress_key = self._build_source_progress_key(group_id, umo) + last_ts = await self.incremental_store.get_last_analyzed_timestamp( + progress_key + ) + + try: + raw_messages = await adapter.fetch_messages( + group_id=session_id, + days=days, + max_count=max_count, + since_ts=last_ts, + ) + except Exception as e: + logger.warning(f"[UMOGroup] 拉取 {umo} 增量消息失败: {e}") + continue + + if not raw_messages: + continue + + unified_messages = cleaner.clean_messages( + raw_messages, bot_self_ids=bot_self_ids, filter_commands=True + ) + + if last_ts > 0: + unified_messages = [ + msg for msg in unified_messages if msg.timestamp > last_ts + ] + + if not unified_messages: + continue + + combined_messages.extend(unified_messages) + source_watermarks[progress_key] = max( + msg.timestamp for msg in unified_messages + ) + if not platform_hint: + platform_hint = platform_id + + if not combined_messages: + logger.warning(f"[UMOGroup] 组 {group_id} 增量分析无新消息，跳过") + return {"success": False, "reason": "no_messages"} + + combined_messages.sort(key=lambda m: m.timestamp) + + if len(combined_messages) < min_messages: + logger.info( + f"[UMOGroup] 组 {group_id} 增量消息数 " + f"({len(combined_messages)}) 未达到阈值 ({min_messages})，跳过本次分析" + ) + return {"success": False, "reason": "below_threshold"} + + statistics = await asyncio.to_thread( + self.statistics_service.calculate_group_statistics, combined_messages + ) + user_activity = await asyncio.to_thread( + self.analysis_domain_service.analyze_user_activity, + combined_messages, + bot_self_ids, + ) + + hourly_msg_counts, hourly_char_counts = self._compute_hourly_counts( + combined_messages + ) + + 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() + ) + chat_quality_enabled = ( + self.config_manager.get_chat_quality_analysis_enabled() + ) + + legacy_messages = self.statistics_service._convert_to_legacy_dict( + combined_messages + ) + unified_msg_origin = ( + f"{platform_hint}:GroupMessage:{group_id}" + if platform_hint + else group_id + ) + + topics = [] + golden_quotes = [] + token_usage = TokenUsage() + chat_quality_review = None + + if topic_enabled or golden_quote_enabled or chat_quality_enabled: + async with self.llm_semaphore: + logger.debug(f"[LLM] 已进入增量分析队列 (UMO Group: {group_id})") + ( + topics, + golden_quotes, + token_usage, + chat_quality_review, + ) = 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, + chat_quality_enabled=chat_quality_enabled, + ) + + new_topics = [ + { + "topic": t.topic, + "contributors": t.contributors, + "detail": t.detail, + "contributor_ids": t.contributor_ids, + } + for t in topics + ] + + new_quotes = [ + { + "content": q.content, + "sender": q.sender, + "reason": q.reason, + "user_id": q.user_id, + } + for q in golden_quotes + ] + + token_usage_dict = { + "prompt_tokens": token_usage.prompt_tokens, + "completion_tokens": token_usage.completion_tokens, + "total_tokens": token_usage.total_tokens, + } + + user_stats = self._convert_user_activity_for_merge( + user_activity, combined_messages + ) + + 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, + } + + chat_quality_dict = None + if chat_quality_review: + chat_quality_dict = { + "title": chat_quality_review.title, + "subtitle": chat_quality_review.subtitle, + "dimensions": [ + { + "name": d.name, + "percentage": d.percentage, + "comment": d.comment, + "color": d.color, + } + for d in chat_quality_review.dimensions + ], + "summary": chat_quality_review.summary, + } + + participant_ids = list({msg.sender_id for msg in combined_messages}) + last_message_timestamp = max( + (msg.timestamp for msg in combined_messages), default=0 + ) + characters_count = sum(msg.get_text_length() for msg in combined_messages) + + batch = IncrementalBatch( + group_id=group_id, + timestamp=time_mod.time(), + messages_count=len(combined_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, + chat_quality_review=chat_quality_dict, + last_message_timestamp=last_message_timestamp, + participant_ids=participant_ids, + ) + + await self.incremental_store.save_batch(batch) + + import time + + safe_now = int(time.time()) + 60 + for progress_key, ts in source_watermarks.items(): + safe_ts = min(ts, safe_now) + await self.incremental_store.update_last_analyzed_timestamp( + progress_key, safe_ts + ) + + logger.info( + f"[UMOGroup] 组 {group_id} 增量分析完成: " + f"消息={len(combined_messages)}, " + f"话题={len(new_topics)}, 金句={len(new_quotes)}" + ) + + return { + "success": True, + "batch_summary": batch.get_summary(), + "messages_count": len(combined_messages), + "group_id": group_id, + "platform_id": platform_hint, + } + + @staticmethod + def _build_source_progress_key(group_id: str, source_umo: str) -> str: + """构建用于单个来源水位线的进度键，避免非法字符影响 KV。""" + + def _normalize(value: str) -> str: + return "".join( + ch if ch.isalnum() or ch in {"_", "-", ":"} else "_" for ch in value + ) + + return f"{_normalize(group_id)}__{_normalize(source_umo)}" + async def execute_incremental_final_report( - self, group_id: str, platform_id: str | None = None + self, + group_id: str, + platform_id: str | None = None, + require_adapter: bool = True, ) -> dict[str, Any]: """ 基于滑动窗口内的增量批次生成最终报告。 @@ -621,8 +964,8 @@ async def execute_incremental_final_report( ) # 5. 获取适配器（报告发送需要） - adapter = self.bot_manager.get_adapter(platform_id) - if not adapter: + adapter = self.bot_manager.get_adapter(platform_id) if platform_id else None + if require_adapter and not adapter: raise ValueError(f"未找到平台 {platform_id} 的适配器") # 6. 执行分析相关的变量准备 diff --git a/src/infrastructure/config/config_manager.py b/src/infrastructure/config/config_manager.py index 28a8caa6..12339264 100644 --- a/src/infrastructure/config/config_manager.py +++ b/src/infrastructure/config/config_manager.py @@ -3,6 +3,7 @@ 负责处理插件配置和PDF依赖检查 """ +import re import sys from astrbot.api import AstrBotConfig @@ -25,6 +26,9 @@ class ConfigManager: - prompts: 提示词模板 """ + UMO_GROUP_PREFIX = "umoGroup:" + UMO_GROUP_ID_INVALID_PATTERN = re.compile(r'[\\/:*?"<>|.;\s\x00-\x1f]') + def __init__(self, config: AstrBotConfig): self.config = config self._playwright_available = False @@ -64,6 +68,8 @@ def is_group_allowed(self, group_id_or_umo: str) -> bool: glist = [str(g) for g in self.get_group_list()] target = str(group_id_or_umo) + target_umo_groups = set(self.find_umo_groups_by_source(target)) + target_simple_id = target.split(":")[-1] if ":" in target else target target_parent_id = ( target_simple_id.split("#", 1)[0] @@ -77,6 +83,9 @@ def _is_match( target_simple_id: str, target_parent_id: str, ) -> bool: + if self.is_umo_group_id(item): + normalized = self.normalize_umo_group_id(item) + return normalized in target_umo_groups if ":" in item: if item == target: return True @@ -745,6 +754,107 @@ def get_incremental_stagger_seconds(self) -> int: """获取多群增量分析的交错间隔（秒），避免 API 压力""" return self._get_group("incremental").get("incremental_stagger_seconds", 30) + # ========== UMO Group 配置 ========== + + def normalize_umo_group_id(self, group_id: str | None) -> str | None: + """规范化 UMO Group ID，自动补全前缀。""" + if not group_id or not isinstance(group_id, str): + return None + group_id = group_id.strip() + if not group_id: + return None + normalized = ( + group_id + if group_id.startswith(self.UMO_GROUP_PREFIX) + else f"{self.UMO_GROUP_PREFIX}{group_id}" + ) + + raw_id = normalized[len(self.UMO_GROUP_PREFIX) :] + if not raw_id or self.UMO_GROUP_ID_INVALID_PATTERN.search(raw_id): + logger.warning(f"UMO Group ID 包含非法字符，已忽略: {group_id}") + return None + return normalized + + def is_umo_group_id(self, value: str | None) -> bool: + """判断字符串是否表示 UMO Group ID。""" + if not value or not isinstance(value, str): + return False + return value.strip().startswith(self.UMO_GROUP_PREFIX) + + def parse_umo_string(self, umo: str) -> tuple[str | None, str | None]: + """ + 解析 UMO 字符串，返回 (platform_id, session_id)。 + 如果格式不合法，返回 (None, None)。 + """ + if not umo or not isinstance(umo, str): + return None, None + parts = umo.split(":", 2) + if len(parts) < 3: + return None, None + platform_id = parts[0].strip() + session_id = parts[2].strip() + if not platform_id or not session_id: + return None, None + return platform_id, session_id + + def get_umo_groups(self) -> list[dict]: + """获取 UMO Group 配置列表。""" + groups = self._get_group("umo_groups") + if isinstance(groups, list): + return groups + return [] + + def get_umo_group_map(self) -> dict[str, dict]: + """以规范化 ID 为键的 UMO Group 映射。""" + mapping: dict[str, dict] = {} + for item in self.get_umo_groups(): + if not isinstance(item, dict): + continue + group_id = self.normalize_umo_group_id(item.get("id")) + if not group_id: + continue + mapping[group_id] = item + return mapping + + def get_umo_group(self, group_id: str) -> dict | None: + """根据 ID 获取单个 UMO Group 配置。""" + normalized = self.normalize_umo_group_id(group_id) + if not normalized: + return None + return self.get_umo_group_map().get(normalized) + + def get_umo_group_sources(self, group_id: str) -> list[str]: + """获取指定 UMO Group 的来源 UMO 列表。""" + group = self.get_umo_group(group_id) + if not group: + return [] + sources = group.get("source_umos", []) + if isinstance(sources, list): + return [str(x).strip() for x in sources if str(x).strip()] + return [] + + def get_umo_group_outputs(self, group_id: str) -> list[str]: + """获取指定 UMO Group 的输出 UMO 列表。""" + group = self.get_umo_group(group_id) + if not group: + return [] + outputs = group.get("output_umos", []) + if isinstance(outputs, list): + return [str(x).strip() for x in outputs if str(x).strip()] + return [] + + def find_umo_groups_by_source(self, source_umo: str) -> list[str]: + """查找包含指定来源 UMO 的所有 UMO Group ID（规范化）。""" + source_norm = str(source_umo).strip() + if not source_norm: + return [] + result: list[str] = [] + for gid, group in self.get_umo_group_map().items(): + sources = group.get("source_umos", []) or [] + if any(source_norm == str(s).strip() for s in sources): + result.append(gid) + return result + @property def playwright_available(self) -> bool: """检查playwright是否可用""" diff --git a/src/infrastructure/reporting/dispatcher.py b/src/infrastructure/reporting/dispatcher.py index 872187b3..3a254243 100644 --- a/src/infrastructure/reporting/dispatcher.py +++ b/src/infrastructure/reporting/dispatcher.py @@ -35,23 +35,31 @@ async def dispatch( group_id: str, analysis_result: dict[str, Any], platform_id: str | None = None, + report_group_id: str | None = None, ): """ 分发分析报告 """ trace_id = TraceContext.get() output_format = self.config_manager.get_output_format() + target_group_id = report_group_id or group_id logger.info( f"[{trace_id}] 正在分发群 {group_id} 的报告 (格式: {output_format})" ) success = False if output_format == "image": - success = await self._dispatch_image(group_id, analysis_result, platform_id) + success = await self._dispatch_image( + group_id, analysis_result, platform_id, target_group_id + ) elif output_format == "pdf": - success = await self._dispatch_pdf(group_id, analysis_result, platform_id) + success = await self._dispatch_pdf( + group_id, analysis_result, platform_id, target_group_id + ) elif output_format == "html": - success = await self._dispatch_html(group_id, analysis_result, platform_id) + success = await self._dispatch_html( + group_id, analysis_result, platform_id, target_group_id + ) else: success = await self._dispatch_text(group_id, analysis_result, platform_id) @@ -61,7 +69,11 @@ async def dispatch( logger.warning(f"[{trace_id}] 群 {group_id} 的报告分发失败") async def _dispatch_image( - self, group_id: str, analysis_result: dict[str, Any], platform_id: str | None + self, + group_id: str, + analysis_result: dict[str, Any], + platform_id: str | None, + report_group_id: str | None = None, ) -> bool: trace_id = TraceContext.get() # 1. 检查渲染函数 @@ -84,7 +96,7 @@ async def avatar_url_getter(user_id: str): image_url, html_content = await self.report_generator.generate_image_report( analysis_result, - group_id, + report_group_id or group_id, self._html_render_func, avatar_url_getter=avatar_url_getter, ) @@ -110,7 +122,11 @@ async def avatar_url_getter(user_id: str): return await self._dispatch_text(group_id, analysis_result, platform_id) async def _dispatch_pdf( - self, group_id: str, analysis_result: dict[str, Any], platform_id: str | None + self, + group_id: str, + analysis_result: dict[str, Any], + platform_id: str | None, + report_group_id: str | None = None, ) -> bool: trace_id = TraceContext.get() # 1. 检查 Playwright @@ -124,7 +140,7 @@ async def _dispatch_pdf( pdf_path = None try: pdf_path = await self.report_generator.generate_pdf_report( - analysis_result, group_id + analysis_result, report_group_id or group_id ) except Exception as e: logger.error(f"[{trace_id}] Failed to generate PDF report: {e}") @@ -147,14 +163,18 @@ async def _dispatch_pdf( return await self._dispatch_text(group_id, analysis_result, platform_id) async def _dispatch_html( - self, group_id: str, analysis_result: dict[str, Any], platform_id: str | None + self, + group_id: str, + analysis_result: dict[str, Any], + platform_id: str | None, + report_group_id: str | None = None, ) -> bool: trace_id = TraceContext.get() html_path = None try: html_path, json_path = await self.report_generator.generate_html_report( - analysis_result, group_id + analysis_result, report_group_id or group_id ) except Exception as e: logger.error(f"[{trace_id}] Failed to generate HTML report: {e}") diff --git a/src/infrastructure/scheduler/auto_scheduler.py b/src/infrastructure/scheduler/auto_scheduler.py index 0016c51b..b69147f6 100644 --- a/src/infrastructure/scheduler/auto_scheduler.py +++ b/src/infrastructure/scheduler/auto_scheduler.py @@ -334,6 +334,152 @@ async def _get_scheduled_targets( ) return result + def _normalize_group_entry(self, value: str, defined_ids: set[str]) -> str | None: + """将配置项规范化为 UMO Group ID（仅当已定义时才返回）。""" + if not value: + return None + raw = str(value).strip() + if not raw: + return None + + if self.config_manager.is_umo_group_id(raw): + normalized = self.config_manager.normalize_umo_group_id(raw) + return normalized if normalized in defined_ids else None + + normalized = self.config_manager.normalize_umo_group_id(raw) + if normalized in defined_ids: + return normalized + return None + + def _is_group_selected( + self, group_id: str, mode: str, configured_ids: set[str] + ) -> bool: + """根据名单模式判断 UMO Group 是否被选中。""" + if mode == "whitelist": + if not configured_ids: + return False + return group_id in configured_ids + return group_id not in configured_ids + + def _prepare_umo_group_sources(self, source_umos: list[str]) -> list[str]: + """过滤并返回可用的来源 UMO（存在适配器且通过基础权限）。""" + valid: list[str] = [] + for umo in source_umos: + umo_str = str(umo).strip() + if not umo_str: + continue + platform_id, session_id = self.config_manager.parse_umo_string(umo_str) + if not platform_id or not session_id: + logger.warning(f"[UMOGroup] 无效来源 UMO，跳过: {umo}") + continue + if not self.config_manager.is_group_allowed(umo_str): + logger.debug( + f"[UMOGroup] 来源 UMO 未通过基础白/黑名单，跳过: {umo_str}" + ) + continue + adapter = self.bot_manager.get_adapter(platform_id) + if not adapter: + logger.warning( + f"[UMOGroup] 平台 {platform_id} 未加载适配器，跳过来源 {umo_str}" + ) + continue + valid.append(umo_str) + return valid + + def _normalize_output_destinations(self, outputs: list[str]) -> list[dict]: + """将输出 UMO 解析为可发送的目标列表。""" + destinations: list[dict] = [] + seen: set[tuple[str, str]] = set() + for umo in outputs: + umo_str = str(umo).strip() + if not umo_str: + continue + platform_id, session_id = self.config_manager.parse_umo_string(umo_str) + if not platform_id or not session_id: + logger.warning(f"[UMOGroup] 无效输出 UMO，跳过: {umo}") + continue + key = (platform_id, session_id) + if key in seen: + continue + seen.add(key) + destinations.append({"platform_id": platform_id, "group_id": session_id}) + return destinations + + def _get_umo_group_targets( + self, mode_filter: str | None = None + ) -> list[dict[str, object]]: + """解析配置中声明的 UMO Group 目标。""" + group_map = self.config_manager.get_umo_group_map() + if not group_map: + return [] + + defined_ids = set(group_map.keys()) + sched_mode = self.config_manager.get_scheduled_group_list_mode() + sched_list = self.config_manager.get_scheduled_group_list() + incr_mode = self.config_manager.get_incremental_group_list_mode() + incr_list = self.config_manager.get_incremental_group_list() + + sched_ids = { + gid + for item in sched_list + if (gid := self._normalize_group_entry(item, defined_ids)) + } + incr_ids = { + gid + for item in incr_list + if (gid := self._normalize_group_entry(item, defined_ids)) + } + + targets: list[dict[str, object]] = [] + + for gid, cfg in group_map.items(): + if mode_filter in (None, "traditional"): + if self._is_group_selected(gid, sched_mode, sched_ids): + targets.append( + { + "group_id": gid, + "mode": "traditional", + "sources": self._prepare_umo_group_sources( + cfg.get("source_umos", []) + ), + "outputs": self._normalize_output_destinations( + cfg.get("output_umos", []) + ), + } + ) + + if mode_filter in (None, "incremental"): + if self._is_group_selected(gid, incr_mode, incr_ids): + targets.append( + { + "group_id": gid, + "mode": "incremental", + "sources": self._prepare_umo_group_sources( + cfg.get("source_umos", []) + ), + "outputs": self._normalize_output_destinations( + cfg.get("output_umos", []) + ), + } + ) + + filtered: list[dict[str, object]] = [] + for t in targets: + if not t["sources"]: + logger.info(f"[UMOGroup] 组 {t['group_id']} 未找到可用来源，跳过该任务") + continue + if not t["outputs"]: + logger.info(f"[UMOGroup] 组 {t['group_id']} 未配置输出 UMO，跳过该任务") + continue + filtered.append(t) + + if filtered: + logger.info( + f"[UMOGroup] 解析到 {len(filtered)} 个 UMO Group 目标" + + (f" (模式过滤: {mode_filter})" if mode_filter else "") + ) + return filtered + # ================================================================ # 统一报告调度入口 # ================================================================ @@ -351,15 +497,18 @@ async def _run_scheduled_report(self): logger.info("定时报告触发 — 开始解析调度目标") all_targets = await self._get_scheduled_targets() + umo_group_targets = self._get_umo_group_targets() - if not all_targets: + if not all_targets and not umo_group_targets: logger.info("没有配置的群聊需要定时分析") return max_concurrent = self.config_manager.get_max_concurrent_tasks() sem = asyncio.Semaphore(max_concurrent) logger.info( - f"定时报告: {len(all_targets)} 个目标 (并发限制: {max_concurrent})" + f"定时报告: {len(all_targets)} 个群目标，" + f"{len(umo_group_targets)} 个 UMO Group 目标 " + f"(并发限制: {max_concurrent})" ) async def dispatch_group(gid, pid, mode): @@ -373,10 +522,23 @@ async def dispatch_group(gid, pid, mode): gid, pid ) + async def dispatch_umo_group(target: dict): + async with sem: + return await self._perform_umo_group_analysis_with_timeout( + target["group_id"], + target["sources"], + target["outputs"], + target["mode"], + ) + tasks = [] stagger = self.config_manager.get_stagger_seconds() or 2 + combined_targets: list[tuple[str, object]] = [ + ("group", (gid, pid, mode)) for gid, pid, mode in all_targets + ] + [("umo_group", target) for target in umo_group_targets] + # 针对定时大任务加入交错等待，减少瞬间峰值延迟 - for idx, (gid, pid, mode) in enumerate(all_targets): + for idx, entry in enumerate(combined_targets): if self._terminating: logger.info("检测到插件正在停止，取消后续任务创建") break @@ -385,10 +547,18 @@ async def dispatch_group(gid, pid, mode): if idx > 0 and stagger > 0: await asyncio.sleep(stagger) - task = asyncio.create_task( - dispatch_group(gid, pid, mode), - name=f"report_{mode}_{gid}", - ) + if entry[0] == "group": + gid, pid, mode = entry[1] + task = asyncio.create_task( + dispatch_group(gid, pid, mode), + name=f"report_{mode}_{gid}", + ) + else: + target = entry[1] + task = asyncio.create_task( + dispatch_umo_group(target), + name=f"report_{target['mode']}_{target['group_id']}", + ) tasks.append(task) results = await asyncio.gather(*tasks, return_exceptions=True) @@ -399,7 +569,8 @@ async def dispatch_group(gid, pid, mode): error_count = 0 for i, result in enumerate(results): - gid, _, _ = all_targets[i] + label, payload = combined_targets[i] + gid = payload[0] if label == "group" else payload["group_id"] if isinstance(result, DuplicateGroupTaskError): skip_count += 1 elif isinstance(result, Exception): @@ -412,7 +583,7 @@ async def dispatch_group(gid, pid, mode): logger.info( f"定时报告完成 — 成功: {success_count}, 跳过: {skip_count}, " - f"失败: {error_count}, 总计: {len(all_targets)}" + f"失败: {error_count}, 总计: {len(combined_targets)}" ) except Exception as e: @@ -433,6 +604,26 @@ async def _perform_auto_analysis_for_group_with_timeout( except Exception as e: logger.error(f"群 {group_id} 分析任务执行失败: {e}") + async def _perform_umo_group_analysis_with_timeout( + self, + group_id: str, + source_umos: list[str], + output_targets: list[dict], + mode: str, + ): + """为 UMO Group 执行聚合分析（带超时控制）。""" + try: + await asyncio.wait_for( + self._perform_umo_group_analysis( + group_id, source_umos, output_targets, mode + ), + timeout=1800, + ) + except asyncio.TimeoutError: + logger.error(f"[UMOGroup] 组 {group_id} 分析超时（30分钟），跳过该组") + except Exception as e: + logger.error(f"[UMOGroup] 组 {group_id} 分析任务失败: {e}") + async def _perform_auto_analysis_for_group( self, group_id: str, target_platform_id: str | None = None ): @@ -490,6 +681,62 @@ async def _perform_auto_analysis_for_group( finally: logger.debug(f"群 {group_id} 自动分析流程结束") + async def _perform_umo_group_analysis( + self, + group_id: str, + source_umos: list[str], + output_targets: list[dict], + mode: str, + ): + """为 UMO Group 执行聚合分析并分发报告。""" + try: + trace_id = TraceContext.generate(prefix="umoGroup", group_name=group_id) + TraceContext.set(trace_id) + + if self._terminating: + return + + logger.info( + f"[UMOGroup] 开始执行{'增量最终' if mode == 'incremental' else '全量'}分析 " + f"(组: {group_id}, 来源数={len(source_umos)}, 输出数={len(output_targets)})" + ) + + if mode == "incremental": + result = await self.analysis_service.execute_incremental_final_report( + group_id=group_id, + platform_id=None, + require_adapter=False, + ) + else: + result = await self.analysis_service.execute_daily_analysis_for_sources( + group_id=group_id, source_umos=source_umos + ) + + if not result.get("success"): + reason = result.get("reason", "unknown") + logger.info(f"[UMOGroup] 组 {group_id} 分析跳过: {reason}") + return + + analysis_result = result["analysis_result"] + + for dest in output_targets: + await self.report_dispatcher.dispatch( + dest["group_id"], + analysis_result, + dest.get("platform_id"), + report_group_id=group_id, + ) + + logger.info(f"[UMOGroup] 组 {group_id} 分析任务执行成功") + + except DuplicateGroupTaskError: + logger.debug(f"[UMOGroup] 组 {group_id} 任务因并发锁冲突跳过") + raise + except Exception as e: + logger.error(f"[UMOGroup] 组 {group_id} 分析执行失败: {e}", exc_info=True) + finally: + logger.debug(f"[UMOGroup] 组 {group_id} 分析流程结束") + # ================================================================ # 增量模式：增量分析 # ================================================================ @@ -501,55 +748,79 @@ async def _run_incremental_analysis(self): try: logger.info("开始执行自动增量分析（并发模式）") - # 仅选取模式为 incremental 的目标群 incr_targets = await self._get_scheduled_targets(mode_filter="incremental") + incr_umo_groups = self._get_umo_group_targets(mode_filter="incremental") - if not incr_targets: + if not incr_targets and not incr_umo_groups: logger.info("没有配置为增量模式的群聊需要增量分析") return - target_list = incr_targets stagger = self.config_manager.get_incremental_stagger_seconds() max_concurrent = self.config_manager.get_max_concurrent_tasks() logger.info( - f"将为 {len(target_list)} 个群聊执行增量分析 " + f"将为 {len(incr_targets)} 个群聊与 {len(incr_umo_groups)} 个 UMO Group 执行增量分析 " f"(并发限制: {max_concurrent}, 交错间隔: {stagger}秒)" ) sem = asyncio.Semaphore(max_concurrent) - async def staggered_incremental(idx, gid, pid): + async def staggered_incremental(idx, entry): if idx > 0 and stagger > 0: await asyncio.sleep(stagger * idx) async with sem: - result = ( - await self._perform_incremental_analysis_for_group_with_timeout( + if entry[0] == "group": + gid, pid = entry[1], entry[2] + 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 + ) + else: + target = entry[1] + gid = target["group_id"] + result = await self._perform_incremental_analysis_for_umo_group_with_timeout( + gid, target["sources"] + ) + + if self.config_manager.get_incremental_report_immediately(): + if isinstance(result, dict) and result.get("success"): + logger.info( + f"增量分析立即报告模式生效，正在为 UMO 组 {gid} 生成报告..." + ) + await self._perform_umo_group_analysis_with_timeout( + gid, + target["sources"], + target["outputs"], + "incremental", + ) return result analysis_tasks = [] - for idx, (gid, pid, _mode) in enumerate(target_list): + combined_targets: list[tuple[str, object]] = [ + ("group", (gid, pid)) for gid, pid, _mode in incr_targets + ] + [("umo_group", target) for target in incr_umo_groups] + + for idx, entry in enumerate(combined_targets): if self._terminating: logger.info("检测到插件正在停止，取消后续增量分析任务创建") break task = asyncio.create_task( - staggered_incremental(idx, gid, pid), - name=f"incremental_group_{gid}", + staggered_incremental(idx, entry), + name=( + f"incremental_group_{entry[1][0]}" + if entry[0] == "group" + else f"incremental_umo_group_{entry[1]['group_id']}" + ), ) analysis_tasks.append(task) @@ -560,7 +831,8 @@ async def staggered_incremental(idx, gid, pid): error_count = 0 for i, result in enumerate(results): - gid, _, _ = target_list[i] + entry = combined_targets[i] + gid = entry[1][0] if entry[0] == "group" else entry[1]["group_id"] if isinstance(result, DuplicateGroupTaskError): skip_count += 1 elif isinstance(result, Exception): @@ -573,7 +845,7 @@ async def staggered_incremental(idx, gid, pid): logger.info( f"增量分析完成 - 成功: {success_count}, 跳过: {skip_count}, " - f"失败: {error_count}, 总计: {len(target_list)}" + f"失败: {error_count}, 总计: {len(combined_targets)}" ) except Exception as e: @@ -652,6 +924,77 @@ async def _perform_incremental_analysis_for_group( finally: logger.debug(f"群 {group_id} 增量分析流程结束") + async def _perform_incremental_analysis_for_umo_group_with_timeout( + self, group_id: str, source_umos: list[str] + ): + """为指定 UMO Group 执行增量分析（带超时控制，10分钟）""" + try: + result = await asyncio.wait_for( + self._perform_incremental_analysis_for_umo_group(group_id, source_umos), + timeout=600, + ) + return result + except asyncio.TimeoutError: + logger.error(f"[UMOGroup] 组 {group_id} 增量分析超时（10分钟），跳过") + return {"success": False, "reason": "timeout"} + except Exception as e: + logger.error(f"[UMOGroup] 组 {group_id} 增量分析任务执行失败: {e}") + return {"success": False, "reason": str(e)} + + async def _perform_incremental_analysis_for_umo_group( + self, group_id: str, source_umos: list[str] + ): + """为 UMO Group 执行增量分析（仅累计批次，不发送报告）。""" + try: + trace_id = TraceContext.generate(prefix="umoGroupIncr", group_name=group_id) + TraceContext.set(trace_id) + + if self._terminating: + return + + logger.info( + f"[UMOGroup] 开始执行增量分析 (组: {group_id}, 来源数={len(source_umos)})" + ) + + if not self.bot_manager.is_ready_for_auto_analysis(): + logger.warning( + f"[UMOGroup] 组 {group_id} 增量分析跳过：bot管理器未就绪" + ) + return {"success": False, "reason": "bot_not_ready"} + + result = ( + await self.analysis_service.execute_incremental_analysis_for_sources( + group_id=group_id, source_umos=source_umos + ) + ) + + if not result.get("success"): + reason = result.get("reason", "unknown") + logger.info(f"[UMOGroup] 组 {group_id} 增量分析跳过: {reason}") + return result + + batch_summary = result.get("batch_summary", {}) + logger.info( + f"[UMOGroup] 组 {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 DuplicateGroupTaskError: + logger.debug( + f"[UMOGroup] 组 {group_id} 增量分析因并发锁冲突而跳过（已在运行）" + ) + return {"success": False, "reason": "already_running"} + except Exception as e: + logger.error( + f"[UMOGroup] 组 {group_id} 增量分析执行失败: {e}", exc_info=True + ) + return {"success": False, "reason": str(e)} + finally: + logger.debug(f"[UMOGroup] 组 {group_id} 增量分析流程结束") + # ================================================================ # 增量最终报告（单群）与回退逻辑 # ================================================================ From 3a58b32ed6d9145f8ab4a2b86a89c2c3fc9ce5df Mon Sep 17 00:00:00 2001 From: Leko Date: Fri, 3 Apr 2026 01:38:14 +0800 Subject: [PATCH 2/3] feat: Enhance UMO Group handling with type definitions and logging improvements --- .../services/analysis_application_service.py | 4 +-- src/infrastructure/config/config_manager.py | 7 +++- src/infrastructure/reporting/dispatcher.py | 12 +++++-- .../scheduler/auto_scheduler.py | 36 ++++++++++++++----- 4 files changed, 43 insertions(+), 16 deletions(-) diff --git a/src/application/services/analysis_application_service.py b/src/application/services/analysis_application_service.py index 572c97e8..bd9f701e 100644 --- a/src/application/services/analysis_application_service.py +++ b/src/application/services/analysis_application_service.py @@ -867,9 +867,7 @@ async def execute_incremental_analysis_for_sources( await self.incremental_store.save_batch(batch) - import time - - safe_now = int(time.time()) + 60 + safe_now = int(time_mod.time()) + 60 for progress_key, ts in source_watermarks.items(): safe_ts = min(ts, safe_now) await self.incremental_store.update_last_analyzed_timestamp( diff --git a/src/infrastructure/config/config_manager.py b/src/infrastructure/config/config_manager.py index 12339264..96703841 100644 --- a/src/infrastructure/config/config_manager.py +++ b/src/infrastructure/config/config_manager.py @@ -68,7 +68,12 @@ def is_group_allowed(self, group_id_or_umo: str) -> bool: glist = [str(g) for g in self.get_group_list()] target = str(group_id_or_umo) - target_umo_groups = set(self.find_umo_groups_by_source(target)) + # 仅当名单中包含 UMO Group ID 时才执行反向索引扫描， + # 避免普通群号/UMO 场景每次都遍历全部 UMO Group。 + if any(self.is_umo_group_id(item) for item in glist): + target_umo_groups = set(self.find_umo_groups_by_source(target)) + else: + target_umo_groups = set() target_simple_id = target.split(":")[-1] if ":" in target else target target_parent_id = ( diff --git a/src/infrastructure/reporting/dispatcher.py b/src/infrastructure/reporting/dispatcher.py index 3a254243..141ebe49 100644 --- a/src/infrastructure/reporting/dispatcher.py +++ b/src/infrastructure/reporting/dispatcher.py @@ -43,9 +43,15 @@ async def dispatch( trace_id = TraceContext.get() output_format = self.config_manager.get_output_format() target_group_id = report_group_id or group_id - logger.info( - f"[{trace_id}] 正在分发群 {group_id} 的报告 (格式: {output_format})" - ) + if report_group_id: + logger.info( + f"[{trace_id}] 正在分发报告 (逻辑群: {report_group_id}, " + f"发送目标群: {group_id}, 格式: {output_format})" + ) + else: + logger.info( + f"[{trace_id}] 正在分发群 {group_id} 的报告 (格式: {output_format})" + ) success = False if output_format == "image": diff --git a/src/infrastructure/scheduler/auto_scheduler.py b/src/infrastructure/scheduler/auto_scheduler.py index b69147f6..7c8cfe3a 100644 --- a/src/infrastructure/scheduler/auto_scheduler.py +++ b/src/infrastructure/scheduler/auto_scheduler.py @@ -5,7 +5,7 @@ import asyncio import time as time_mod -from typing import Any +from typing import Any, Literal, TypedDict from apscheduler.triggers.cron import CronTrigger @@ -17,6 +17,22 @@ from ..reporting.dispatcher import ReportDispatcher +class UMOOutputDestination(TypedDict): + """UMO Group 报告输出目标。""" + + platform_id: str + group_id: str + + +class UMOGroupTarget(TypedDict): + """UMO Group 任务目标。""" + + group_id: str + mode: Literal["traditional", "incremental"] + sources: list[str] + outputs: list[UMOOutputDestination] + + class AutoScheduler: """自动调度器，支持传统模式和增量模式""" @@ -386,9 +402,11 @@ def _prepare_umo_group_sources(self, source_umos: list[str]) -> list[str]: valid.append(umo_str) return valid - def _normalize_output_destinations(self, outputs: list[str]) -> list[dict]: + def _normalize_output_destinations( + self, outputs: list[str] + ) -> list[UMOOutputDestination]: """将输出 UMO 解析为可发送的目标列表。""" - destinations: list[dict] = [] + destinations: list[UMOOutputDestination] = [] seen: set[tuple[str, str]] = set() for umo in outputs: umo_str = str(umo).strip() @@ -407,7 +425,7 @@ def _normalize_output_destinations(self, outputs: list[str]) -> list[dict]: def _get_umo_group_targets( self, mode_filter: str | None = None - ) -> list[dict[str, object]]: + ) -> list[UMOGroupTarget]: """解析配置中声明的 UMO Group 目标。""" group_map = self.config_manager.get_umo_group_map() if not group_map: @@ -430,7 +448,7 @@ def _get_umo_group_targets( if (gid := self._normalize_group_entry(item, defined_ids)) } - targets: list[dict[str, object]] = [] + targets: list[UMOGroupTarget] = [] for gid, cfg in group_map.items(): if mode_filter in (None, "traditional"): @@ -463,7 +481,7 @@ def _get_umo_group_targets( } ) - filtered: list[dict[str, object]] = [] + filtered: list[UMOGroupTarget] = [] for t in targets: if not t["sources"]: logger.info(f"[UMOGroup] 组 {t['group_id']} 未找到可用来源，跳过该任务") @@ -522,7 +540,7 @@ async def dispatch_group(gid, pid, mode): gid, pid ) - async def dispatch_umo_group(target: dict): + async def dispatch_umo_group(target: UMOGroupTarget): async with sem: return await self._perform_umo_group_analysis_with_timeout( target["group_id"], @@ -608,7 +626,7 @@ async def _perform_umo_group_analysis_with_timeout( self, group_id: str, source_umos: list[str], - output_targets: list[dict], + output_targets: list[UMOOutputDestination], mode: str, ): """为 UMO Group 执行聚合分析（带超时控制）。""" @@ -685,7 +703,7 @@ async def _perform_umo_group_analysis( self, group_id: str, source_umos: list[str], - output_targets: list[dict], + output_targets: list[UMOOutputDestination], mode: str, ): """为 UMO Group 执行聚合分析并分发报告。""" From b238ff41b848a362de16026db121adafaa1ae900 Mon Sep 17 00:00:00 2001 From: Leko Date: Fri, 3 Apr 2026 20:04:15 +0800 Subject: [PATCH 3/3] feat: Update UMO Group configuration to use template_list and enhance data cleaning --- _conf_schema.json | 51 +++++++++++---------- src/infrastructure/config/config_manager.py | 9 +++- 2 files changed, 35 insertions(+), 25 deletions(-) diff --git a/_conf_schema.json b/_conf_schema.json index 75d1112b..6f4eabca 100644 --- a/_conf_schema.json +++ b/_conf_schema.json @@ -100,32 +100,35 @@ }, "umo_groups": { "description": "UMO 分组", - "type": "list", + "type": "template_list", "hint": "将多个群的 UMO 组合为一个 UMO Group，用于聚合分析和指定输出 UMO。UMO Group ID 会以 umoGroup:ID 的形式在其他配置中引用。", "default": [], - "items": { - "type": "object", - "items": { - "id": { - "type": "string", - "description": "UMO Group ID", - "hint": "用于标识此 UMO Group，引用时以 umoGroup:ID 的形式填写。仅允许字母、数字、下划线、短横线，禁止包含空格、点号、分号等特殊字符。" - }, - "source_umos": { - "type": "list", - "description": "来源 UMO 列表", - "hint": "参与聚合分析的来源 UMO，格式为 platform:GroupMessage:session_id。", - "items": { - "type": "string" - } - }, - "output_umos": { - "type": "list", - "description": "输出 UMO 列表", - "hint": "聚合报告要发送到的目标 UMO，支持填写多个 UMO。", - "default": [], - "items": { - "type": "string" + "templates": { + "umo_group": { + "name": "UMO 组合", + "hint": "一个 UMO 组合，用于配置来源 UMO 与输出 UMO。", + "items": { + "id": { + "type": "string", + "description": "UMO Group ID", + "hint": "用于标识此 UMO Group，引用时以 umoGroup:ID 的形式填写。仅允许字母、数字、下划线、短横线，禁止包含空格、点号、分号等特殊字符。" + }, + "source_umos": { + "type": "list", + "description": "来源 UMO 列表", + "hint": "参与聚合分析的来源 UMO。", + "items": { + "type": "string" + } + }, + "output_umos": { + "type": "list", + "description": "输出 UMO 列表", + "hint": "聚合报告要发送到的目标 UMO，支持填写多个 UMO。", + "default": [], + "items": { + "type": "string" + } } } } diff --git a/src/infrastructure/config/config_manager.py b/src/infrastructure/config/config_manager.py index 96703841..3a19c0a4 100644 --- a/src/infrastructure/config/config_manager.py +++ b/src/infrastructure/config/config_manager.py @@ -806,7 +806,14 @@ def get_umo_groups(self) -> list[dict]: """获取 UMO Group 配置列表。""" groups = self._get_group("umo_groups") if isinstance(groups, list): - return groups + cleaned: list[dict] = [] + for item in groups: + if not isinstance(item, dict): + continue + entry = item.copy() + entry.pop("__template_key", None) + cleaned.append(entry) + return cleaned return [] def get_umo_group_map(self) -> dict[str, dict]: