本文详细阐述了在 `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 loggingfrom telegram import Update, Botfrom 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 和目标用户 IDBOT_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, filtersasync 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 始终拥有最新且准确的聊天信息。
以上就是Telegram Bot v20 启动时执行任务与信息获取指南的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1377139.html
微信扫一扫 
支付宝扫一扫 
