mirror of
https://github.com/AstrBotDevs/AstrBot
synced 2026-07-15 17:30:13 +08:00
feat(telegram): 添加媒体组（相册）支持 / add media group (album) support (#4893)
* 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 <media_group_id> 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 <noreply@anthropic.com> * 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 <noreply@anthropic.com> * fix(telegram): handle missing message in media group processing and improve logging messages --------- Co-authored-by: Ubuntu <ubuntu@localhost.localdomain> Co-authored-by: Claude Sonnet 4.5 <noreply@anthropic.com> Co-authored-by: Soulter <905617992@qq.com>
This commit is contained in:
DD斩首
@@ -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}")
|
||||
|
||||
Reference in New Issue
Block a user