本文详细阐述了在 `python-telegram-bot` v20 中，如何在 bot 启动时，即 `run_polling()` 之前，执行自定义逻辑和发送信息。核心在于利用 `post_init_handler` 回调函数，并通过 `application.bot` 实例进行 telegram api 调用。文章也明确指出，telegram bot api 不提供直接获取 bot 所属所有聊天列表的方法，需要开发者自行通过事件更新来维护。
引言：在 Bot 启动时执行任务的挑战
在开发 Telegram Bot 时，有时我们需要在 Bot 开始接收并处理更新（即 run_polling() 启动）之前，执行一些初始化任务，例如发送一条启动消息、加载特定配置或进行数据检查。对于 python-telegram-bot (PTB) v20 版本，由于其异步特性和 Application 架构的引入，许多旧版（如使用 Updater/Dispatcher）的示例不再适用，这给新开发者带来了困惑。本文将针对这一常见需求，提供一个清晰的教程，指导您如何在 Bot 启动阶段安全有效地执行自定义逻辑。
1. 正确的 API 调用方式
要在 Bot 启动时与 Telegram API 交互，您只需要一个 Bot 实例。在 Application 架构中，这个实例已经通过 application.bot 属性提供。这意味着您无需手动构建新的 Bot 对象，可以直接利用 application.bot 来调用各种 Telegram API 方法，例如 send_message、get_me 等。
示例：发送一条启动消息
import logging
from telegram import Update, Bot
from telegram.ext import Application, ApplicationBuilder, ContextTypes, PicklePersistence
# 配置日志
logging.basicConfig(
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
)
logger = logging.getLogger(__name__)
# 替换为您的 Bot Token 和目标用户 ID
BOT_TOKEN = "YOUR_BOT_TOKEN"
TARGET_USER_ID = 123456789 # 替换为实际的用户ID，Bot 会向此用户发送启动消息
PERSISTENCE_FILE_PATH = "bot_data.pkl" # 持久化文件路径
async def post_init_handler(application: Application) -> None:
"""
在 Bot 启动并构建 Application 之后，但在开始轮询之前执行的异步回调。
"""
logger.info("Bot 启动中：执行 post_init_handler...")
bot_id = application.bot.id # 获取 Bot 自身的 ID
logger.info(f"Bot ID: {bot_id}")
# 使用 application.bot 发送消息
try:
await application.bot.send_message(
chat_id=TARGET_USER_ID,
text=f"Bot 已启动！我的 ID 是：{bot_id}\n准备好接收更新。"
)
logger.info(f"成功向用户 {TARGET_USER_ID} 发送了启动消息。")
except Exception as e:
logger.error(f"发送启动消息失败: {e}")
async def post_stop_handler(application: Application) -> None:
"""
在 Bot 停止后执行的异步回调。
"""
logger.info("Bot 已停止：执行 post_stop_handler...")
try:
await application.bot.send_message(
chat_id=TARGET_USER_ID,
text="Bot 已停止运行。"
)
logger.info(f"成功向用户 {TARGET_USER_ID} 发送了停止消息。")
except Exception as e:
logger.error(f"发送停止消息失败: {e}")
def main() -> None:
"""主函数，设置并运行 Bot"""
persistence_object = PicklePersistence(filepath=PERSISTENCE_FILE_PATH)
application = (
ApplicationBuilder()
.token(BOT_TOKEN)
.persistence(persistence=persistence_object)
.post_init(post_init_handler) # 注册启动前回调
.post_stop(post_stop_handler) # 注册停止后回调
.build()
)
logger.info("Bot Application 已构建，即将开始轮询...")
application.run_polling(allowed_updates=Update.ALL_TYPES) # 开始轮询
logger.info("Bot 轮询已停止。")
if __name__ == "__main__":
main()注意事项：
- Application.create_task 是一个用于在 PTB 应用程序中创建和管理异步任务的便捷函数，但它与直接进行 Telegram API 请求是不同的概念。当您需要直接调用 Bot API 时，直接使用 application.bot 即可。
- post_init_handler 必须是一个 async 函数，因为 Bot API 调用是异步的。
2. 正确的代码放置位置
post_init_handler 是执行 Bot 启动逻辑的理想位置。这个回调函数在 Application 对象构建完毕之后、但在 application.run_polling() 开始接收更新之前被调用。这确保了您的初始化逻辑在 Bot 完全准备好处理消息之前完成，并且可以安全地访问 application.bot 实例。
将代码放置在 application.build() 和 application.run_polling() 之间，直接访问 application.bot 可能会失败，因为在 post_init_handler 被调用之前，application.bot 可能尚未完全初始化或绑定到事件循环。post_init_handler 专门为此目的而设计，因此是更推荐和更可靠的选择。
3. 获取 Bot 所属聊天数据（及其限制）
开发者常常希望在 Bot 启动时获取 Bot 所加入的所有私聊、群组和超级群组的列表，以及 Bot 在这些聊天中的权限等信息。然而，Telegram Bot API 并没有提供一个直接的 API 调用来获取 Bot 所属的所有聊天列表。
这意味着您无法通过一个简单的 application.bot.get_all_chats() 这样的方法来一次性获取所需数据。这是 Bot API 的一个设计限制。
如何跟踪 Bot 所属的聊天？
唯一可靠的方法是手动跟踪。这通常通过监听 chat_member 更新来实现。当 Bot 被添加到群组、从群组移除、或其成员状态发生变化时，Telegram 会发送 chat_member 更新。开发者需要：
- 注册 chat_member 处理器： 监听这些更新。
- 维护一个持久化列表： 每当收到 chat_member 更新时，根据更新内容（例如，Bot 加入了新群组，或者从旧群组中被移除），更新您自己的持久化存储（例如数据库、文件或 PicklePersistence）。
- 启动时加载： 在 post_init_handler 中，您可以从这个持久化存储中加载之前维护的聊天列表。
虽然这种方法需要一些额外的开发工作，并且列表的初始构建需要时间（Bot 只能在收到更新后才得知其加入的聊天），但它是最准确和唯一可行的方案。它避免了与实际聊天状态脱钩的风险。
示例代码（概念性，非完整实现）：
以下代码片段展示了如何在 post_init_handler 中加载一个假想的持久化聊天列表，并发送给指定用户。请注意，_load_chats_from_persistence() 是一个需要您自行实现的函数，用于从持久化存储中读取数据。
# ... (前面的导入和 main 函数结构保持不变) ...
async def _load_chats_from_persistence(application: Application) -> list[tuple]:
"""
模拟从持久化存储中加载 Bot 所属的聊天信息。
实际应用中，您需要在此处实现从数据库或文件读取逻辑。
"""
# 假设 persistence_object 存储了 chat_data
# 实际中，您可能需要定义一个更复杂的结构来存储这些数据
if application.persistence and application.persistence.bot_data:
# 这是一个简化的示例，假设 chat_data 直接存储在 bot_data 中
# 实际可能需要一个特定的键，如 application.persistence.bot_data.get("known_chats", [])
known_chats = application.persistence.bot_data.get("known_chats", {})
logger.info(f"从持久化加载了 {len(known_chats)} 个已知聊天。")
# 转换为所需的格式 (chat id, username/title, chat type, is_owner, admin_rights)
formatted_chats = []
for chat_id, chat_info in known_chats.items():
formatted_chats.append(
(
chat_id,
chat_info.get("title", chat_info.get("username", "N/A")),
chat_info.get("type", "N/A"),
chat_info.get("is_owner", "N/A"), # 需要在 chat_member 更新中判断并存储
chat_info.get("admin_rights", "N/A") # 需要在 chat_member 更新中判断并存储
)
)
return formatted_chats
return []
async def post_init_handler(application: Application) -> None:
"""
在 Bot 启动并构建 Application 之后，但在开始轮询之前执行的异步回调。
"""
logger.info("Bot 启动中：执行 post_init_handler...")
bot_id = application.bot.id
logger.info(f"Bot ID: {bot_id}")
# 1. 发送启动消息
try:
await application.bot.send_message(
chat_id=TARGET_USER_ID,
text=f"Bot 已启动！我的 ID 是：{bot_id}\n准备好接收更新。"
)
logger.info(f"成功向用户 {TARGET_USER_ID} 发送了启动消息。")
except Exception as e:
logger.error(f"发送启动消息失败: {e}")
# 2. 获取并发送 Bot 已知聊天列表（从持久化存储中加载）
known_chats_data = await _load_chats_from_persistence(application)
if known_chats_data:
message_parts = ["Bot 已知的聊天列表："]
for chat_id, title, chat_type, is_owner, admin_rights in known_chats_data:
message_parts.append(
f"{chat_id},{title},{chat_type},{is_owner},{admin_rights}"
)
full_message = "\n".join(message_parts)
try:
await application.bot.send_message(
chat_id=TARGET_USER_ID,
text=full_message
)
logger.info(f"成功向用户 {TARGET_USER_ID} 发送了已知聊天列表。")
except Exception as e:
logger.error(f"发送已知聊天列表失败: {e}")
else:
logger.info("没有从持久化存储中找到已知的聊天数据。")
# ... (main 函数保持不变) ...实现 chat_member 处理器来维护聊天列表的简要说明：
您需要添加一个 ChatMemberHandler 到您的 Application，并在其回调中更新持久化存储。
from telegram.ext import ChatMemberHandler, MessageHandler, filters
async def track_chats(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""处理 chat_member 更新以维护 Bot 所属的聊天列表。"""
if not update.chat_member:
return
chat = update.chat_member.chat
old_member = update.chat_member.old_chat_member
new_member = update.chat_member.new_chat_member
bot_data = context.application.persistence.bot_data # 获取持久化存储的 bot_data
if "known_chats" not in bot_data:
bot_data["known_chats"] = {}
chat_id = chat.id
chat_info = {
"title": chat.title,
"username": chat.username,
"type": chat.type,
# ... 更多您想存储的聊天信息
}
if new_member.user.id == context.bot.id: # 针对 Bot 自身的成员状态变化
if new_member.status == "member": # Bot 被添加到聊天
bot_data["known_chats"][chat_id] = chat_info
logger.info(f"Bot 加入了聊天: {chat.title} ({chat_id})")
elif new_member.status == "administrator": # Bot 成为管理员
bot_data["known_chats"][chat_id] = chat_info # 更新信息
# 可以在这里解析并存储 admin_rights
logger.info(f"Bot 在聊天 {chat.title} ({chat_id}) 中成为管理员。")
elif new_member.status == "left" or new_member.status == "kicked": # Bot 被移除
if chat_id in bot_data["known_chats"]:
del bot_data["known_chats"][chat_id]
logger.info(f"Bot 离开了聊天: {chat.title} ({chat_id})")
# 确保持久化数据被保存
context.application.persistence.flush()
def main() -> None:
# ... (前面的 ApplicationBuilder 部分) ...
application.add_handler(ChatMemberHandler(track_chats, ChatMemberHandler.MY_CHAT_MEMBER))
# ... (run_polling 部分) ...总结
在 python-telegram-bot v20 中，利用 ApplicationBuilder 的 post_init 回调是执行 Bot 启动时逻辑的最佳实践。您可以通过 application.bot 实例轻松地进行各种 Telegram API 调用，例如发送启动消息。然而，需要注意的是，Telegram Bot API 并不提供直接获取 Bot 所属所有聊天列表的功能。要实现此目的，您必须通过监听 chat_member 更新并自行维护一个持久化的聊天列表来解决。这种方法虽然需要更多的设置，但能确保您的 Bot 始终拥有最新且准确的聊天信息。
