本教程详细讲解如何在 pyrogram 中有效处理和转发 telegram 媒体组消息，避免重复发送问题。文章阐述了 telegram 媒体组的特性，并提供了一种基于消息流识别和 `copy_media_group` 方法的解决方案，确保用户机器人能够准确无误地复制和转发完整的媒体组。
引言：理解 Pyrogram 中的媒体组消息
在 Pyrogram 中开发用户机器人时，处理 Telegram 的媒体组（如相册或文件集合）是一个常见的挑战。当机器人接收到一个媒体组时，Pyrogram 会将组内的每个媒体项作为独立的消息触发 on_message 事件。这导致的一个常见问题是，如果直接尝试转发每个触发的消息，最终会发送多份相同的媒体项，而不是一个完整的媒体组。
Telegram 将媒体组视为一种视觉上的分组机制，而非一个单一的容器对象。这意味着，虽然用户界面上它们显示为一个整体，但在底层 API 层面，它们仍然是多条独立的消息，只是通过 media_group_id 字段关联起来。因此，解决问题的关键在于正确识别并等待媒体组中的所有消息都到达，然后一次性转发整个组。
解决方案核心：识别与批量转发
要正确处理媒体组并避免重复转发，核心思路是：
- 识别媒体组消息： 通过检查 message.media_group_id 属性来判断当前消息是否属于一个媒体组。
- 获取完整媒体组： 使用 app.get_media_group() 方法获取与当前 media_group_id 相关联的所有消息。
- 确定组内最后一条消息： 在 on_message 事件多次触发时，只有当处理到媒体组的最后一条消息时，才执行转发操作。
- 批量转发： 使用 app.copy_media_group() 方法一次性转发整个媒体组。
实现步骤与示例代码
下面是一个 Pyrogram 用户机器人示例，演示了如何根据上述核心思路来处理和转发媒体组。
首先，确保你的环境中安装了 pyrogram 和 pyyaml：
pip install pyrogram pyyaml
然后，创建一个 config.yaml 文件来存储你的 API 凭据和目标频道 ID：
id: YOUR_API_ID hash: YOUR_API_HASH channels: - SOURCE_CHAT_ID_1 # 来源聊天ID，可以是数字或用户名 - SOURCE_CHAT_ID_2 my_channel: TARGET_CHANNEL_ID # 目标频道ID
请将 YOUR_API_ID、YOUR_API_HASH、SOURCE_CHAT_ID_X 和 TARGET_CHANNEL_ID 替换为你的实际值。SOURCE_CHAT_ID_X 可以是单个 ID 或一个列表。
接下来是 Python 代码实现：
import pyrogram
from pyrogram import filters
import yaml
import asyncio
# 加载配置
try:
with open("config.yaml", "r") as f:
config = yaml.safe_load(f)
except FileNotFoundError:
print("Error: config.yaml not found. Please create one with your API ID, API HASH, and channel IDs.")
exit()
api_id = config["id"]
api_hash = config["hash"]
source_channels = config["channels"]
target_channel = config["my_channel"]
# 初始化 Pyrogram 客户端
app = pyrogram.Client(
'my_account',
api_id=api_id,
api_hash=api_hash
)
# 用于存储已处理的媒体组ID，避免重复处理
processed_media_groups = set()
@app.on_message(filters.chat(source_channels))
async def copy_to_my_channel(client, message):
# 如果消息没有 media_group_id，则它是单条消息
if message.media_group_id is None:
print(f"Received single message (ID: {message.id}). Copying...")
await client.copy_message(
from_chat_id=message.chat.id,
chat_id=target_channel,
message_id=message.id
)
print(f"Single message (ID: {message.id}) copied.")
else:
# 这是一个媒体组消息
media_group_id = message.media_group_id
# 检查是否已经处理过这个媒体组，如果是，则跳过
if media_group_id in processed_media_groups:
# print(f"Media group {media_group_id} already processed. Skipping message {message.id}.")
return
print(f"Received message (ID: {message.id}) belonging to media group {media_group_id}.")
# 获取媒体组中的所有消息
# 注意：get_media_group 可能会在媒体组消息尚未完全到达时被调用，
# 导致返回不完整的列表。可以添加一个短暂停顿或重试机制来提高健壮性。
await asyncio.sleep(0.5) # 给Telegram一点时间来聚合所有消息
full_media_group = await client.get_media_group(
chat_id=message.chat.id,
message_id=message.id # 任何属于该媒体组的消息ID都可以
)
# 查找当前消息在完整媒体组中的索引
current_message_index = -1
for i, msg in enumerate(full_media_group):
if msg.id == message.id:
current_message_index = i
break
# 只有当当前消息是媒体组中的最后一条消息时，才执行转发
# 这里的判断逻辑需要更严谨，因为消息到达的顺序不总是线性的。
# 最可靠的方法是等待一段时间，然后一次性处理。
# 但为了演示，我们先使用这种基于“最后一条消息”的简化逻辑。
# 更好的做法是收集所有消息ID，然后在检测到“完成”时发送。
# 对于Pyrogram的get_media_group，它通常会返回完整的组，
# 所以我们只需确保只执行一次copy_media_group。
# 标记此媒体组为已处理，防止后续触发重复发送
processed_media_groups.add(media_group_id)
# 提取所有消息的ID
message_ids_to_copy = [msg.id for msg in full_media_group]
print(f"Detected complete media group {media_group_id} with IDs: {message_ids_to_copy}. Copying...")
# 使用 copy_media_group 转发整个媒体组
await client.copy_media_group(
chat_id=target_channel,
from_chat_id=message.chat.id,
message_ids=message_ids_to_copy
)
print(f"Media group {media_group_id} published to {target_channel}.")
# 运行客户端
if __name__ == '__main__':
print("Bot started. Listening for messages...")
app.run()代码解析与注意事项
- config.yaml 配置： 将敏感信息（如 API ID 和 Hash）和可变配置（如频道 ID）外部化到 YAML 文件中，方便管理和部署。
- @app.on_message(filters.chat(source_channels))： 这是一个消息处理器，它会监听来自 config.yaml 中指定 source_channels 的所有消息。
-
单条消息处理：
- if message.media_group_id is None:：如果消息没有 media_group_id，则它是一个独立的消息（如纯文本、单个图片等）。
- await client.copy_message(...)：直接使用 copy_message 方法转发此单条消息。
-
媒体组消息处理：
- else: 块处理带有 media_group_id 的消息。
- processed_media_groups = set()：引入一个集合来存储已处理的 media_group_id。这是防止重复转发的关键机制。当 on_message 对同一个媒体组中的多条消息触发时，只有第一次会进入实际处理逻辑。
- await asyncio.sleep(0.5)：这是一个经验性的等待，因为 Telegram 在发送媒体组时，并非所有消息都会同时到达。短暂的延迟可以增加 get_media_group 返回完整列表的概率。对于生产环境，可能需要更复杂的重试或消息收集机制。
- full_media_group = await client.get_media_group(chat_id=message.chat.id, message_id=message.id)：这是获取整个媒体组的关键方法。message_id 可以是媒体组中任意一条消息的 ID。Pyrogram 会使用这个 ID 去查询该媒体组中的所有相关消息。
- processed_media_groups.add(media_group_id)：一旦我们决定处理一个媒体组（通常是第一次遇到其任何一条消息时），就将其 media_group_id 添加到 processed_media_groups 集合中，这样后续属于同一媒体组的消息触发 on_message 时，就会被 if media_group_id in processed_media_groups: 语句跳过，避免重复处理。
- message_ids_to_copy = [msg.id for msg in full_media_group]：从 full_media_group 中提取所有消息的 ID，以备批量转发。
- await client.copy_media_group(chat_id=target_channel, from_chat_id=message.chat.id, message_ids=message_ids_to_copy)：这是 Pyrogram 提供的专门用于转发整个媒体组的方法。它接收一个消息 ID 列表，并将其作为一个整体发送到目标聊天。
总结
通过上述方法，我们能够有效地在 Pyrogram 中处理 Telegram 媒体组消息。关键在于理解 Telegram 媒体组的底层机制，利用 message.media_group_id 进行识别，并通过 client.get_media_group() 获取完整组信息，最终使用 client.copy_media_group() 进行批量转发。结合 processed_media_groups 集合来避免重复处理，可以确保机器人行为的准确性和效率。这个模式不仅适用于转发，也适用于任何需要对整个媒体组进行统一操作的场景。
