
本文详细介绍了在 `python-telegram-bot` v20 中,如何在 bot 启动后、开始轮询前执行自定义逻辑,并与 telegram api 交互。重点阐述了 `post_init_handler` 的正确使用方式,如何通过 `application.bot` 发送信息,以及 bot api 不直接提供获取所有聊天列表的接口,因此需要通过手动追踪 `chatmemberupdated` 更新并结合持久化来维护聊天信息。
在开发 python-telegram-bot 应用程序时,特别是在版本 20 及更高版本中,开发者常常面临一个挑战:如何在 Bot 启动后、开始处理用户更新(即轮询)之前,执行一些初始化逻辑,例如获取 Bot 的自身信息、发送一条启动通知,或加载并处理一些持久化数据。这涉及到 ApplicationBuilder 的配置、asyncio 的使用,以及对 Telegram Bot API 行为的深入理解。本文将针对这些问题提供一个全面的教程。
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 交互,你只需要一个 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 所在的所有聊天(私聊、群组、超级群组)列表。然而,Telegram Bot API 并不提供直接获取 Bot 所在所有聊天列表的接口。 这是一个重要的限制。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 处理器:
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 中发送存储的列表:
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("初始化完成,已发送聊天列表(如果存在)。")在 python-telegram-bot v20 中,post_init_handler 提供了一个理想的入口点,用于在 Bot 开始轮询前执行自定义的初始化逻辑。通过 application.bot 实例,可以方便地进行 Telegram API 调用,例如发送启动消息。然而,获取 Bot 所在的所有聊天列表并非一个直接的 API 调用,需要通过监听 ChatMemberUpdated 更新并结合持久化机制来手动构建和维护。理解这些机制是构建健壮和功能完善的 Telegram Bot 的关键。
以上就是Telegram Bot 启动时获取与发送信息的指南的详细内容,更多请关注php中文网其它相关文章!
每个人都需要一台速度更快、更稳定的 PC。随着时间的推移,垃圾文件、旧注册表数据和不必要的后台进程会占用资源并降低性能。幸运的是,许多工具可以让 Windows 保持平稳运行。
Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号