Telegram Bot 启动时定制化操作与信息获取指南

霞舞

发布时间：2025-10-11 14:45:01

1366人浏览过

来源于php中文网

原创

Telegram Bot 启动时定制化操作与信息获取指南

本文深入探讨了在 `python-telegram-bot` v20 中，如何在 bot 启动时执行定制化操作和获取信息。重点介绍了 `applicationbuilder` 的 `post_init_handler` 回调函数，展示了如何在其中安全地进行 telegram api 调用，并明确指出 bot api 不提供直接获取 bot 所属所有聊天列表的方法，强调了通过 `chatmemberupdated` 更新手动维护列表的必要性与最佳实践。

引言：Telegram Bot 启动时的定制化操作

在开发 Telegram Bot 时，我们经常需要在 Bot 启动完成、但尚未开始接收用户更新之前执行一些初始化任务。这些任务可能包括发送启动通知、加载配置、或者执行一次性数据收集等。对于 python-telegram-bot (PTB) v20 版本，由于其采用了 asyncio 和 ApplicationBuilder 的新架构，传统的 Updater/Dispatcher 模式不再适用，因此理解正确的启动流程和 API 调用方式变得尤为重要。

本教程将详细讲解如何在 Bot 启动阶段（特别是在 Application 构建完成但 run_polling() 尚未执行之前）安全地执行异步操作，包括进行 Telegram API 调用，并探讨获取 Bot 所属聊天列表的实际方法和限制。

核心概念：post_init_handler 的作用

python-telegram-bot 提供了 ApplicationBuilder 来构建 Bot 的核心 Application 实例。其中，post_init 回调函数是专门为在 Bot 启动前执行自定义逻辑而设计的。

执行时机：post_init_handler 在 Application 实例完全构建并初始化完毕后、但在 application.run_polling() 开始监听更新之前被调用。这是一个理想的时机来执行一次性的启动前任务。
异步特性：post_init_handler 必须是一个 async 函数，因为其中的操作通常涉及异步 I/O，例如与 Telegram API 的交互。
访问 Bot 实例：在 post_init_handler 中，您可以通过传入的 application 对象访问到已完全初始化的 Bot 实例，即 application.bot。这个 Bot 实例可以直接用于进行 Telegram API 调用。

在 post_init_handler 中执行 Telegram API 调用

要在 post_init_handler 中进行 API 调用，您只需使用 application.bot 实例。它提供了所有标准的 Telegram Bot API 方法，例如 send_message、get_me 等。

示例：发送一条启动消息

以下代码展示了如何在 Bot 启动时向特定用户发送一条“Hello World”消息：

from telegram import Update, Application
from telegram.ext import ApplicationBuilder, PicklePersistence
import asyncio

# 假设您有一个配置文件存储了bot_token和persistent_data_file_path
bot_token = "YOUR_BOT_TOKEN"
persistent_data_file_path = "bot_data.pkl"
TARGET_USER_ID = 123456789 # 替换为实际的用户ID

async def post_init_handler(application: Application) -> None:
    """
    在 Bot 启动前执行的初始化逻辑。
    """
    print("Bot 应用程序初始化中...")

    # 访问 Bot 实例并获取 Bot ID
    bot_info = await application.bot.get_me()
    print(f"Bot ID: {bot_info.id}, Username: @{bot_info.username}")

    # 向特定用户发送一条启动消息
    try:
        await application.bot.send_message(
            chat_id=TARGET_USER_ID,
            text=f"Bot 已启动！Bot ID: {bot_info.id}, Username: @{bot_info.username}"
        )
        print(f"启动消息已发送至用户 {TARGET_USER_ID}")
    except Exception as e:
        print(f"发送启动消息失败: {e}")

async def post_stop_handler(application: Application) -> None:
    """
    在 Bot 停止后执行的清理逻辑。
    """
    print("Bot 应用程序已停止。")

def main() -> None:
    persistence_object = PicklePersistence(filepath=persistent_data_file_path)
    application = (
        ApplicationBuilder()
        .token(bot_token)
        .persistence(persistence=persistence_object)
        .post_init(post_init_handler) # 注册 post_init_handler
        .post_stop(post_stop_handler) # 注册 post_stop_handler
        .build()
    )

    # run_polling() 将在 post_init_handler 执行完成后开始
    application.run_polling(allowed_updates=Update.ALL_TYPES)

if __name__ == "__main__":
    main()

Application.create_task 的辨析

Application.create_task 是 python-telegram-bot 提供的一个便捷函数，用于在 Bot 的事件循环中调度异步任务。它本质上是对 asyncio.create_task 的封装，确保任务在 Bot 的生命周期内正确管理。然而，对于简单的、一次性的 API 调用，直接 await application.bot.some_api_call() 即可，无需通过 create_task。create_task 更适用于需要在后台持续运行或独立于主流程执行的异步任务。

