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