From dbe8e33c4b459392567165979b493405466b2cf9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?DD=E6=96=A9=E9=A6=96?= <155905740+DDZS987@users.noreply.github.com> Date: Sun, 8 Feb 2026 13:22:45 +0800 Subject: [PATCH] =?UTF-8?q?feat(telegram):=20=E6=B7=BB=E5=8A=A0=E5=AA=92?= =?UTF-8?q?=E4=BD=93=E7=BB=84=EF=BC=88=E7=9B=B8=E5=86=8C=EF=BC=89=E6=94=AF?= =?UTF-8?q?=E6=8C=81=20/=20add=20media=20group=20(album)=20support=20(#489?= =?UTF-8?q?3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(telegram): 添加媒体组（相册）支持 / add media group (album) support ## 功能说明 支持 Telegram 的媒体组消息（相册），将多张图片/视频合并为一条消息处理，而不是分散成多条消息。 ## 主要改动 ### 1. 初始化媒体组缓存 (__init__) - 添加 `media_group_cache` 字典存储待处理的媒体组消息 - 使用 2.5 秒超时收集媒体组消息（基于社区最佳实践） - 最大等待时间 10 秒（防止永久等待） ### 2. 消息处理流程 (message_handler) - 检测 `media_group_id` 判断是否为媒体组消息 - 媒体组消息走特殊处理流程，避免分散处理 ### 3. 媒体组消息缓存 (handle_media_group_message) - 缓存收到的媒体组消息 - 使用 APScheduler 实现防抖（debounce）机制 - 每收到新消息时重置超时计时器 - 超时后触发统一处理 ### 4. 媒体组合并处理 (process_media_group) - 从缓存中取出所有媒体项 - 使用第一条消息作为基础（保留文本、回复等信息） - 依次添加所有图片、视频、文档到消息链 - 将合并后的消息发送到处理流程 ## 技术方案论证 Telegram Bot API 在处理媒体组时的设计限制： 1. 将媒体组的每个消息作为独立的 update 发送 2. 每个 update 带有相同的 `media_group_id` 3. **不提供**组的总数、结束标志或一次性完整组的机制 因此，bot 必须自行收集消息，并通过硬编码超时（timeout/delay）等待可能延迟到达的消息。 这是目前唯一可靠的方案，被官方实现、主流框架和开发者社区广泛采用。 ### 官方和社区证据： - **Telegram Bot API 服务器实现（tdlib）**：明确指出缺少结束标志或总数信息 https://github.com/tdlib/telegram-bot-api/issues/643 - **Telegram Bot API 服务器 issue**：讨论媒体组处理的不便性，推荐使用超时机制 https://github.com/tdlib/telegram-bot-api/issues/339 - **Telegraf（Node.js 框架）**：专用媒体组中间件使用 timeout 控制等待时间 https://github.com/DieTime/telegraf-media-group - **StackOverflow 讨论**：无法一次性获取媒体组所有文件，必须手动收集 https://stackoverflow.com/questions/50180048/telegram-api-get-all-uploaded-photos-by-media-group-id - **python-telegram-bot 社区**：确认媒体组消息单独到达，需手动处理 https://github.com/python-telegram-bot/python-telegram-bot/discussions/3143 - **Telegram Bot API 官方文档**：仅定义 `media_group_id` 为可选字段，不提供获取完整组的接口 https://core.telegram.org/bots/api#message ## 实现细节 - 使用 2.5 秒超时收集媒体组消息（基于社区最佳实践） - 最大等待时间 10 秒（防止永久等待） - 采用防抖（debounce）机制：每收到新消息重置计时器 - 利用 APScheduler 实现延迟处理和任务调度 ## 测试验证 - ✅ 发送 5 张图片相册，成功合并为一条消息 - ✅ 保留原始文本说明和回复信息 - ✅ 支持图片、视频、文档混合的媒体组 - ✅ 日志显示 Processing media group with 5 items ## 代码变更 - 文件：astrbot/core/platform/sources/telegram/tg_adapter.py - 新增代码：124 行 - 新增方法：handle_media_group_message(), process_media_group() Co-Authored-By: Claude Sonnet 4.5 * refactor(telegram): 优化媒体组处理性能和可靠性 根据代码审查反馈改进： 1. 实现 media_group_max_wait 防止无限延迟 - 跟踪媒体组创建时间，超过最大等待时间立即处理 - 最坏情况下 10 秒内必定处理，防止消息持续到达导致无限延迟 2. 移除手动 job 查找优化性能 - 删除 O(N) 的 get_jobs() 循环扫描 - 依赖 replace_existing=True 自动替换任务 3. 重用 convert_message 减少代码重复 - 统一所有媒体类型转换逻辑 - 未来添加新媒体类型只需修改一处 Co-Authored-By: Claude Sonnet 4.5 * fix(telegram): handle missing message in media group processing and improve logging messages --------- Co-authored-by: Ubuntu Co-authored-by: Claude Sonnet 4.5 Co-authored-by: Soulter <905617992@qq.com> --- .../platform/sources/telegram/tg_adapter.py | 128 +++++++++++++++++- 1 file changed, 126 insertions(+), 2 deletions(-) diff --git a/astrbot/core/platform/sources/telegram/tg_adapter.py b/astrbot/core/platform/sources/telegram/tg_adapter.py index e7f2a102e..1e92972d6 100644 --- a/astrbot/core/platform/sources/telegram/tg_adapter.py +++ b/astrbot/core/platform/sources/telegram/tg_adapter.py @@ -89,6 +89,16 @@ class TelegramPlatformAdapter(Platform): self.scheduler = AsyncIOScheduler() + # Media group handling + # Cache structure: {media_group_id: {"created_at": datetime, "items": [(update, context), ...]}} + self.media_group_cache: dict[str, dict] = {} + self.media_group_timeout = self.config.get( + "telegram_media_group_timeout", 2.5 + ) # seconds - debounce delay between messages + self.media_group_max_wait = self.config.get( + "telegram_media_group_max_wait", 10.0 + ) # max seconds - hard cap to prevent indefinite delay + @override async def send_by_session( self, @@ -225,6 +235,13 @@ class TelegramPlatformAdapter(Platform): async def message_handler(self, update: Update, context: ContextTypes.DEFAULT_TYPE): logger.debug(f"Telegram message: {update.message}") + + # Handle media group messages + if update.message and update.message.media_group_id: + await self.handle_media_group_message(update, context) + return + + # Handle regular messages abm = await self.convert_message(update, context) if abm: await self.handle_msg(abm) @@ -399,6 +416,113 @@ class TelegramPlatformAdapter(Platform): return message + async def handle_media_group_message( + self, update: Update, context: ContextTypes.DEFAULT_TYPE + ): + """Handle messages that are part of a media group (album). + + Caches incoming messages and schedules delayed processing to collect all + media items before sending to the pipeline. Uses debounce mechanism with + a hard cap (max_wait) to prevent indefinite delay. + """ + from datetime import datetime, timedelta + + if not update.message: + return + + media_group_id = update.message.media_group_id + if not media_group_id: + return + + # Initialize cache for this media group if needed + if media_group_id not in self.media_group_cache: + self.media_group_cache[media_group_id] = { + "created_at": datetime.now(), + "items": [], + } + logger.debug(f"Create media group cache: {media_group_id}") + + # Add this message to the cache + entry = self.media_group_cache[media_group_id] + entry["items"].append((update, context)) + logger.debug( + f"Add message to media group {media_group_id}, " + f"currently has {len(entry['items'])} items.", + ) + + # Calculate delay: if already waited too long, process immediately; + # otherwise use normal debounce timeout + elapsed = (datetime.now() - entry["created_at"]).total_seconds() + if elapsed >= self.media_group_max_wait: + delay = 0 + logger.debug( + f"Media group {media_group_id} has reached max wait time " + f"({elapsed:.1f}s >= {self.media_group_max_wait}s), processing immediately.", + ) + else: + delay = self.media_group_timeout + logger.debug( + f"Scheduled media group {media_group_id} to be processed in {delay} seconds " + f"(already waited {elapsed:.1f}s)" + ) + + # Schedule/reschedule processing (replace_existing=True handles debounce) + job_id = f"media_group_{media_group_id}" + self.scheduler.add_job( + self.process_media_group, + "date", + run_date=datetime.now() + timedelta(seconds=delay), + args=[media_group_id], + id=job_id, + replace_existing=True, + ) + + async def process_media_group(self, media_group_id: str): + """Process a complete media group by merging all collected messages. + + Args: + media_group_id: The unique identifier for this media group + """ + if media_group_id not in self.media_group_cache: + logger.warning(f"Media group {media_group_id} not found in cache") + return + + entry = self.media_group_cache.pop(media_group_id) + updates_and_contexts = entry["items"] + if not updates_and_contexts: + logger.warning(f"Media group {media_group_id} is empty") + return + + logger.info( + f"Processing media group {media_group_id}, total {len(updates_and_contexts)} items" + ) + + # Use the first update to create the base message (with reply, caption, etc.) + first_update, first_context = updates_and_contexts[0] + abm = await self.convert_message(first_update, first_context) + + if not abm: + logger.warning( + f"Failed to convert the first message of media group {media_group_id}" + ) + return + + # Add additional media from remaining updates by reusing convert_message + for update, context in updates_and_contexts[1:]: + # Convert the message but skip reply chains (get_reply=False) + extra = await self.convert_message(update, context, get_reply=False) + if not extra: + continue + + # Merge only the message components (keep base session/meta from first) + abm.message.extend(extra.message) + logger.debug( + f"Added {len(extra.message)} components to media group {media_group_id}" + ) + + # Process the merged message + await self.handle_msg(abm) + async def handle_msg(self, message: AstrBotMessage): message_event = TelegramPlatformEvent( message_str=message.message_str, @@ -426,6 +550,6 @@ class TelegramPlatformAdapter(Platform): if self.application.updater is not None: await self.application.updater.stop() - logger.info("Telegram 适配器已被关闭") + logger.info("Telegram adapter has been closed.") except Exception as e: - logger.error(f"Telegram 适配器关闭时出错: {e}") + logger.error(f"Error occurred while closing Telegram adapter: {e}")