DreamStudio

SD兄弟产品！AI 图像生成器

下载

获取 Bot 所在聊天列表的限制与策略

一个常见的需求是 Bot 在启动时获取其所有所属的私人聊天、群组和频道列表。然而，Telegram Bot API 并没有提供一个直接的 API 调用来列出 Bot 所在的所有聊天。这是 Bot API 的一个设计限制。

唯一的可靠方法：通过 ChatMemberUpdated 更新手动维护列表

由于 Bot API 的限制，唯一可靠的方法是 Bot 手动追踪其加入和离开的聊天。这通常通过以下方式实现：

监听 ChatMemberUpdated 更新：当 Bot 被添加到群组/频道、或从群组/频道移除时，Telegram 会发送 ChatMemberUpdated 更新。通过注册一个 ChatMemberHandler，您可以捕获这些事件。
维护持久化存储：Bot 需要一个机制来持久化存储它所知道的聊天信息。当收到 ChatMemberUpdated 更新时，根据更新内容（例如 new_chat_member.status）来添加、更新或移除持久化列表中的聊天记录。
启动时加载：在 post_init_handler 或其他适当的启动逻辑中，Bot 可以从持久化存储中加载之前维护的聊天列表。

python-telegram-bot 的 chatmemberbot.py 示例展示了这种手动追踪的方法。虽然这种方法需要更多的代码来实现，并且依赖于 Bot 能够接收到所有相关的 ChatMemberUpdated 更新，但它是目前最准确且唯一可行的解决方案。

持久化存储的注意事项：

使用 PicklePersistence 或自定义的持久化方案来存储聊天列表。
确保在 Bot 关闭时数据能够正确保存，在启动时能够正确加载。
考虑数据损坏或丢失的情况，以及如何处理与实际状态不一致的可能性。

完整示例：Bot 启动时发送自身信息与聊天列表（概念性）

虽然无法直接获取所有聊天列表，但我们可以结合 post_init_handler 和手动维护的列表，在启动时发送 Bot 自身信息以及已知的聊天信息。

from telegram import Update, Application, ChatMemberUpdated, Chat
from telegram.ext import ApplicationBuilder, PicklePersistence, ChatMemberHandler
import asyncio
import logging

logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)

bot_token = "YOUR_BOT_TOKEN"
persistent_data_file_path = "bot_data.pkl"
ADMIN_USER_ID = 123456789 # 替换为您的管理员用户ID

# 用于存储 Bot 已知聊天的字典 {chat_id: chat_info_tuple}
# chat_info_tuple 结构: (chat_id, chat_title, chat_type, is_bot_owner, bot_admin_rights)
known_chats = {}

async def post_init_handler(application: Application) -> None:
    """
    Bot 启动前的初始化逻辑。
    """
    logger.info("Bot 应用程序初始化中...")

    # 从持久化存储中加载已知的聊天列表
    # 注意：PTB 的 persistence 默认会处理 bot_data, chat_data, user_data
    # 如果 known_chats 是单独维护的，需要手动加载或集成到 bot_data 中
    if "known_chats" in application.bot_data:
        global known_chats
        known_chats = application.bot_data["known_chats"]
        logger.info(f"从持久化存储加载了 {len(known_chats)} 个已知聊天。")
    else:
        logger.info("未找到已知聊天列表，从头开始。")

    # 获取 Bot 自身信息
    bot_info = await application.bot.get_me()
    logger.info(f"Bot ID: {bot_info.id}, Username: @{bot_info.username}")

    # 准备要发送的启动信息
    startup_message_parts = [
        f"Bot 已启动！",
        f"Bot ID: {bot_info.id}",
        f"Username: @{bot_info.username}",
        "\n--- 已知聊天列表 ---"
    ]

    if known_chats:
        for chat_id, chat_data in known_chats.items():
            # 格式化聊天信息
            chat_id_str, title, chat_type, is_owner, admin_rights = chat_data
            admin_info = "(ignored)"
            if admin_rights:
                admin_info = f"拥有权限: {admin_rights.to_dict()}" # 假设 admin_rights 是 ChatMemberAdministrator 对象
            elif is_owner:
                admin_info = "是所有者"

            startup_message_parts.append(
                f"{chat_id_str},{title},{chat_type},{is_owner},{admin_info}"
            )
    else:
        startup_message_parts.append("无已知聊天。请等待 Bot 加入新的群组或频道。")

    # 将信息发送给管理员
    try:
        await application.bot.send_message(
            chat_id=ADMIN_USER_ID,
            text="\n".join(startup_message_parts)
        )
        logger.info(f"启动信息已发送至管理员 {ADMIN_USER_ID}")
    except Exception as e:
        logger.error(f"发送启动信息失败: {e}")

