本文详细介绍了在 `python-telegram-bot` v20 中，如何在 bot 启动后、开始轮询前执行自定义逻辑，并与 telegram api 交互。重点阐述了 `post_init_handler` 的正确使用方式，如何通过 `application.bot` 发送信息，以及 bot api 不直接提供获取所有聊天列表的接口，因此需要通过手动追踪 `chatmemberupdated` 更新并结合持久化来维护聊天信息。
引言：Telegram Bot 启动时的信息处理挑战
在开发 python-telegram-bot 应用程序时，特别是在版本 20 及更高版本中，开发者常常面临一个挑战：如何在 Bot 启动后、开始处理用户更新（即轮询）之前，执行一些初始化逻辑，例如获取 Bot 的自身信息、发送一条启动通知，或加载并处理一些持久化数据。这涉及到 ApplicationBuilder 的配置、asyncio 的使用，以及对 Telegram Bot API 行为的深入理解。本文将针对这些问题提供一个全面的教程。
正确执行启动逻辑的位置：post_init_handler
python-telegram-bot 提供了一个专门的回调函数 post_init_handler，它正是为在 Bot 启动前执行自定义逻辑而设计的。这个回调函数会在 Application 对象构建完成且 Bot 实例可用后，但在 run_polling() 开始监听更新之前被调用。这是一个 async 函数，因此可以在其中执行异步操作。
使用示例：
from telegram.ext import ApplicationBuilder, Application
from telegram.ext import PicklePersistence # 假设使用持久化
import asyncio
# 定义持久化文件路径
persistent_data_file_path = "bot_data.pkl"
async def post_init_handler(application: Application) -> None:
"""
在 Bot 启动后、开始轮询前执行的初始化逻辑。
"""
print("Bot 初始化中...")
# Bot 实例在此处已完全可用
bot_id = application.bot.id
print(f"Bot ID: {bot_id}")
# 示例：发送一条启动消息给特定用户
target_user_id = 123456789 # 替换为实际的用户ID
await application.bot.send_message(
chat_id=target_user_id,
text=f"Bot (ID: {bot_id}) 已成功启动并初始化。"
)
# 可以在此处加载持久化数据并进行处理
# 例如：从 application.persistence.get_bot_data() 获取数据
async def post_stop_handler(application: Application) -> None:
"""
在 Bot 停止前执行的清理逻辑。
"""
print("Bot 停止中...")
# 可以在此处执行关闭前的清理工作，例如保存数据
def main() -> None:
# 配置持久化对象
persistence_object = PicklePersistence(filepath=persistent_data_file_path)
# 构建 Application 实例
application = (
ApplicationBuilder()
.token("YOUR_BOT_TOKEN") # 替换为你的 Bot Token
.persistence(persistence=persistence_object)
.post_init(post_init_handler) # 注册 post_init_handler
.post_stop(post_stop_handler) # 注册 post_stop_handler
.build()
)
# 启动 Bot 轮询
application.run_polling()
if __name__ == "__main__":
main()在 post_init_handler 中，application.bot 实例已经可用，可以直接用于调用 Telegram Bot API 的方法。
在启动时进行 Telegram API 调用
要与 Telegram API 交互，你只需要一个 Bot 实例。在 post_init_handler 中，application.bot 已经提供了这个实例，因此你可以直接使用它来发送消息、获取 Bot 信息等。
例如，发送一条简单的消息：
async def post_init_handler(application: Application) -> None:
# ... 其他初始化逻辑 ...
target_chat_id = 123456789 # 替换为目标聊天的 ID
await application.bot.send_message(text="Hello World from Bot startup!", chat_id=target_chat_id)关于 Application.create_task 的说明：
Application.create_task 是 python-telegram-bot 提供的一个便利函数，用于在 Bot 的事件循环中创建和管理异步任务。它通常用于执行一些需要在后台运行的、与主轮询逻辑并行但又需要访问 Application 上下文的任务。然而，它并非用于直接进行一次性的 Telegram API 请求，而是用于更复杂的异步编程场景。对于在 post_init_handler 中直接进行 API 调用，直接使用 await application.bot.method() 即可。
获取 Bot 所在聊天列表的策略
原问题中提到希望在启动时获取 Bot 所在的所有聊天（私聊、群组、超级群组）列表。然而，Telegram Bot API 并不提供直接获取 Bot 所在所有聊天列表的接口。 这是一个重要的限制。Bot 只能“知道”那些通过更新（例如，收到消息、被添加到群组、群成员状态改变等）与它交互过的聊天。
因此，唯一可靠的方法是手动追踪。这意味着你需要：
-
监听 ChatMemberUpdated 更新：
- 当 Bot 被添加到一个群组时，会收到 ChatMemberUpdated 更新。
- 当用户将 Bot 添加到私聊时，也会触发类似的更新或收到第一条消息。
- 当 Bot 的管理员权限发生变化，或 Bot 被从群组中移除时，也会收到相应的更新。
-
维护一个本地的聊天列表：
- 通过处理这些 ChatMemberUpdated 更新，你可以构建一个包含聊天 ID、聊天类型、标题/用户名、Bot 在该聊天中的状态（是否是所有者、管理员权限等）的列表。
- 这个列表应该存储在 Bot 的持久化存储中，例如使用 PicklePersistence 或自定义的数据库。
-
持久化数据：
- 为了确保 Bot 重启后仍能访问这些信息，必须将这个聊天列表持久化到磁盘或数据库中。PicklePersistence 是一个方便的选项，它会自动保存和加载 Bot 的数据。
概念性实现流程：
-
定义一个持久化存储结构： 在 application.bot_data (如果使用 PicklePersistence) 中存储一个字典，键为 chat_id，值为包含聊天信息的对象。
# 假设在某个地方定义了持久化数据结构 # application.bot_data 可以在 post_init_handler 中访问 # 结构示例： # application.bot_data['known_chats'] = { # chat_id_1: { # 'title': 'Chat A', # 'type': 'group', # 'is_owner': True, # 'admin_rights': {...} # }, # chat_id_2: {...} # } -
创建 ChatMemberUpdated 处理器：
- 注册一个 ChatMemberHandler 来监听 ChatMemberUpdated 更新。
- 在这个处理器中，当 Bot 的成员状态发生变化时，更新 application.bot_data 中的 known_chats 字典。
from telegram.ext import ChatMemberHandler, ContextTypes from telegram import ChatMember async def chat_member_update(update: ChatMemberUpdated, context: ContextTypes.DEFAULT_TYPE) -> None: """处理 Bot 成员状态更新，维护聊天列表。""" chat = update.effective_chat bot_member: ChatMember = update.new_chat_member if chat.id not in context.bot_data.get('known_chats', {}): context.bot_data.setdefault('known_chats', {})[chat.id] = {} chat_info = context.bot_data['known_chats'][chat.id] chat_info['title'] = chat.title or chat.full_name # 对于私聊是 full_name chat_info['type'] = chat.type chat_info['username'] = chat.username # 对于群组可能是 None if bot_member.status == ChatMember.OWNER: chat_info['is_owner'] = True chat_info['admin_rights'] = bot_member.rights.to_dict() if bot_member.rights else None elif bot_member.status == ChatMember.ADMINISTRATOR: chat_info['is_owner'] = False chat_info['admin_rights'] = bot_member.rights.to_dict() if bot_member.rights else None elif bot_member.status == ChatMember.MEMBER: chat_info['is_owner'] = False chat_info['admin_rights'] = None elif bot_member.status == ChatMember.LEFT or bot_member.status == ChatMember.KICKED: # 如果 Bot 离开了聊天，则从列表中移除 if chat.id in context.bot_data.get('known_chats', {}): del context.bot_data['known_chats'][chat.id] return # Bot 离开了，无需继续更新信息 # 更多状态处理... # 在 main() 中添加处理器 # application.add_handler(ChatMemberHandler(chat_member_update, chat_member_types=ChatMemberHandler.MY_CHAT_MEMBER)) -
在 post_init_handler 中发送存储的列表：
- 在 post_init_handler 中，可以从 application.bot_data 加载这个已持久化的聊天列表。
- 然后，将这些信息格式化并发送给指定的管理员用户。
async def post_init_handler(application: Application) -> None: print("Bot 初始化中...") target_admin_user_id = 123456789 # 替换为管理员用户ID known_chats = application.bot_data.get('known_chats', {}) if known_chats: message_parts = ["Bot 所在聊天列表:"] for chat_id, info in known_chats.items(): title_or_username = info.get('title') or info.get('username', 'N/A') chat_type = info.get('type', 'N/A') is_owner = info.get('is_owner', False) admin_rights = info.get('admin_rights', {}) admin_rights_str = f"({', '.join(k for k, v in admin_rights.items() if v)})" if admin_rights else "(无)" message_parts.append( f"- ID: {chat_id}, 名称: {title_or_username}, 类型: {chat_type}, " f"是否所有者: {is_owner}, 管理权限: {admin_rights_str}" ) full_message = "\n".join(message_parts) # 确保消息不会过长，Telegram 消息有长度限制 if len(full_message) > 4096: full_message = full_message[:4000] + "\n...(消息过长，已截断)" await application.bot.send_message( chat_id=target_admin_user_id, text=full_message ) else: await application.bot.send_message( chat_id=target_admin_user_id, text="Bot 尚未记录任何聊天信息。" ) print("初始化完成，已发送聊天列表（如果存在）。")
注意事项与最佳实践
- Bot API 限制： 再次强调，Telegram Bot API 没有提供直接获取 Bot 所在所有聊天列表的方法。所有关于 Bot 所在聊天的信息都必须通过监听更新事件来逐步构建。
- 持久化是关键： 任何需要在 Bot 重启后保留的数据（如聊天列表）都必须进行持久化。PicklePersistence 是 python-telegram-bot 内置的一个简单易用的选项。
- 数据同步： 即使通过 ChatMemberUpdated 追踪，也可能存在短暂的数据不同步情况。例如，Bot 在离线期间被添加或移除，或者持久化文件损坏。因此，在发送这些信息时，应注明其基于已追踪的数据。
- 错误处理： 在 post_init_handler 中进行 API 调用时，应考虑添加错误处理机制（如 try-except 块），以防网络问题或 API 限制导致调用失败。
- 消息长度： Telegram 消息有长度限制（通常为 4096 字符）。如果聊天列表非常长，需要考虑分批发送或截断消息。
总结
在 python-telegram-bot v20 中，post_init_handler 提供了一个理想的入口点，用于在 Bot 开始轮询前执行自定义的初始化逻辑。通过 application.bot 实例，可以方便地进行 Telegram API 调用，例如发送启动消息。然而，获取 Bot 所在的所有聊天列表并非一个直接的 API 调用，需要通过监听 ChatMemberUpdated 更新并结合持久化机制来手动构建和维护。理解这些机制是构建健壮和功能完善的 Telegram Bot 的关键。
