""" 群日常分析插件 基于群聊记录生成精美的日常分析报告，包含话题总结、用户画像、统计数据等 重构版本 - 使用模块化架构，支持跨平台 """ import asyncio import os from collections.abc import AsyncGenerator, Callable from pathlib import Path from astrbot.api import AstrBotConfig from astrbot.api import logger as astrbot_logger from astrbot.api.event import AstrMessageEvent, filter from astrbot.api.event.filter import PermissionType from astrbot.api.star import Context, Star, StarTools # File is only available via astrbot.core (internal API — may change). from astrbot.core.message.components import File from .src.application.commands.template_command_service import ( TemplateCommandService, ) from .src.application.services.analysis_application_service import ( AnalysisApplicationService, DuplicateGroupTaskError, ) from .src.application.services.message_processing_service import ( MessageProcessingService, ) from .src.domain.services.analysis_domain_service import AnalysisDomainService from .src.domain.services.incremental_merge_service import IncrementalMergeService from .src.domain.services.statistics_service import StatisticsService from .src.infrastructure.analysis.llm_analyzer import LLMAnalyzer from .src.infrastructure.config.config_manager import ConfigManager from .src.infrastructure.messaging.message_sender import MessageSender from .src.infrastructure.persistence.history_manager import HistoryManager from .src.infrastructure.persistence.incremental_store import IncrementalStore from .src.infrastructure.persistence.platform_group_registry import ( PlatformGroupRegistry, ) from .src.infrastructure.platform.bot_manager import BotManager from .src.infrastructure.platform.template_preview import ( TelegramTemplatePreviewHandler, TemplatePreviewRouter, ) from .src.infrastructure.reporting.generators import ReportGenerator from .src.infrastructure.scheduler.auto_scheduler import AutoScheduler from .src.infrastructure.visualization.activity_charts import ActivityVisualizer from .src.shared.constants import PLUGIN_NAME from .src.shared.trace_context import TraceContext, TraceLogFilter from .src.utils.logger import logger from .src.utils.resilience import GlobalRateLimiter class GroupDailyAnalysis(Star): """群分析插件主类""" # ── 显式类型声明 (由 __init__ 初始化) ── config: AstrBotConfig config_manager: ConfigManager bot_manager: BotManager history_manager: HistoryManager report_generator: ReportGenerator html_render: Callable platform_group_registry: PlatformGroupRegistry statistics_service: StatisticsService analysis_domain_service: AnalysisDomainService llm_analyzer: LLMAnalyzer incremental_store: IncrementalStore incremental_merge_service: IncrementalMergeService analysis_service: AnalysisApplicationService message_processing_service: MessageProcessingService template_command_service: TemplateCommandService telegram_template_preview_handler: TelegramTemplatePreviewHandler template_preview_router: TemplatePreviewRouter auto_scheduler: AutoScheduler message_sender: MessageSender def __init__(self, context: Context, config: AstrBotConfig): super().__init__(context) self.config = config # 1. 基础设施层 self.config_manager = ConfigManager(config) self.bot_manager = BotManager(self.config_manager) self.bot_manager.set_context(context) self.bot_manager.set_plugin_instance(self) self.history_manager = HistoryManager(self) plugin_data_dir = StarTools.get_data_dir(PLUGIN_NAME) self.report_generator = ReportGenerator(self.config_manager, plugin_data_dir) # Telegram 注册表 (持久层) self.platform_group_registry = PlatformGroupRegistry(self) # 2. 领域层 activity_visualizer = ActivityVisualizer() self.statistics_service = StatisticsService(activity_visualizer) self.analysis_domain_service = AnalysisDomainService() # 3. 分析核心 (LLM Bridge) self.llm_analyzer = LLMAnalyzer(context, self.config_manager) # 4. 增量分析组件 self.incremental_store = IncrementalStore(self) self.incremental_merge_service = IncrementalMergeService() # 5. 应用层 self.analysis_service = AnalysisApplicationService( self.config_manager, self.bot_manager, self.history_manager, self.report_generator, self.llm_analyzer, self.statistics_service, self.analysis_domain_service, incremental_store=self.incremental_store, incremental_merge_service=self.incremental_merge_service, ) # 消息处理服务 self.message_processing_service = MessageProcessingService( context, self.platform_group_registry ) self.template_command_service = TemplateCommandService( plugin_root=os.path.dirname(__file__) ) self.telegram_template_preview_handler = TelegramTemplatePreviewHandler( config_manager=self.config_manager, template_service=self.template_command_service, ) self.template_preview_router = TemplatePreviewRouter( handlers=[self.telegram_template_preview_handler] ) # 调度与发送 self.message_sender = MessageSender(self.bot_manager, self.config_manager) self.auto_scheduler = AutoScheduler( self.config_manager, self.analysis_service, self.bot_manager, self.report_generator, self.html_render, plugin_instance=self, ) # 同步全局限流并进行初始化配置 GlobalRateLimiter.get_instance(self.config_manager.get_llm_max_concurrent()) self._initialized = False self._terminating = False # 生命周期标志 self._init_lock = asyncio.Lock() self._background_tasks: set[asyncio.Task] = set() # 异步注册任务，处理插件重载情况 try: loop = asyncio.get_running_loop() self._init_task = loop.create_task( self._run_initialization("Plugin Reload/Init") ) self._background_tasks.add(self._init_task) self._init_task.add_done_callback(self._background_tasks.discard) except RuntimeError: self._init_task = None # orchestrators 缓存已移至 应用层逻辑 (分析服务) 或 暂时移除以简化。 # 如果需要高性能缓存，后续可由 AnalysisApplicationService 内部维护。 @filter.on_platform_loaded() async def on_platform_loaded(self): """平台加载完成后初始化""" await self._run_initialization("Platform Loaded") async def _run_initialization(self, source: str): """统一初始化逻辑""" async with self._init_lock: # 如果已经成功发现过平台，且不是来自 Platform Loaded 的强制触发，则跳过 if ( self._initialized and self.bot_manager and self.bot_manager.get_platform_count() > 0 and source != "Platform Loaded" ): return # 稍微延迟，确保 context 和环境稳定 # 针对极少数环境，2秒可能不足以让平台管理器就绪，增加到 5秒 await asyncio.sleep(5) # [加固] 如果在等待期间插件已被卸载（terminate），则直接退出 if not self.bot_manager: return try: # 注册 TraceID 过滤器 trace_filter = TraceLogFilter() if not any( isinstance(f, TraceLogFilter) for f in astrbot_logger.filters ): astrbot_logger.addFilter(trace_filter) astrbot_logger.info("[Trace] TraceID 日志追踪已启用") logger.info(f"正在执行插件初始化 (来源: {source})...") # 0. 自动升级旧版 prompt 模板（str.format -> string.Template）并回写配置 try: self.config_manager.upgrade_prompt_templates() except Exception as e: logger.warning(f"自动升级 prompt 模板失败: {e}") # 1. 尝试发现 bot 实例 await self.bot_manager.initialize_from_config() # 2. 注册预览路由器 if self.template_preview_router: await self.template_preview_router.ensure_handlers_registered( self.context ) # 3. 强制注册定时分析任务 if self.auto_scheduler: self.auto_scheduler.schedule_jobs(self.context) self._initialized = True self._discovery_run = True logger.info(f"插件任务注册完成 (来源: {source})") except Exception as e: logger.error(f"插件初始化失败: {e}", exc_info=True) async def terminate(self): """插件被卸载/停用时调用，清理资源""" if self._terminating: return self._terminating = True try: logger.info("开始清理群日常分析插件资源...") # 1. 停止所有后台任务 if self._background_tasks: logger.info(f"正在取消 {len(self._background_tasks)} 个运行中的任务...") for task in self._background_tasks: if not task.done(): task.cancel() # 等待任务结束，给予 3 秒宽限期 try: await asyncio.wait(list(self._background_tasks), timeout=3.0) except Exception: pass self._background_tasks.clear() # 2. 停止各个组件 (顺序：先调度器，后底层服务) if self.auto_scheduler: logger.debug("正在停止自动调度器...") self.auto_scheduler.unschedule_jobs(self.context) if self.template_preview_router: await self.template_preview_router.unregister_handlers() if self.report_generator: await self.report_generator.close() # 3. [关键修复] 只有在任务全部清理后，才清理引用。 # 实际上，在 terminate 结束后，self 本身就会被 GC 释放， # 这里的显式 None 更多是为了协助循环引用清理，但由于异步任务存在竞态， # 我们可以通过 check _terminating 标志位来保护。 # 为了彻底解决 #125，我们保留引用，让 GC 自然回收。 logger.info("群日常分析插件资源清理完成") except Exception as e: logger.error(f"插件资源清理失败: {e}") # ==================== Telegram 消息拦截器 ==================== @filter.event_message_type(filter.EventMessageType.GROUP_MESSAGE) @filter.platform_adapter_type(filter.PlatformAdapterType.TELEGRAM) async def intercept_telegram_messages(self, event: AstrMessageEvent): """ 拦截 Telegram 群消息并存储到数据库 委托给 MessageProcessingService 处理 """ try: await self.message_processing_service.process_message(event) except (ValueError, RuntimeError) as e: logger.warning(f"[Telegram] 消息存储失败: {e}") except Exception as e: logger.error(f"[Telegram] 消息存储异常: {e}", exc_info=True) @filter.event_message_type(filter.EventMessageType.GROUP_MESSAGE) @filter.platform_adapter_type( filter.PlatformAdapterType.QQOFFICIAL | filter.PlatformAdapterType.QQOFFICIAL_WEBHOOK ) async def intercept_qq_official_messages(self, event: AstrMessageEvent): """缓存 QQ 官方机器人群消息；频道消息不在本插件适配范围内。""" raw_message = getattr(getattr(event, "message_obj", None), "raw_message", None) if isinstance(raw_message, dict): author = raw_message.get("author") or {} group_openid = str(raw_message.get("group_openid", "") or "").strip() member_openid = str( author.get("member_openid", "") if isinstance(author, dict) else "" ).strip() else: author = getattr(raw_message, "author", None) group_openid = str(getattr(raw_message, "group_openid", "") or "").strip() member_openid = str(getattr(author, "member_openid", "") or "").strip() if not group_openid or not member_openid: return try: await self.message_processing_service.process_message(event) except (ValueError, RuntimeError) as e: logger.warning(f"[QQOfficial] 消息存储失败: {e}") except Exception as e: logger.error(f"[QQOfficial] 消息存储异常: {e}", exc_info=True) async def get_telegram_seen_group_ids( self, platform_id: str | None = None ) -> list[str]: """读取 Telegram 已见群/话题列表（给调度器回退使用）。""" return await self.platform_group_registry.get_all_group_ids(platform_id) async def get_seen_group_ids(self, platform_id: str | None = None) -> list[str]: """读取任意事件驱动平台已经见过的群组。""" return await self.platform_group_registry.get_all_group_ids(platform_id) def _get_group_id_from_event(self, event: AstrMessageEvent) -> str | None: """从消息事件中安全获取群组 ID""" # 保留此辅助方法，因为在其他 command 中仍被频繁使用 try: group_id = event.get_group_id() return group_id if group_id else None except Exception: return None def _get_platform_id_from_event(self, event: AstrMessageEvent) -> str: """从消息事件中获取平台唯一 ID""" # 保留此辅助方法，因为在其他 command 中仍被频繁使用 try: return event.get_platform_id() except Exception: # 后备方案：从元数据获取 if ( hasattr(event, "platform_meta") and event.platform_meta and hasattr(event.platform_meta, "id") ): return event.platform_meta.id return "default" # ================================================================ # 图片报告上传到群文件 / 群相册（仅 QQ 平台 image 格式） # ================================================================ async def _try_upload_image(self, group_id: str, image_url: str, platform_id: str): """ 尝试将图片报告上传到群文件和/或群相册（静默处理，失败仅日志提示）。 """ import base64 import re import tempfile from datetime import datetime enable_file = self.config_manager.get_enable_group_file_upload() enable_album = self.config_manager.get_enable_group_album_upload() if not enable_file and not enable_album: return adapter = self.bot_manager.get_adapter(platform_id) if not adapter or not hasattr(adapter, "upload_group_file_to_folder"): return # 1. 构造一个更友好的文件名 now = datetime.now() timestamp = now.strftime("%H%M") date_str = now.strftime("%Y-%m-%d") # 默认基础名和后缀 ext = ( ".jpg" if (".jpg" in image_url.lower() or ".jpeg" in image_url.lower()) else ".png" ) nice_filename = f"群分析报告_{group_id}_{date_str}_{timestamp}{ext}" try: # 尝试通过适配器获取群名称，使文件名更具辨识度 group_info = await adapter.get_group_info(group_id) if group_info and group_info.group_name: # 过滤非法文件名字符：\ / : * ? " < > | safe_name = re.sub(r'[\\/:*?"<>|]', "", group_info.group_name).strip() if safe_name: nice_filename = ( f"群分析报告_{safe_name}_{date_str}_{timestamp}{ext}" ) except Exception: pass # 2. 将内容准备为文件或数据 image_file = None created_temp = False MAX_PAYLOAD_SIZE = 20 * 1024 * 1024 # 20MB 限制 try: data = None if image_url.startswith("base64://"): base64_str = image_url[len("base64://") :] if len(base64_str) * 3 / 4 > MAX_PAYLOAD_SIZE: logger.warning("图片上传失败：Base64 负载过大") return data = base64.b64decode(base64_str) elif image_url.startswith("data:"): parts = image_url.split(",", 1) if len(parts) == 2: if len(parts[1]) * 3 / 4 > MAX_PAYLOAD_SIZE: logger.warning("图片上传失败：Data URI 负载过大") return data = base64.b64decode(parts[1]) elif os.path.isfile(image_url): image_file = os.path.abspath(image_url) if data and not image_file: # 使用 tempfile 生成唯一后缀，防止并发冲突 fd, image_file = tempfile.mkstemp(suffix=ext, prefix="group_report_") try: with os.fdopen(fd, "wb") as f: f.write(data) created_temp = True except Exception: os.close(fd) raise if not image_file: return # 3. 执行上传：群文件 if enable_file: try: folder_name = self.config_manager.get_group_file_folder() folder_id = None if folder_name: folder_id = await adapter.find_or_create_folder( # type: ignore[attr-defined] group_id, folder_name ) await adapter.upload_group_file_to_folder( # type: ignore[attr-defined] group_id=group_id, file_path=image_file, folder_id=folder_id, filename=nice_filename, # 显式传递漂亮的文件名 ) except Exception as e: logger.warning(f"群文件上传失败 (群 {group_id}): {e}") if enable_album and hasattr(adapter, "upload_group_album"): try: album_name = self.config_manager.get_group_album_name() strict_mode = self.config_manager.get_group_album_strict_mode() album_id = None if hasattr(adapter, "find_album_id"): if album_name: album_id = await adapter.find_album_id(group_id, album_name) # type: ignore[attr-defined] if not album_id and strict_mode: logger.info( f"群相册严格模式开启：在群 {group_id} 中未找到名为 '{album_name}' 的相册，停止上传。" ) return elif strict_mode: logger.info( f"群相册严格模式开启：未设置目标相册名称，停止上传以防止操作群 {group_id} 的默认相册。" ) return await adapter.upload_group_album( # type: ignore[attr-defined] group_id, image_file, album_id=album_id, album_name=album_name, strict_mode=strict_mode, ) except Exception as e: logger.warning(f"群相册上传失败 (群 {group_id}): {e}") except Exception as e: logger.warning(f"图片上传处理异常: {e}") finally: if created_temp and image_file and os.path.exists(image_file): try: os.remove(image_file) except OSError: pass @filter.command("群分析", alias={"group_analysis"}) @filter.permission_type(PermissionType.ADMIN) async def analyze_group_daily( self, event: AstrMessageEvent, days: int | None = None ): """ 分析群聊日常活动（跨平台支持） 用法: /群分析 [天数] """ if self._terminating: return current_task = asyncio.current_task() if current_task: self._background_tasks.add(current_task) try: event.should_call_llm(True) # 阻止默认 LLM 解析 group_id = self._get_group_id_from_event(event) platform_id = self._get_platform_id_from_event(event) if not group_id: yield event.plain_result("❌ 请在群聊中使用此命令") return # 更新bot实例 self.bot_manager.update_from_event(event) # 优先使用 UMO 进行权限检查 (兼容白名单 UMO 格式) check_target = getattr(event, "unified_msg_origin", None) if not check_target: check_target = f"{platform_id}:GroupMessage:{group_id}" if not self.config_manager.is_group_allowed(check_target): # Fallback checks (simple ID) are handled inside is_group_allowed logic if list item has no colon # But if list item HAS colon, we need precise match. # If prompt fails, try simple ID as fallback for permissive cases? # No, config_manager.is_group_allowed already handles simple ID matching if whitelist item is simple ID. yield event.plain_result("❌ 此群未启用日常分析功能") return # 获取群名以生成语义化的 TraceID group_name = "" try: adapter = self.bot_manager.get_adapter(platform_id) if adapter: info = await adapter.get_group_info(group_id) if info and info.group_name: group_name = info.group_name except Exception: pass # 设置 TraceID (语义化格式: manual_群名_HHmm) trace_id = TraceContext.generate( prefix="manual", group_name=group_name or group_id ) TraceContext.set(trace_id) # 表情回应 或 文本提示（二选一，由配置开关控制） adapter = self.bot_manager.get_adapter(platform_id) orig_msg_id = getattr(event.message_obj, "message_id", None) adapter_platform_name = ( (adapter.get_platform_name() if adapter else "").strip().lower() ) # QQ 官方机器人 API v2 不支持本插件使用的表情回应接口， # 因此始终沿用原有的文字进度提示，避免触发无效的 reaction 请求。 use_text_reply = ( adapter_platform_name in {"qq_official", "qq_official_webhook"} or self.config_manager.get_enable_analysis_reply() ) if use_text_reply: yield event.plain_result("🔍 正在启动分析引擎，正在拉取最近消息...") elif adapter and orig_msg_id: await adapter.set_reaction( event.get_group_id(), orig_msg_id, "analysis_started" ) # 调用 DDD 应用级服务 result = await self.analysis_service.execute_daily_analysis( group_id=group_id, platform_id=platform_id, manual=True, days=days ) if not result.get("success"): reason = result.get("reason") if reason == "no_messages": yield event.plain_result("❌ 未找到足够的群聊记录") elif reason == "muted": logger.warning( f"群 {group_id} 开启了全群禁言或对 Bot 禁言，跳过回复以防抛出发送异常" ) else: yield event.plain_result("❌ 分析失败，原因未知") return if not use_text_reply and adapter and orig_msg_id: await adapter.set_reaction( event.get_group_id(), orig_msg_id, "analysis_done" ) async for res in self._send_analysis_report(event, result): yield res except DuplicateGroupTaskError: yield event.plain_result("📊 该群的分析任务正在执行中，请稍后再试哦~") except asyncio.CancelledError: logger.info("群分析任务被取消 (插件重载或卸载)") except Exception as e: logger.error(f"群分析失败: {e}", exc_info=True) yield event.plain_result( f"❌ 分析失败: {str(e)}。请检查网络连接和LLM配置，或联系管理员" ) finally: if current_task: self._background_tasks.discard(current_task) async def _send_analysis_report( self, event: AstrMessageEvent, result: dict ) -> AsyncGenerator: """处理分析结果的渲染和发送""" if self._terminating or not self.config_manager: logger.warning("插件正在关闭，停止发送报告") return group_id = result["group_id"] platform_id = result["platform_id"] analysis_result = result["analysis_result"] adapter = result["adapter"] output_format = self.config_manager.get_output_format()[0] is_qq_official = adapter.get_platform_name() == "qq_official" # 定义获取回调 async def avatar_url_getter(user_id: str) -> str | None: return await adapter.get_user_avatar_url(user_id) async def nickname_getter(user_id: str) -> str | None: try: member = await adapter.get_member_info(group_id, user_id) if member: return member.card or member.nickname except Exception: pass return None if output_format == "image": image_url, html_content = await self.report_generator.generate_image_report( analysis_result, group_id, self.html_render, avatar_url_getter=avatar_url_getter, nickname_getter=nickname_getter, avatar_cache_namespace=platform_id, allow_alphanumeric_user_ids=is_qq_official, ) if image_url: caption = TraceContext.make_report_caption() sent = await adapter.send_image(group_id, image_url, caption=caption) if sent: await self._try_upload_image(group_id, image_url, platform_id) return # 成功发送 # 如果图片生成或发送失败，直接回退到文本 logger.warning(f"图片报告发送失败，正在发送文本回退报告。群: {group_id}") await self._send_text_reports( group_id, analysis_result, is_qq_official, adapter ) return elif output_format == "html": html_path, json_path = await self.report_generator.generate_html_report( analysis_result, group_id, avatar_url_getter=avatar_url_getter, nickname_getter=nickname_getter, avatar_cache_namespace=platform_id, allow_alphanumeric_user_ids=is_qq_official, ) if html_path: is_only_url = self.config_manager.get_html_only_url() base_url = self.config_manager.get_html_base_url() if is_only_url: if base_url and base_url.strip(): # 获取配置中的输出目录 html_output_dir = self.config_manager.get_html_output_dir() # 若用户配置为空，使用默认目录 if not html_output_dir: from astrbot.api.star import StarTools html_output_dir = os.path.join( StarTools.get_data_dir(PLUGIN_NAME), "self_hosted_html_reports", ) # 计算相对路径并转换为URL rel_path = os.path.relpath(html_path, html_output_dir) url_path = rel_path.replace(os.sep, "/") report_url = f"{base_url.rstrip('/')}/{url_path.lstrip('/')}" yield event.plain_result( f"📊 今日群聊分析报告已生成：\n{report_url}" ) return # 拦截成功，直接退出，不再发文件 else: logger.warning( f"手动触发群 {group_id} 开启了仅发送外链，但未配置 html_base_url，回退至发送文件。" ) caption = self.report_generator.build_html_caption(html_path) # 发送 HTML 文件 sender = getattr(self, "message_sender", None) if sender: sent = await sender.send_file( group_id, html_path, caption=caption, platform_id=platform_id, ) else: sent = await adapter.send_file(group_id, html_path) if sent and caption: await adapter.send_text(group_id, caption) if not sent: yield event.chain_result( [File(name=Path(html_path).name, file=html_path)] ) if caption: yield event.plain_result(caption) else: yield event.plain_result("⚠️ HTML 生成失败。") else: await self._send_text_reports( group_id, analysis_result, is_qq_official, adapter ) async def _generate_text_reports( self, analysis_result: dict, use_qq_official_markdown: bool ) -> tuple[str, str | None]: """Generate text or QQ-official-markdown reports.""" if use_qq_official_markdown: return await self.report_generator.generate_qq_official_markdown_report( analysis_result, self.html_render ) return self.report_generator.generate_text_report(analysis_result), None async def _send_text_reports( self, group_id: str, analysis_result: dict, use_qq_official_markdown: bool, adapter, ) -> bool: """Send text reports via platform adapter.""" tr, fr = await self._generate_text_reports( analysis_result, use_qq_official_markdown ) if use_qq_official_markdown: return await adapter.send_text_report(group_id, tr, fallback_content=fr) return await adapter.send_text_report(group_id, tr) @filter.command("设置格式", alias={"set_format"}) @filter.permission_type(PermissionType.ADMIN) async def set_output_format(self, event: AstrMessageEvent, format_input: str = ""): """ 设置分析报告输出格式（跨平台支持） 用法: /设置格式 [格式名称或序号] 或 image,html 等逗号分隔的组合 """ # 命令由插件处理，禁用默认 LLM 回退。 event.should_call_llm(True) available_formats = ["image", "text", "html"] format_display_names = { "image": "图片格式 (默认)", "text": "文本格式", "html": "交互式 HTML 网页", } if not format_input: current = ", ".join(self.config_manager.get_output_format()) format_list_str = "\n".join( [ f"【{i}】{f} - {format_display_names[f]}" for i, f in enumerate(available_formats, start=1) ] ) yield event.plain_result(f"""📊 当前输出格式: {current} 可用格式: {format_list_str} 用法: /设置格式 [名称或序号] 如 /设置格式 image,html""") return target_format = None # 尝试由序号选择 if format_input.isdigit(): idx = int(format_input) - 1 if 0 <= idx < len(available_formats): target_format = available_formats[idx] # 尝试按名称选择 if not target_format: input_lower = format_input.lower() if input_lower in available_formats: target_format = input_lower # 支持逗号分隔的多个格式 if not target_format: parts = [f.strip() for f in format_input.replace("，", ",").split(",")] if all(p in available_formats for p in parts) and len(parts) > 1: try: self.config_manager.set_output_format(parts) yield event.plain_result(f"✅ 输出格式已设置为: {', '.join(parts)}") except Exception as e: yield event.plain_result(f"❌ 设置失败: {e}") return if not target_format: yield event.plain_result( f"❌ 无效的格式类型 '{format_input}'。可用: {', '.join(available_formats)} 或序号 1-{len(available_formats)}" ) return try: self.config_manager.set_output_format(target_format) # type: ignore[arg-type] yield event.plain_result(f"✅ 输出格式已设置为: {target_format}") except Exception as e: yield event.plain_result(f"❌ 设置失败: {e}") @filter.command("设置模板", alias={"set_template"}) @filter.permission_type(PermissionType.ADMIN) async def set_report_template( self, event: AstrMessageEvent, template_input: str = "" ): """ 设置分析报告模板（跨平台支持） 用法: /设置模板 [模板名称或序号] """ # 命令由插件处理，禁用默认 LLM 回退。 event.should_call_llm(True) available_templates = ( await self.template_command_service.list_available_templates() ) if not template_input: current_template = self.config_manager.get_report_template() template_list_str = "\n".join( [f"【{i}】{t}" for i, t in enumerate(available_templates, start=1)] ) yield event.plain_result(f"""🎨 当前报告模板: {current_template} 可用模板: {template_list_str} 用法: /设置模板 [模板名称或序号] 💡 使用 /查看模板 查看预览图""") return template_name, parse_error = self.template_command_service.parse_template_input( template_input, available_templates ) if parse_error: yield event.plain_result(parse_error) return if not template_name: yield event.plain_result(f"❌ 无法解析模板输入: {template_input}") return if not await self.template_command_service.template_exists(template_name): yield event.plain_result(f"❌ 模板 '{template_name}' 不存在") return self.config_manager.set_report_template(template_name) yield event.plain_result(f"✅ 报告模板已设置为: {template_name}") @filter.command("查看模板", alias={"view_templates"}) @filter.permission_type(PermissionType.ADMIN) async def view_templates(self, event: AstrMessageEvent): """ 查看所有可用的报告模板及预览图（跨平台支持） 用法: /查看模板 """ # 命令由插件处理，禁用默认 LLM 回退。 event.should_call_llm(True) available_templates = ( await self.template_command_service.list_available_templates() ) if not available_templates: yield event.plain_result("❌ 未找到任何可用的报告模板") return platform_id = self._get_platform_id_from_event(event) await self.template_preview_router.ensure_handlers_registered(self.context) ( handled, handler_results, ) = await self.template_preview_router.handle_view_templates( event=event, platform_id=platform_id, available_templates=available_templates, ) if handled: for result in handler_results: yield result return current_template = self.config_manager.get_report_template() bot_id = event.get_self_id() preview_nodes = self.template_command_service.build_template_preview_nodes( available_templates=available_templates, current_template=current_template, bot_id=bot_id, ) yield event.chain_result([preview_nodes]) @filter.command("分析设置", alias={"analysis_settings"}) @filter.permission_type(PermissionType.ADMIN) async def analysis_settings(self, event: AstrMessageEvent, action: str = "status"): """ 管理分析设置（跨平台支持） 用法: /分析设置 [enable|disable|status|reload|test] - enable: 启用当前群的分析功能 - disable: 禁用当前群的分析功能 - status: 查看当前状态 - reload: 重新加载配置并重启定时任务 - test: 测试自动分析功能 - filter_bot: 切换是否在分析中包含机器人自己的消息 - incremental_debug: 切换增量分析立即报告模式（调试用） """ group_id = self._get_group_id_from_event(event) if not group_id: yield event.plain_result("❌ 请在群聊中使用此命令") return if action == "enable": async for result in self._handle_settings_enable(event, group_id): yield result elif action == "disable": async for result in self._handle_settings_disable(event, group_id): yield result elif action == "reload": self.auto_scheduler.schedule_jobs(self.context) yield event.plain_result("✅ 已重新加载配置并重启定时任务") elif action == "test": check_target = getattr(event, "unified_msg_origin", None) if not check_target: check_target = ( f"{self._get_platform_id_from_event(event)}:GroupMessage:{group_id}" ) if not self.config_manager.is_group_allowed(check_target): yield event.plain_result("❌ 请先启用当前群的分析功能") return yield event.plain_result("🧪 开始测试自动分析功能...") # 更新bot实例（用于测试） self.bot_manager.update_from_event(event) try: await self.auto_scheduler._perform_auto_analysis_for_group(group_id) yield event.plain_result("✅ 自动分析测试完成，请查看群消息") except DuplicateGroupTaskError: yield event.plain_result("📊 该群的分析任务正在执行中，请稍后再试哦~") except Exception as e: yield event.plain_result(f"❌ 自动分析测试失败: {str(e)}") elif action == "incremental_debug": current_state = self.config_manager.get_incremental_report_immediately() new_state = not current_state self.config_manager.set_incremental_report_immediately(new_state) status_text = "已启用" if new_state else "已禁用" yield event.plain_result(f"✅ 增量分析立即报告模式: {status_text}") elif action == "filter_bot": current = self.config_manager.get_filter_bot_messages() new_state = not current self.config_manager.set_filter_bot_messages(new_state) status_text = "已启用" if new_state else "已禁用" yield event.plain_result(f"✅ 过滤机器人消息: {status_text}") else: # status check_target = getattr(event, "unified_msg_origin", None) if not check_target: check_target = ( f"{self._get_platform_id_from_event(event)}:GroupMessage:{group_id}" ) is_allowed = self.config_manager.is_group_allowed(check_target) status = "已启用" if is_allowed else "未启用" mode = self.config_manager.get_group_list_mode() auto_status = ( "已启用" if self.config_manager.is_auto_analysis_enabled() else "未启用" ) auto_time = self.config_manager.get_auto_analysis_time() output_format = self.config_manager.get_output_format()[0] min_threshold = self.config_manager.get_min_messages_threshold() # 增量分析状态 incremental_enabled = self.config_manager.get_incremental_enabled() incremental_status_text = "未启用" if incremental_enabled: interval = self.config_manager.get_incremental_interval_minutes() max_daily = self.config_manager.get_incremental_max_daily_analyses() active_start = self.config_manager.get_incremental_active_start_hour() active_end = self.config_manager.get_incremental_active_end_hour() incremental_status_text = ( f"已启用 (间隔{interval}分钟, 最多{max_daily}次/天, " f"活跃时段{active_start}:00-{active_end}:00)" ) debug_report = self.config_manager.get_incremental_report_immediately() debug_status = "✅ 开启" if debug_report else "❌ 关闭" filter_bot = self.config_manager.get_filter_bot_messages() filter_bot_status = "✅ 开启" if filter_bot else "❌ 关闭" yield event.plain_result(f"""📊 当前群分析功能状态: • 群分析功能: {status} (模式: {mode}) • 自动分析: {auto_status} ({auto_time}) • 增量分析: {incremental_status_text} • 调试模式: {debug_status} (增量立即报告) • 过滤机器人: {filter_bot_status} • 输出格式: {output_format} • 最小消息数: {min_threshold} 💡 可用命令: enable, disable, status, reload, test, filter_bot, incremental_debug 💡 支持的输出格式: image, text (图片包含活跃度可视化) 💡 其他命令: /设置格式, /增量状态""") @filter.command("增量状态", alias={"incremental_status"}) @filter.permission_type(PermissionType.ADMIN) async def incremental_status(self, event: AstrMessageEvent): """查看当前增量分析状态（滑动窗口）""" group_id = self._get_group_id_from_event(event) if not group_id: yield event.plain_result("❌ 请在群聊中使用此命令") return if not self.config_manager.get_incremental_enabled(): yield event.plain_result("ℹ️ 增量分析模式未启用，请在插件配置中开启") return import time as time_mod # 计算滑动窗口范围 analysis_days = self.config_manager.get_analysis_days() window_end = time_mod.time() window_start = window_end - (analysis_days * 24 * 3600) # 查询窗口内的批次 batches = await self.incremental_store.query_batches( group_id, window_start, window_end ) if not batches: from datetime import datetime start_str = datetime.fromtimestamp(window_start).strftime("%m-%d %H:%M") end_str = datetime.fromtimestamp(window_end).strftime("%m-%d %H:%M") yield event.plain_result( f"📊 滑动窗口 ({start_str} ~ {end_str}) 内尚无增量分析数据" ) return # 合并批次获取聚合视图 state = self.incremental_merge_service.merge_batches( batches, window_start, window_end ) summary = state.get_summary() yield event.plain_result( f"📊 增量分析状态 (窗口: {summary['window']})\n" f"• 分析次数: {summary['total_analyses']}\n" f"• 累计消息: {summary['total_messages']}\n" f"• 话题数: {summary['topics_count']}\n" f"• 金句数: {summary['quotes_count']}\n" f"• 参与者: {summary['participants']}\n" f"• 高峰时段: {summary['peak_hours']}" ) async def _handle_settings_enable(self, event: AstrMessageEvent, group_id: str): """协助逻辑：处理启用设置的分支逻辑""" mode = self.config_manager.get_group_list_mode() target_id = event.unified_msg_origin or group_id if mode == "whitelist": glist = self.config_manager.get_group_list() if not self.config_manager.is_group_allowed(target_id): glist.append(target_id) self.config_manager.set_group_list(glist) yield event.plain_result(f"✅ 已将当前群加入白名单\nID: {target_id}") self.auto_scheduler.schedule_jobs(self.context) else: yield event.plain_result("ℹ️ 当前群已在白名单中") elif mode == "blacklist": glist = self.config_manager.get_group_list() removed = False if target_id in glist: glist.remove(target_id) removed = True if group_id in glist: glist.remove(group_id) removed = True if removed: self.config_manager.set_group_list(glist) yield event.plain_result("✅ 已将当前群从黑名单移除") self.auto_scheduler.schedule_jobs(self.context) else: yield event.plain_result("ℹ️ 当前群不在黑名单中") else: yield event.plain_result("ℹ️ 当前为无限制模式，所有群聊默认启用") async def _handle_settings_disable(self, event: AstrMessageEvent, group_id: str): """协助逻辑：处理禁用设置的分支逻辑""" mode = self.config_manager.get_group_list_mode() target_id = event.unified_msg_origin or group_id if mode == "whitelist": glist = self.config_manager.get_group_list() removed = False if target_id in glist: glist.remove(target_id) removed = True if group_id in glist: glist.remove(group_id) removed = True if removed: self.config_manager.set_group_list(glist) yield event.plain_result("✅ 已将当前群从白名单移除") self.auto_scheduler.schedule_jobs(self.context) else: yield event.plain_result("ℹ️ 当前群不在白名单中") elif mode == "blacklist": glist = self.config_manager.get_group_list() if self.config_manager.is_group_allowed(target_id): glist.append(target_id) self.config_manager.set_group_list(glist) yield event.plain_result(f"✅ 已将当前群加入黑名单\nID: {target_id}") self.auto_scheduler.schedule_jobs(self.context) else: yield event.plain_result("ℹ️ 当前群已在黑名单中") else: yield event.plain_result("ℹ️ 当前为无限制模式，如需禁用请切换到黑名单模式")