async def post_stop_handler(application: Application) -> None:
    """
    Bot 停止后的清理逻辑。
    """
    logger.info("Bot 应用程序已停止。")
    # 确保在停止前保存 known_chats 到持久化存储
    application.bot_data["known_chats"] = known_chats
    logger.info("已知聊天列表已保存到持久化存储。")


async def track_chats_handler(update: Update, context) -> None:
    """
    处理 ChatMemberUpdated 更新，维护 Bot 所在聊天列表。
    """
    chat_member: ChatMemberUpdated = update.chat_member

    # 确保是 Bot 自身的成员状态变化
    if chat_member.new_chat_member.user.id != context.bot.id:
        return

    chat: Chat = chat_member.chat
    chat_id = str(chat.id)
    chat_title = chat.title or chat.username or f"Private Chat with {chat.first_name}"
    chat_type = chat.type

    # 检查 Bot 的新状态
    new_status = chat_member.new_chat_member.status
    old_status = chat_member.old_chat_member.status if chat_member.old_chat_member else None

    is_bot_owner = False
    bot_admin_rights = None # None 或 ChatMemberAdministrator 对象

    if new_status == ChatMemberUpdated.MEMBER or new_status == ChatMemberUpdated.ADMINISTRATOR:
        # Bot 加入或成为管理员
        if new_status == ChatMemberUpdated.ADMINISTRATOR:
            bot_admin_rights = chat_member.new_chat_member.can_manage_chat # 示例，实际应获取具体权限
            # 进一步判断是否为所有者，通常通过 is_creator 字段判断
            if hasattr(chat_member.new_chat_member, 'is_creator') and chat_member.new_chat_member.is_creator:
                is_bot_owner = True

        # 存储或更新聊天信息
        known_chats[chat_id] = (chat_id, chat_title, chat_type, is_bot_owner, bot_admin_rights)
        logger.info(f"Bot 加入或更新了聊天: {chat_title} ({chat_id}), 类型: {chat_type}")

    elif new_status == ChatMemberUpdated.LEFT or new_status == ChatMemberUpdated.KICKED:
        # Bot 离开或被踢出
        if chat_id in known_chats:
            del known_chats[chat_id]
            logger.info(f"Bot 离开了聊天: {chat_title} ({chat_id})")

    # 将更新后的 known_chats 存储到 bot_data，以便持久化
    context.application.bot_data["known_chats"] = known_chats


def main() -> None:
    persistence_object = PicklePersistence(filepath=persistent_data_file_path)
    application = (
        ApplicationBuilder()
        .token(bot_token)
        .persistence(persistence=persistence_object)
        .post_init(post_init_handler)
        .post_stop(post_stop_handler)
        .build()
    )

    # 注册 ChatMemberHandler 来追踪 Bot 所在聊天
    application.add_handler(ChatMemberHandler(track_chats_handler, ChatMemberHandler.CHAT_MEMBER))

    logger.info("Bot 正在启动...")
    application.run_polling(allowed_updates=Update.ALL_TYPES)

if __name__ == "__main__":
    main()

注意事项与最佳实践

异步操作的正确使用：post_init_handler 是一个 async 函数，所有内部的异步操作（如 application.bot.send_message）都必须使用 await 关键字。
错误处理：在 post_init_handler 中执行的任何 API 调用都应该包含适当的错误处理（try-except 块），以防止因网络问题或 API 错误导致 Bot 启动失败。
持久化存储：对于需要长期维护的数据（如已知的聊天列表），务必使用 PicklePersistence 或其他持久化方案来确保数据在 Bot 重启后不会丢失。
Bot API 限制：再次强调，Telegram Bot API 没有直接获取所有 Bot 所在聊天列表的方法。手动追踪是唯一的解决方案，这意味着 Bot 需要运行一段时间才能建立起完整的聊天列表。
allowed_updates 的配置：为了接收 ChatMemberUpdated 更新，确保在 run_polling() 中配置了 allowed_updates 包含 Update.CHAT_MEMBER 类型（或使用 Update.ALL_TYPES）。

总结

python-telegram-bot v20 提供了强大的 ApplicationBuilder 机制，通过 post_init_handler 回调函数，开发者可以优雅地在 Bot 启动前执行各种初始化任务和 API 调用。虽然在获取 Bot 所在聊天列表方面存在 Bot API 的固有限制，但通过结合 ChatMemberHandler 和持久化存储，我们可以构建一个健壮的系统来手动追踪和管理这些信息。理解这些机制对于开发功能完善、可靠的 Telegram Bot 至关重要。

Python中使用try语句写入文件时内容未保存的解决方案

Python 中使用 try 语句写入文件失败的常见原因及最佳实践

Python 中基于时间戳保留每篇文章最新操作记录的去重方法

Python 中按时间戳保留每篇文章最新操作记录的去重方法

Python 用户输入空格处理与健壮性错误控制完整指南