feat: 模板预览路由化并增强 Telegram 按钮交互稳定性 (#83)

* feat: 路由化模板预览并完善Telegram交互回退

* fix: 忽略模板预览文档回退的重复编辑异常

* fix: 在回调阶段主动清理并校验模板预览会话过期

* fix: 处理Telegram回调注册在平台热替换场景的重绑与清理

* fix(set_report_template): 删除冗余的检查

---------

Co-authored-by: SXP-Simon <sxp20061207@163.com>

main.py

@@ -14,6 +14,9 @@ from astrbot.api.event.filter import PermissionType
 from astrbot.api.star import Context, Star
 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,
 )
@@ -31,6 +34,10 @@ from .src.infrastructure.persistence.telegram_group_registry import (
     TelegramGroupRegistry,
 )
 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.scheduler.retry import RetryManager
@@ -83,6 +90,16 @@ class QQGroupDailyAnalysis(Star):
         self.message_processing_service = MessageProcessingService(
             context, self.telegram_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.retry_manager = RetryManager(
@@ -102,32 +119,6 @@ class QQGroupDailyAnalysis(Star):
         # 异步注册任务，处理插件重载情况
         asyncio.create_task(self._run_initialization("Plugin Reload/Init"))
 
-    def _resolve_template_base_dir(self) -> str:
-        """解析报告模板目录（兼容新旧目录结构）"""
-        plugin_root = os.path.dirname(__file__)
-        candidate_dirs = [
-            os.path.join(
-                plugin_root, "src", "infrastructure", "reporting", "templates"
-            ),
-            os.path.join(plugin_root, "src", "reports", "templates"),
-        ]
-        for candidate in candidate_dirs:
-            if os.path.isdir(candidate):
-                return candidate
-        return candidate_dirs[0]
-
-    def _resolve_template_preview_path(self, template_name: str) -> str | None:
-        """解析模板预览图路径（兼容新旧命名和目录）"""
-        plugin_root = os.path.dirname(__file__)
-
-        candidate_paths = [
-            os.path.join(plugin_root, "assets", f"{template_name}-demo.jpg"),
-        ]
-        for candidate in candidate_paths:
-            if os.path.exists(candidate):
-                return candidate
-        return None
-
     # orchestrators 缓存已移至 应用层逻辑 (分析服务) 或 暂时移除以简化。
     # 如果需要高性能缓存，后续可由 AnalysisApplicationService 内部维护。
 
@@ -164,6 +155,9 @@ class QQGroupDailyAnalysis(Star):
             discovered = await self.bot_manager.initialize_from_config()
             if discovered:
                 logger.info("Bot管理器初始化成功")
+                await self.template_preview_router.ensure_handlers_registered(
+                    self.context
+                )
                 # 启动调度器
                 self.auto_scheduler.schedule_jobs(self.context)
             else:
@@ -191,6 +185,8 @@ class QQGroupDailyAnalysis(Star):
 
             if self.retry_manager:
                 await self.retry_manager.stop()
+            if self.template_preview_router:
+                await self.template_preview_router.unregister_handlers()
 
             # 重置实例属性
             self.auto_scheduler = None
@@ -199,6 +195,8 @@ class QQGroupDailyAnalysis(Star):
             self.config_manager = None
             self.message_processing_service = None
             self.telegram_group_registry = None
+            self.template_preview_router = None
+            self.telegram_template_preview_handler = None
 
             logger.info("QQ群日常分析插件资源清理完成")
 
@@ -431,22 +429,12 @@ class QQGroupDailyAnalysis(Star):
         设置分析报告模板（跨平台支持）
         用法: /设置模板 [模板名称或序号]
         """
-        # 获取模板目录和可用模板列表
-        template_base_dir = self._resolve_template_base_dir()
+        # 命令由插件处理，禁用默认 LLM 回退。
+        event.should_call_llm(True)
 
-        def _list_templates_sync():
-            if os.path.exists(template_base_dir):
-                return sorted(
-                    [
-                        d
-                        for d in os.listdir(template_base_dir)
-                        if os.path.isdir(os.path.join(template_base_dir, d))
-                        and not d.startswith("__")
-                    ]
-                )
-            return []
-
-        available_templates = await asyncio.to_thread(_list_templates_sync)
+        available_templates = (
+            await self.template_command_service.list_available_templates()
+        )
 
         if not template_input:
             current_template = self.config_manager.get_report_template()
@@ -462,22 +450,14 @@ class QQGroupDailyAnalysis(Star):
 💡 使用 /查看模板 查看预览图""")
             return
 
-        # 判断输入是序号还是模板名称
-        template_name = template_input
-        if template_input.isdigit():
-            index = int(template_input)
-            if 1 <= index <= len(available_templates):
-                template_name = available_templates[index - 1]
-            else:
-                yield event.plain_result(
-                    f"❌ 无效的序号 '{template_input}'，有效范围: 1-{len(available_templates)}"
-                )
-                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
 
-        # 检查模板是否存在
-        template_dir = os.path.join(template_base_dir, template_name)
-        template_exists = await asyncio.to_thread(os.path.exists, template_dir)
-        if not template_exists:
+        if not await self.template_command_service.template_exists(template_name):
             yield event.plain_result(f"❌ 模板 '{template_name}' 不存在")
             return
 
@@ -491,69 +471,40 @@ class QQGroupDailyAnalysis(Star):
         查看所有可用的报告模板及预览图（跨平台支持）
         用法: /查看模板
         """
-        from astrbot.api.message_components import Image, Node, Nodes, Plain
+        # 命令由插件处理，禁用默认 LLM 回退。
+        event.should_call_llm(True)
 
-        # 获取模板目录
-        template_dir = self._resolve_template_base_dir()
-
-        def _list_templates_sync():
-            if os.path.exists(template_dir):
-                return sorted(
-                    [
-                        d
-                        for d in os.listdir(template_dir)
-                        if os.path.isdir(os.path.join(template_dir, d))
-                        and not d.startswith("__")
-                    ]
-                )
-            return []
-
-        available_templates = await asyncio.to_thread(_list_templates_sync)
+        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()
-        bot_name = "模板预览"
-
-        # 圆圈数字序号
-        circle_numbers = ["①", "②", "③", "④", "⑤", "⑥", "⑦", "⑧", "⑨", "⑩"]
-
-        # 构建合并转发消息节点列表
-        node_list = []
-
-        # 添加标题节点
-        header_content = [
-            Plain(
-                f"🎨 可用报告模板列表\n📌 当前使用: {current_template}\n💡 使用 /设置模板 [序号] 切换"
-            )
-        ]
-        node_list.append(Node(uin=bot_id, name=bot_name, content=header_content))
-
-        # 为每个模板创建一个节点
-        for index, template_name in enumerate(available_templates):
-            current_mark = " ✅" if template_name == current_template else ""
-            num_label = (
-                circle_numbers[index]
-                if index < len(circle_numbers)
-                else f"({index + 1})"
-            )
-
-            node_content = [Plain(f"{num_label} {template_name}{current_mark}")]
-
-            # 添加预览图
-            preview_image_path = self._resolve_template_preview_path(template_name)
-            if preview_image_path:
-                node_content.append(Image.fromFileSystem(preview_image_path))
-
-            node_list.append(Node(uin=bot_id, name=template_name, content=node_content))
-
-        # 使用 Nodes 包装成一个合并转发消息
-        yield event.chain_result([Nodes(node_list)])
+        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("安装PDF", alias={"install_pdf"})
     @filter.permission_type(PermissionType.ADMIN)

src/application/commands/__init__.py

@@ -0,0 +1,5 @@
+"""命令相关应用服务。"""
+
+from .template_command_service import TemplateCommandService
+
+__all__ = ["TemplateCommandService"]

src/application/commands/template_command_service.py

@@ -0,0 +1,114 @@
+"""模板管理相关命令服务。"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+
+from astrbot.api.message_components import Image, Node, Nodes, Plain
+
+
+class TemplateCommandService:
+    """封装模板命令的文件系统与消息构建逻辑。"""
+
+    _CIRCLE_NUMBERS = ["①", "②", "③", "④", "⑤", "⑥", "⑦", "⑧", "⑨", "⑩"]
+
+    def __init__(self, plugin_root: str):
+        self.plugin_root = plugin_root
+
+    def resolve_template_base_dir(self) -> str:
+        """解析报告模板目录（兼容新旧目录结构）。"""
+        candidate_dirs = [
+            os.path.join(
+                self.plugin_root, "src", "infrastructure", "reporting", "templates"
+            ),
+            os.path.join(self.plugin_root, "src", "reports", "templates"),
+        ]
+        for candidate in candidate_dirs:
+            if os.path.isdir(candidate):
+                return candidate
+        return candidate_dirs[0]
+
+    def resolve_template_preview_path(self, template_name: str) -> str | None:
+        """解析模板预览图路径。"""
+        candidate_paths = [
+            os.path.join(self.plugin_root, "assets", f"{template_name}-demo.jpg"),
+        ]
+        for candidate in candidate_paths:
+            if os.path.exists(candidate):
+                return candidate
+        return None
+
+    async def list_available_templates(self) -> list[str]:
+        """列出所有可用模板。"""
+        template_base_dir = self.resolve_template_base_dir()
+
+        def _list_templates_sync() -> list[str]:
+            if os.path.exists(template_base_dir):
+                return sorted(
+                    [
+                        d
+                        for d in os.listdir(template_base_dir)
+                        if os.path.isdir(os.path.join(template_base_dir, d))
+                        and not d.startswith("__")
+                    ]
+                )
+            return []
+
+        return await asyncio.to_thread(_list_templates_sync)
+
+    async def template_exists(self, template_name: str) -> bool:
+        """检查模板目录是否存在。"""
+        template_dir = os.path.join(self.resolve_template_base_dir(), template_name)
+        return await asyncio.to_thread(os.path.exists, template_dir)
+
+    def parse_template_input(
+        self, template_input: str, available_templates: list[str]
+    ) -> tuple[str | None, str | None]:
+        """解析模板输入（支持模板名或序号）。"""
+        if not template_input:
+            return None, "❌ 模板参数不能为空"
+
+        if template_input.isdigit():
+            index = int(template_input)
+            if 1 <= index <= len(available_templates):
+                return available_templates[index - 1], None
+            return (
+                None,
+                f"❌ 无效的序号 '{template_input}'，有效范围: 1-{len(available_templates)}",
+            )
+
+        return template_input, None
+
+    def build_template_preview_nodes(
+        self,
+        available_templates: list[str],
+        current_template: str,
+        bot_id: str,
+    ) -> Nodes:
+        """构建模板预览的合并消息节点。"""
+        node_list = []
+
+        header_content = [
+            Plain(
+                f"🎨 可用报告模板列表\n📌 当前使用: {current_template}\n💡 使用 /设置模板 [序号] 切换"
+            )
+        ]
+        node_list.append(Node(uin=bot_id, name="模板预览", content=header_content))
+
+        for index, template_name in enumerate(available_templates):
+            current_mark = " ✅" if template_name == current_template else ""
+            num_label = (
+                self._CIRCLE_NUMBERS[index]
+                if index < len(self._CIRCLE_NUMBERS)
+                else f"({index + 1})"
+            )
+
+            node_content = [Plain(f"{num_label} {template_name}{current_mark}")]
+            preview_image_path = self.resolve_template_preview_path(template_name)
+            if preview_image_path:
+                node_content.append(Image.fromFileSystem(preview_image_path))
+
+            node_list.append(Node(uin=bot_id, name=template_name, content=node_content))
+
+        return Nodes(node_list)

src/infrastructure/platform/template_preview/__init__.py

@@ -0,0 +1,6 @@
+"""平台模板预览交互能力。"""
+
+from .router import TemplatePreviewRouter
+from .telegram_preview_handler import TelegramTemplatePreviewHandler
+
+__all__ = ["TelegramTemplatePreviewHandler", "TemplatePreviewRouter"]

src/infrastructure/platform/template_preview/router.py

@@ -0,0 +1,63 @@
+"""模板预览平台路由。"""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+class TemplatePreviewRouter:
+    """统一分发不同平台的模板预览处理器。"""
+
+    def __init__(self, handlers: list[Any] | None = None):
+        self._handlers: list[Any] = handlers or []
+
+    def add_handler(self, handler: Any) -> None:
+        """注册一个平台处理器。"""
+        self._handlers.append(handler)
+
+    async def ensure_handlers_registered(self, context: Any) -> None:
+        """让处理器完成初始化（如注册回调）。"""
+        for handler in self._handlers:
+            register_func = getattr(
+                handler, "ensure_callback_handlers_registered", None
+            )
+            if callable(register_func):
+                await register_func(context)
+
+    async def unregister_handlers(self) -> None:
+        """统一注销处理器资源。"""
+        for handler in self._handlers:
+            unregister_func = getattr(handler, "unregister_callback_handlers", None)
+            if callable(unregister_func):
+                await unregister_func()
+
+    async def handle_view_templates(
+        self,
+        event: Any,
+        platform_id: str,
+        available_templates: list[str],
+    ) -> tuple[bool, list[Any]]:
+        """
+        处理 /查看模板 交互。
+
+        返回:
+        - handled: 是否已由某个平台处理器接管
+        - results: 需要回传给框架的消息结果列表
+        """
+        for handler in self._handlers:
+            supports_func = getattr(handler, "supports", None)
+            if not callable(supports_func) or not supports_func(event):
+                continue
+
+            handle_func = getattr(handler, "handle_view_templates", None)
+            if not callable(handle_func):
+                continue
+
+            handled, results = await handle_func(
+                event=event,
+                platform_id=platform_id,
+                available_templates=available_templates,
+            )
+            return handled, results
+
+        return False, []

src/infrastructure/platform/template_preview/telegram_preview_handler.py

@@ -0,0 +1,676 @@
+"""Telegram 模板预览交互处理。"""
+
+from __future__ import annotations
+
+import re
+import time
+import uuid
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any
+
+from ....utils.logger import logger
+
+if TYPE_CHECKING:
+    from astrbot.api.event import AstrMessageEvent
+
+    from ....application.commands.template_command_service import TemplateCommandService
+    from ...config.config_manager import ConfigManager
+
+try:
+    from telegram import (
+        InlineKeyboardButton,
+        InlineKeyboardMarkup,
+        InputMediaDocument,
+        InputMediaPhoto,
+        Update,
+    )
+    from telegram.error import BadRequest
+    from telegram.ext import CallbackQueryHandler, ContextTypes
+
+    TELEGRAM_RUNTIME_AVAILABLE = True
+except Exception:
+    TELEGRAM_RUNTIME_AVAILABLE = False
+    InlineKeyboardButton = None
+    InlineKeyboardMarkup = None
+    InputMediaPhoto = None
+    InputMediaDocument = None
+    Update = None
+    BadRequest = Exception
+    CallbackQueryHandler = None
+    ContextTypes = None
+
+
+@dataclass
+class _PreviewSession:
+    token: str
+    platform_id: str
+    chat_id: int | str
+    message_thread_id: int | None
+    message_id: int
+    requester_id: int
+    templates: list[str]
+    index: int
+    created_at: float
+
+    @property
+    def current_template(self) -> str:
+        return self.templates[self.index]
+
+
+class TelegramTemplatePreviewHandler:
+    """Telegram 按钮预览处理器（←/确定/→）。"""
+
+    _SESSION_TTL_SECONDS = 2 * 60 * 60
+    _MAX_SESSIONS = 200
+    _CONNECT_TIMEOUT = 20
+    _READ_TIMEOUT = 120
+    _WRITE_TIMEOUT = 120
+    _POOL_TIMEOUT = 20
+
+    def __init__(
+        self,
+        config_manager: ConfigManager,
+        template_service: TemplateCommandService,
+    ):
+        self.config_manager = config_manager
+        self.template_service = template_service
+        self._sessions: dict[str, _PreviewSession] = {}
+        self._registered_platform_ids: set[str] = set()
+        self._handlers: dict[str, tuple[Any, Any]] = {}
+        self._platform_clients: dict[str, Any] = {}
+        self._callback_prefix = f"qda_tpl_{uuid.uuid4().hex[:8]}"
+
+    @staticmethod
+    def supports(event: AstrMessageEvent) -> bool:
+        """判断是否 Telegram 事件。"""
+        try:
+            return (event.get_platform_name() or "").lower() == "telegram"
+        except Exception:
+            return False
+
+    # 向后兼容旧调用名
+    is_telegram_event = supports
+
+    async def ensure_callback_handlers_registered(self, context: Any) -> None:
+        """为所有 Telegram 平台注册按钮回调处理器。"""
+        if not TELEGRAM_RUNTIME_AVAILABLE:
+            return
+        if not context or not hasattr(context, "platform_manager"):
+            return
+
+        platforms = context.platform_manager.get_insts()
+        seen_platform_ids: set[str] = set()
+        for platform in platforms:
+            platform_id, platform_name = self._extract_platform_meta(platform)
+            if platform_name != "telegram":
+                continue
+            if not platform_id:
+                continue
+
+            seen_platform_ids.add(platform_id)
+            client = self._extract_platform_client(platform)
+            if client is not None:
+                self._platform_clients[platform_id] = client
+
+            application = getattr(platform, "application", None)
+            if not application:
+                continue
+
+            existing = self._handlers.get(platform_id)
+            if existing:
+                old_application, old_handler = existing
+                if old_application is application:
+                    self._registered_platform_ids.add(platform_id)
+                    continue
+
+                # 平台对象热替换：解绑旧 application 上的 handler 后重绑
+                try:
+                    old_application.remove_handler(old_handler)
+                    logger.info(
+                        f"[TemplatePreview][Telegram] 检测到 application 变更，已解绑旧回调: platform_id={platform_id}"
+                    )
+                except Exception as e:
+                    logger.debug(
+                        f"[TemplatePreview][Telegram] 解绑旧回调失败: platform_id={platform_id}, err={e}"
+                    )
+                self._handlers.pop(platform_id, None)
+                self._registered_platform_ids.discard(platform_id)
+
+            try:
+                handler = CallbackQueryHandler(
+                    self._on_callback_query,
+                    pattern=rf"^{re.escape(self._callback_prefix)}:",
+                )
+                application.add_handler(handler)
+                self._registered_platform_ids.add(platform_id)
+                self._handlers[platform_id] = (application, handler)
+                logger.info(
+                    f"[TemplatePreview][Telegram] 已注册回调处理器: platform_id={platform_id}"
+                )
+            except Exception as e:
+                logger.warning(
+                    f"[TemplatePreview][Telegram] 注册回调处理器失败: platform_id={platform_id}, err={e}"
+                )
+
+        # 兜底清理：平台下线后移除残留 handler，避免资源泄漏
+        stale_ids = [
+            platform_id
+            for platform_id in list(self._handlers.keys())
+            if platform_id not in seen_platform_ids
+        ]
+        for stale_platform_id in stale_ids:
+            old_application, old_handler = self._handlers.pop(stale_platform_id)
+            try:
+                old_application.remove_handler(old_handler)
+                logger.info(
+                    f"[TemplatePreview][Telegram] 已清理离线平台回调: platform_id={stale_platform_id}"
+                )
+            except Exception as e:
+                logger.debug(
+                    f"[TemplatePreview][Telegram] 清理离线平台回调失败: platform_id={stale_platform_id}, err={e}"
+                )
+            self._registered_platform_ids.discard(stale_platform_id)
+            self._platform_clients.pop(stale_platform_id, None)
+
+    async def unregister_callback_handlers(self) -> None:
+        """卸载已注册的回调处理器（插件终止时调用）。"""
+        if not TELEGRAM_RUNTIME_AVAILABLE:
+            return
+
+        for platform_id, (application, handler) in list(self._handlers.items()):
+            try:
+                application.remove_handler(handler)
+                logger.info(
+                    f"[TemplatePreview][Telegram] 已移除回调处理器: platform_id={platform_id}"
+                )
+            except Exception as e:
+                logger.debug(
+                    f"[TemplatePreview][Telegram] 移除回调处理器失败: platform_id={platform_id}, err={e}"
+                )
+        self._handlers.clear()
+        self._registered_platform_ids.clear()
+        self._platform_clients.clear()
+
+    async def send_preview_message(
+        self,
+        event: AstrMessageEvent,
+        platform_id: str,
+        available_templates: list[str],
+    ) -> bool:
+        """
+        在 Telegram 中发送可交互模板预览消息。
+
+        返回：
+        - True: 已由本处理器发送消息（调用方不应再走默认回复）
+        - False: 无法处理，调用方应走原有降级路径
+        """
+        if not TELEGRAM_RUNTIME_AVAILABLE:
+            return False
+        if not available_templates:
+            return False
+
+        client = self._get_event_client(event, platform_id)
+        if client is None:
+            logger.warning("[TemplatePreview][Telegram] 无法获取 Telegram client")
+            return False
+
+        target = self._resolve_chat_target(event)
+        if target is None:
+            return False
+        chat_id, message_thread_id = target
+
+        try:
+            requester_id = int(str(event.get_sender_id()))
+        except Exception:
+            logger.warning(
+                "[TemplatePreview][Telegram] sender_id 非法，无法创建交互会话"
+            )
+            return False
+
+        current_template = self.config_manager.get_report_template()
+        if current_template in available_templates:
+            index = available_templates.index(current_template)
+        else:
+            index = 0
+
+        token = uuid.uuid4().hex[:8]
+        keyboard = self._build_keyboard(token)
+        caption = self._build_caption(
+            template_name=available_templates[index],
+            index=index,
+            total=len(available_templates),
+        )
+
+        image_path = self.template_service.resolve_template_preview_path(
+            available_templates[index]
+        )
+        if not image_path:
+            return False
+
+        payload: dict[str, Any] = {"chat_id": chat_id, "reply_markup": keyboard}
+        if message_thread_id is not None:
+            payload["message_thread_id"] = message_thread_id
+
+        try:
+            with open(image_path, "rb") as image_file:
+                sent_msg = await client.send_photo(
+                    photo=image_file,
+                    caption=caption,
+                    connect_timeout=self._CONNECT_TIMEOUT,
+                    read_timeout=self._READ_TIMEOUT,
+                    write_timeout=self._WRITE_TIMEOUT,
+                    pool_timeout=self._POOL_TIMEOUT,
+                    **payload,
+                )
+        except BadRequest as e:
+            if not self._is_photo_dimension_error(e):
+                raise
+            with open(image_path, "rb") as image_file:
+                sent_msg = await client.send_document(
+                    document=image_file,
+                    caption=caption,
+                    connect_timeout=self._CONNECT_TIMEOUT,
+                    read_timeout=self._READ_TIMEOUT,
+                    write_timeout=self._WRITE_TIMEOUT,
+                    pool_timeout=self._POOL_TIMEOUT,
+                    **payload,
+                )
+
+        self._sessions[token] = _PreviewSession(
+            token=token,
+            platform_id=platform_id,
+            chat_id=chat_id,
+            message_thread_id=message_thread_id,
+            message_id=sent_msg.message_id,
+            requester_id=requester_id,
+            templates=available_templates.copy(),
+            index=index,
+            created_at=time.time(),
+        )
+        self._cleanup_expired_sessions()
+        logger.info(
+            "[TemplatePreview][Telegram] 已发送交互预览: "
+            f"platform_id={platform_id} chat_id={chat_id} token={token} templates={len(available_templates)}"
+        )
+        return True
+
+    async def send_preview_image_fallback(
+        self,
+        event: AstrMessageEvent,
+        platform_id: str,
+        template_name: str,
+    ) -> bool:
+        """TG 回退路径：直接发送单张预览图（不经过 event.image_result）。"""
+        if not TELEGRAM_RUNTIME_AVAILABLE:
+            return False
+
+        image_path = self.template_service.resolve_template_preview_path(template_name)
+        if not image_path:
+            return False
+
+        client = self._get_event_client(event, platform_id)
+        if client is None:
+            logger.warning("[TemplatePreview][Telegram] 回退发图失败：无法获取 client")
+            return False
+
+        target = self._resolve_chat_target(event)
+        if target is None:
+            return False
+        chat_id, message_thread_id = target
+
+        payload: dict[str, Any] = {
+            "chat_id": chat_id,
+            "caption": f"🖼 当前模板预览: {template_name}",
+            "connect_timeout": self._CONNECT_TIMEOUT,
+            "read_timeout": self._READ_TIMEOUT,
+            "write_timeout": self._WRITE_TIMEOUT,
+            "pool_timeout": self._POOL_TIMEOUT,
+        }
+        if message_thread_id is not None:
+            payload["message_thread_id"] = message_thread_id
+
+        try:
+            with open(image_path, "rb") as image_file:
+                await client.send_photo(photo=image_file, **payload)
+        except BadRequest as e:
+            if not self._is_photo_dimension_error(e):
+                raise
+            with open(image_path, "rb") as image_file:
+                await client.send_document(document=image_file, **payload)
+        return True
+
+    async def handle_view_templates(
+        self,
+        event: AstrMessageEvent,
+        platform_id: str,
+        available_templates: list[str],
+    ) -> tuple[bool, list[Any]]:
+        """统一处理 Telegram 的 /查看模板 流程。"""
+        if not self.supports(event):
+            return False, []
+
+        results: list[Any] = []
+
+        async def _append_fallback_results() -> None:
+            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)]
+            )
+            results.append(
+                event.plain_result(
+                    f"""🎨 可用报告模板列表
+📌 当前使用: {current_template}
+
+{template_list_str}
+
+💡 使用 /设置模板 [序号] 切换"""
+                )
+            )
+
+            try:
+                sent_preview = await self.send_preview_image_fallback(
+                    event=event,
+                    platform_id=platform_id,
+                    template_name=current_template,
+                )
+                if not sent_preview:
+                    results.append(event.plain_result("⚠️ 当前模板预览图发送失败"))
+            except Exception as image_err:
+                logger.warning(f"[TemplatePreview][Telegram] 回退发图失败: {image_err}")
+                results.append(event.plain_result("⚠️ 当前模板预览图发送失败"))
+
+        try:
+            sent = await self.send_preview_message(
+                event=event,
+                platform_id=platform_id,
+                available_templates=available_templates,
+            )
+            if sent:
+                return True, results
+            await _append_fallback_results()
+            return True, results
+        except Exception as e:
+            logger.warning(
+                f"[TemplatePreview][Telegram] 交互预览发送失败，回退普通模式: {e}"
+            )
+            await _append_fallback_results()
+            return True, results
+
+    async def _on_callback_query(
+        self, update: Update, context: ContextTypes.DEFAULT_TYPE
+    ) -> None:
+        if not TELEGRAM_RUNTIME_AVAILABLE:
+            return
+        if not update.callback_query or not update.callback_query.data:
+            return
+
+        self._cleanup_expired_sessions()
+
+        query = update.callback_query
+        data = query.data
+        parts = data.split(":")
+        if len(parts) != 3:
+            await query.answer("无效操作", show_alert=False)
+            return
+
+        _, token, action = parts
+        session = self._sessions.get(token)
+        if not session:
+            await query.answer("预览会话已过期，请重新发送 /查看模板", show_alert=True)
+            return
+        if time.time() - session.created_at > self._SESSION_TTL_SECONDS:
+            self._sessions.pop(token, None)
+            await query.answer("预览会话已过期，请重新发送 /查看模板", show_alert=True)
+            return
+
+        if not query.from_user:
+            await query.answer("无法识别操作者", show_alert=False)
+            return
+        if int(query.from_user.id) != session.requester_id:
+            await query.answer("仅命令发起人可操作该预览", show_alert=True)
+            return
+
+        if not query.message:
+            await query.answer("消息已失效", show_alert=False)
+            return
+
+        if query.message.message_id != session.message_id or str(
+            query.message.chat_id
+        ) != str(session.chat_id):
+            await query.answer("预览状态不一致，请重新发送 /查看模板", show_alert=True)
+            return
+
+        if action == "prev":
+            session.index = (session.index - 1) % len(session.templates)
+            await self._edit_preview_message(query, session)
+            await query.answer()
+            return
+
+        if action == "next":
+            session.index = (session.index + 1) % len(session.templates)
+            await self._edit_preview_message(query, session)
+            await query.answer()
+            return
+
+        if action == "apply":
+            template_name = session.current_template
+            self.config_manager.set_report_template(template_name)
+            await self._edit_preview_message(query, session, applied=True)
+            await query.answer(f"已设置模板: {template_name}", show_alert=False)
+            logger.info(
+                "[TemplatePreview][Telegram] 已应用模板: "
+                f"platform_id={session.platform_id} template={template_name} requester={session.requester_id}"
+            )
+            return
+
+        await query.answer("未知操作", show_alert=False)
+
+    async def _edit_preview_message(
+        self, query: Any, session: _PreviewSession, applied: bool = False
+    ) -> None:
+        template_name = session.current_template
+        caption = self._build_caption(
+            template_name=template_name,
+            index=session.index,
+            total=len(session.templates),
+            applied=applied,
+        )
+        keyboard = self._build_keyboard(session.token)
+        image_path = self.template_service.resolve_template_preview_path(template_name)
+        if not image_path:
+            await query.edit_message_caption(
+                caption=caption,
+                reply_markup=keyboard,
+            )
+            return
+
+        try:
+            with open(image_path, "rb") as image_file:
+                media = InputMediaPhoto(media=image_file, caption=caption)
+                await query.edit_message_media(
+                    media=media,
+                    reply_markup=keyboard,
+                    connect_timeout=self._CONNECT_TIMEOUT,
+                    read_timeout=self._READ_TIMEOUT,
+                    write_timeout=self._WRITE_TIMEOUT,
+                    pool_timeout=self._POOL_TIMEOUT,
+                )
+        except BadRequest as e:
+            if "message is not modified" in str(e).lower():
+                return
+            if self._is_photo_dimension_error(e):
+                try:
+                    with open(image_path, "rb") as image_file:
+                        media = InputMediaDocument(media=image_file, caption=caption)
+                        await query.edit_message_media(
+                            media=media,
+                            reply_markup=keyboard,
+                            connect_timeout=self._CONNECT_TIMEOUT,
+                            read_timeout=self._READ_TIMEOUT,
+                            write_timeout=self._WRITE_TIMEOUT,
+                            pool_timeout=self._POOL_TIMEOUT,
+                        )
+                except BadRequest as document_error:
+                    if "message is not modified" in str(document_error).lower():
+                        return
+                    raise
+                return
+            raise
+
+    def _build_keyboard(self, token: str) -> Any:
+        return InlineKeyboardMarkup(
+            [
+                [
+                    InlineKeyboardButton(
+                        text="←",
+                        callback_data=f"{self._callback_prefix}:{token}:prev",
+                    ),
+                    InlineKeyboardButton(
+                        text="确定",
+                        callback_data=f"{self._callback_prefix}:{token}:apply",
+                    ),
+                    InlineKeyboardButton(
+                        text="→",
+                        callback_data=f"{self._callback_prefix}:{token}:next",
+                    ),
+                ]
+            ]
+        )
+
+    def _build_caption(
+        self,
+        template_name: str,
+        index: int,
+        total: int,
+        applied: bool = False,
+    ) -> str:
+        current_active = self.config_manager.get_report_template()
+        active_mark = "✅ 当前生效" if template_name == current_active else "未生效"
+        apply_mark = "\n\n✅ 已应用该模板" if applied else ""
+        return (
+            f"🎨 模板预览 ({index + 1}/{total})\n"
+            f"当前项: {template_name}\n"
+            f"状态: {active_mark}\n\n"
+            "操作: ← 上一个 / 确定应用 / → 下一个"
+            f"{apply_mark}"
+        )
+
+    @staticmethod
+    def _extract_platform_meta(platform: Any) -> tuple[str | None, str | None]:
+        metadata = getattr(platform, "metadata", None)
+        if not metadata and hasattr(platform, "meta"):
+            try:
+                metadata = platform.meta()
+            except Exception:
+                metadata = None
+
+        platform_id = None
+        platform_name = None
+        if metadata:
+            if isinstance(metadata, dict):
+                platform_id = metadata.get("id")
+                platform_name = metadata.get("type") or metadata.get("name")
+            else:
+                platform_id = getattr(metadata, "id", None)
+                platform_name = getattr(metadata, "type", None) or getattr(
+                    metadata, "name", None
+                )
+        if platform_name:
+            platform_name = str(platform_name).lower()
+        if platform_id:
+            platform_id = str(platform_id)
+        return platform_id, platform_name
+
+    @staticmethod
+    def _extract_platform_client(platform: Any) -> Any | None:
+        client = None
+        if hasattr(platform, "get_client"):
+            try:
+                client = platform.get_client()
+            except Exception:
+                client = None
+        if client is None:
+            client = getattr(platform, "client", None)
+        if client is None:
+            application = getattr(platform, "application", None)
+            if application is not None:
+                client = getattr(application, "bot", None)
+        if client is None:
+            return None
+        if not hasattr(client, "send_photo"):
+            return None
+        return client
+
+    @staticmethod
+    def _get_raw_event_client(event: AstrMessageEvent) -> Any | None:
+        client = getattr(event, "client", None)
+        if client:
+            return client
+        return getattr(event, "bot", None)
+
+    def _get_event_client(
+        self, event: AstrMessageEvent, platform_id: str | None = None
+    ) -> Any | None:
+        client = self._get_raw_event_client(event)
+        if client is not None and hasattr(client, "send_photo"):
+            return client
+        if platform_id:
+            cached = self._platform_clients.get(platform_id)
+            if cached is not None:
+                return cached
+        return None
+
+    @staticmethod
+    def _resolve_chat_target(
+        event: AstrMessageEvent,
+    ) -> tuple[int | str, int | None] | None:
+        try:
+            group_id = event.get_group_id()
+        except Exception:
+            group_id = ""
+
+        if group_id:
+            raw_target = str(group_id)
+        else:
+            try:
+                raw_target = str(event.get_sender_id())
+            except Exception:
+                return None
+
+        chat_part = raw_target
+        thread_id: int | None = None
+        if "#" in raw_target:
+            chat_part, thread_part = raw_target.split("#", 1)
+            try:
+                thread_id = int(thread_part)
+            except (TypeError, ValueError):
+                thread_id = None
+
+        try:
+            chat_id: int | str = int(chat_part)
+        except (TypeError, ValueError):
+            chat_id = chat_part
+        return chat_id, thread_id
+
+    def _cleanup_expired_sessions(self) -> None:
+        now = time.time()
+        expired_tokens = [
+            token
+            for token, session in self._sessions.items()
+            if now - session.created_at > self._SESSION_TTL_SECONDS
+        ]
+        for token in expired_tokens:
+            self._sessions.pop(token, None)
+
+        if len(self._sessions) <= self._MAX_SESSIONS:
+            return
+
+        ordered = sorted(self._sessions.items(), key=lambda item: item[1].created_at)
+        overflow_count = len(self._sessions) - self._MAX_SESSIONS
+        for token, _ in ordered[:overflow_count]:
+            self._sessions.pop(token, None)
+
+    @staticmethod
+    def _is_photo_dimension_error(err: Exception) -> bool:
+        message = str(err).lower()
+        return "photo_invalid_dimensions" in message or "invalid dimensions" in message