diff --git a/assets/retro_futurism-demo.jpg b/assets/retro_futurism-demo.jpg index ac43056..6e3279b 100644 Binary files a/assets/retro_futurism-demo.jpg and b/assets/retro_futurism-demo.jpg differ diff --git a/main.py b/main.py index 818138f..a4cb3a9 100644 --- a/main.py +++ b/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) diff --git a/src/application/commands/__init__.py b/src/application/commands/__init__.py new file mode 100644 index 0000000..26c01f8 --- /dev/null +++ b/src/application/commands/__init__.py @@ -0,0 +1,5 @@ +"""命令相关应用服务。""" + +from .template_command_service import TemplateCommandService + +__all__ = ["TemplateCommandService"] diff --git a/src/application/commands/template_command_service.py b/src/application/commands/template_command_service.py new file mode 100644 index 0000000..95966d3 --- /dev/null +++ b/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) diff --git a/src/infrastructure/platform/template_preview/__init__.py b/src/infrastructure/platform/template_preview/__init__.py new file mode 100644 index 0000000..e8175ea --- /dev/null +++ b/src/infrastructure/platform/template_preview/__init__.py @@ -0,0 +1,6 @@ +"""平台模板预览交互能力。""" + +from .router import TemplatePreviewRouter +from .telegram_preview_handler import TelegramTemplatePreviewHandler + +__all__ = ["TelegramTemplatePreviewHandler", "TemplatePreviewRouter"] diff --git a/src/infrastructure/platform/template_preview/router.py b/src/infrastructure/platform/template_preview/router.py new file mode 100644 index 0000000..fc4ca5b --- /dev/null +++ b/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, [] diff --git a/src/infrastructure/platform/template_preview/telegram_preview_handler.py b/src/infrastructure/platform/template_preview/telegram_preview_handler.py new file mode 100644 index 0000000..bbb8895 --- /dev/null +++ b/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