如何用python-telegram-bot实现Telegram群组消息全量转发？

Telegram Bot API没有提供直接获取聊天所有历史消息的接口，但可以通过分批拉取历史消息+循环转发的方式实现需求，以下是具体实现方案：

前提条件 #

• 机器人在源群组和目标群组均为管理员，且拥有「读取消息历史」「发送消息」权限
• 使用python-telegram-bot v20及以上版本（异步API版本）

核心实现代码 #

批量拉取并转发历史消息的函数 #

import asyncio
from telegram import Update
from telegram.ext import ContextTypes, CommandHandler, Application

async def forward_all_group_messages(context: ContextTypes.DEFAULT_TYPE):
    source_chat_id = "<替换为源群组ID>"
    target_chat_id = "<替换为目标群组ID>"
    offset_id = 0  # 起始偏移ID，从最新消息开始拉取
    batch_limit = 100  # 每次拉取的最大消息数（Telegram API限制最大值为100）

    while True:
        # 拉取一批历史消息
        messages = await context.bot.get_messages(
            chat_id=source_chat_id,
            offset_id=offset_id,
            limit=batch_limit
        )

        if not messages:
            break  # 无更多消息时终止循环

        # 反转消息顺序，保证按时间先后转发（API默认从新到旧返回消息）
        for msg in reversed(messages):
            try:
                await msg.forward(chat_id=target_chat_id)
                # 添加延迟避免触发API频率限制
                await asyncio.sleep(0.5)
            except Exception as e:
                print(f"转发消息ID {msg.id} 失败: {str(e)}")

        # 更新偏移ID，拉取更早的消息
        offset_id = messages[-1].id - 1

触发转发的命令处理器 #

async def trigger_forward(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("开始批量转发历史消息...")
    await forward_all_group_messages(context)
    await update.message.reply_text("历史消息转发完成！")

def main():
    # 替换为你的机器人Token
    bot_token = "YOUR_BOT_TOKEN"
    application = Application.builder().token(bot_token).build()

    # 添加命令触发处理器
    application.add_handler(CommandHandler("start_forward", trigger_forward))

    application.run_polling()

if __name__ == "__main__":
    main()

注意事项 #

• API频率限制：Telegram对Bot的请求频率有限制，批量转发时必须添加延迟（如示例中的0.5秒），否则可能导致账号被临时封禁
• 权限验证：务必确认机器人在两个群组的管理员权限包含所需操作权限，否则会抛出权限错误
• 异常处理：部分消息（如已删除消息、私密内容）可能无法拉取或转发，需通过异常捕获避免程序中断
• 大历史量处理：如果群组历史消息极多，首次转发耗时较长，建议在低峰时段执行

内容的提问来源于stack exchange，提问作者Amin Sanei