Python Telegram Bot v20启动时获取/发送聊天信息的实现方案

完整实现代码 #

from telegram import ChatMember
from telegram.ext import ApplicationBuilder, PicklePersistence, Application

# 配置参数
BOT_TOKEN = "你的BotToken"
PERSISTENT_DATA_PATH = "bot_data.pickle"
TARGET_USER_ID = 123456789  # 替换为接收通知的用户ID

async def post_init_handler(application: Application) -> None:
    print("开始初始化聊天信息收集")
    bot = application.bot
    bot_id = bot.id
    unique_chat_ids = set()

    # 1. 从最近的更新中提取聊天ID
    try:
        updates = await bot.get_updates(offset=-1, limit=1000)  # 获取最近1000条更新
        for update in updates:
            if update.effective_chat:
                unique_chat_ids.add(update.effective_chat.id)
    except Exception as e:
        print(f"获取更新失败: {str(e)}")

    # 2. 从持久化数据中提取历史聊天ID
    try:
        if application.persistence:
            chat_ids_from_persistence = application.persistence.chat_data.keys()
            unique_chat_ids.update(chat_ids_from_persistence)
    except Exception as e:
        print(f"读取持久化聊天数据失败: {str(e)}")

    # 3. 遍历所有聊天ID，收集详细信息
    chat_summary = "Bot所在聊天信息汇总：\n\n"
    for chat_id in unique_chat_ids:
        try:
            # 获取聊天基本信息
            chat = await bot.get_chat(chat_id)
            # 获取Bot在该聊天中的身份信息
            member = await bot.get_chat_member(chat_id, bot_id)

            # 整理聊天名称
            chat_name = chat.title if chat.title else (chat.username if chat.username else "无名称")
            # 整理权限信息
            if member.status == "creator":
                owner_status = "是"
                admin_perms = "创建者（拥有全部权限）"
            elif member.status == "administrator":
                owner_status = "否"
                perms = member.permissions
                perm_list = []
                if perms.can_delete_messages: perm_list.append("删除消息")
                if perms.can_manage_chat: perm_list.append("管理聊天")
                if perms.can_restrict_members: perm_list.append("限制成员")
                if perms.can_promote_members: perm_list.append("提升管理员")
                if perms.can_change_info: perm_list.append("修改信息")
                if perms.can_invite_users: perm_list.append("邀请用户")
                if perms.can_pin_messages: perm_list.append("置顶消息")
                admin_perms = ", ".join(perm_list) if perm_list else "无具体权限"
            else:
                owner_status = "否"
                admin_perms = "非管理员"

            # 拼接信息
            chat_summary += (
                f"聊天ID：{chat_id}\n"
                f"名称/用户名：{chat_name}\n"
                f"聊天类型：{chat.type}\n"
                f"是否为创建者：{owner_status}\n"
                f"管理员权限：{admin_perms}\n\n"
            )
        except Exception as e:
            chat_summary += f"聊天ID：{chat_id}，获取信息失败：{str(e)}\n\n"

    # 4. 将汇总信息发送给指定用户
    try:
        await bot.send_message(chat_id=TARGET_USER_ID, text=chat_summary)
        print("聊天信息已发送给指定用户")
    except Exception as e:
        print(f"发送消息失败: {str(e)}")

def main() -> None:
    persistence = PicklePersistence(filepath=PERSISTENT_DATA_PATH)
    application = (
        ApplicationBuilder()
        .token(BOT_TOKEN)
        .persistence(persistence)
        .post_init(post_init_handler)
        .build()
    )
    application.run_polling()

if __name__ == "__main__":
    main()

关键问题解答 #

1. API调用方式 #

不需要使用application.create_task，直接在post_init_handler中用await调用Bot的异步API方法即可。post_init是Python-Telegram-Bot提供的异步初始化钩子，会在build()完成后、run_polling()启动前被自动await执行，所有内部的异步逻辑都会执行完毕后才进入轮询阶段，完全符合需求。

2. 代码位置选择 #

放在post_init_handler是最优解。如果直接写在build()和run_polling()之间的同步代码块中，需要手动用asyncio.run()包裹异步调用，反而增加复杂度；而post_init是官方设计的初始化逻辑入口，能确保Bot实例已完全初始化，避免出现bot_id获取失败的问题。

3. 聊天列表获取方案 #

Telegram Bot API没有提供直接获取Bot所有聊天的接口，因此采用两种方式互补覆盖：

• 调用get_updates获取最近的更新记录，提取其中的聊天ID，覆盖Bot近期参与的聊天
• 从持久化存储的chat_data中读取历史聊天ID，覆盖Bot有过交互的旧聊天
• 对两个来源的ID去重后，逐个调用get_chat和get_chat_member获取详细信息

注意事项 #

• 调整get_updates的limit参数可以覆盖更多历史更新，但最大不能超过1000（Telegram API限制）
• 若Bot被踢出某个聊天，调用get_chat或get_chat_member会抛出异常，代码中已加入异常捕获，避免流程中断
• 确保TARGET_USER_ID对应的用户已与Bot发起过私聊，否则Bot无法发送消息

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