astrbot_plugin_qq_group_daily_analysis/main.py

"""
群日常分析插件
基于群聊记录生成精美的日常分析报告，包含话题总结、用户画像、统计数据等

重构版本 - 使用模块化架构，支持跨平台
"""

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() if self.config_manager.get_show_report_caption() else ""
                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("ℹ️ 当前为无限制模式，如需禁用请切换到黑名单模式")