Telegram Bot 启动时获取与发送信息的指南

DDD
发布: 2025-10-11 14:28:26
原创
981人浏览过

Telegram Bot 启动时获取与发送信息的指南

本文详细介绍了在 `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() 即可。

Designify
Designify

拖入图片便可自动去除背景✨

Designify 90
查看详情 Designify

获取 Bot 所在聊天列表的策略

原问题中提到希望在启动时获取 Bot 所在的所有聊天(私聊、群组、超级群组)列表。然而,Telegram Bot API 并不提供直接获取 Bot 所在所有聊天列表的接口。 这是一个重要的限制。Bot 只能“知道”那些通过更新(例如,收到消息、被添加到群组、群成员状态改变等)与它交互过的聊天。

因此,唯一可靠的方法是手动追踪。这意味着你需要:

  1. 监听 ChatMemberUpdated 更新:
    • 当 Bot 被添加到一个群组时,会收到 ChatMemberUpdated 更新。
    • 当用户将 Bot 添加到私聊时,也会触发类似的更新或收到第一条消息。
    • 当 Bot 的管理员权限发生变化,或 Bot 被从群组中移除时,也会收到相应的更新。
  2. 维护一个本地的聊天列表:
    • 通过处理这些 ChatMemberUpdated 更新,你可以构建一个包含聊天 ID、聊天类型、标题/用户名、Bot 在该聊天中的状态(是否是所有者、管理员权限等)的列表。
    • 这个列表应该存储在 Bot 的持久化存储中,例如使用 PicklePersistence 或自定义的数据库。
  3. 持久化数据:
    • 为了确保 Bot 重启后仍能访问这些信息,必须将这个聊天列表持久化到磁盘或数据库中。PicklePersistence 是一个方便的选项,它会自动保存和加载 Bot 的数据。

概念性实现流程:

  1. 定义一个持久化存储结构: 在 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: {...}
    # }
    登录后复制
  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))
    登录后复制
  3. 在 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("初始化完成,已发送聊天列表(如果存在)。")
    登录后复制

注意事项与最佳实践

  1. Bot API 限制: 再次强调,Telegram Bot API 没有提供直接获取 Bot 所在所有聊天列表的方法。所有关于 Bot 所在聊天的信息都必须通过监听更新事件来逐步构建。
  2. 持久化是关键: 任何需要在 Bot 重启后保留的数据(如聊天列表)都必须进行持久化。PicklePersistence 是 python-telegram-bot 内置的一个简单易用的选项。
  3. 数据同步: 即使通过 ChatMemberUpdated 追踪,也可能存在短暂的数据不同步情况。例如,Bot 在离线期间被添加或移除,或者持久化文件损坏。因此,在发送这些信息时,应注明其基于已追踪的数据。
  4. 错误处理: 在 post_init_handler 中进行 API 调用时,应考虑添加错误处理机制(如 try-except 块),以防网络问题或 API 限制导致调用失败。
  5. 消息长度: Telegram 消息有长度限制(通常为 4096 字符)。如果聊天列表非常长,需要考虑分批发送或截断消息。

总结

在 python-telegram-bot v20 中,post_init_handler 提供了一个理想的入口点,用于在 Bot 开始轮询前执行自定义的初始化逻辑。通过 application.bot 实例,可以方便地进行 Telegram API 调用,例如发送启动消息。然而,获取 Bot 所在的所有聊天列表并非一个直接的 API 调用,需要通过监听 ChatMemberUpdated 更新并结合持久化机制来手动构建和维护。理解这些机制是构建健壮和功能完善的 Telegram Bot 的关键。

以上就是Telegram Bot 启动时获取与发送信息的指南的详细内容,更多请关注php中文网其它相关文章!

最佳 Windows 性能的顶级免费优化软件
最佳 Windows 性能的顶级免费优化软件

每个人都需要一台速度更快、更稳定的 PC。随着时间的推移,垃圾文件、旧注册表数据和不必要的后台进程会占用资源并降低性能。幸运的是,许多工具可以让 Windows 保持平稳运行。

下载
来源:php中文网
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系admin@php.cn
最新问题
开源免费商场系统广告
热门教程
更多>
最新下载
更多>
网站特效
网站源码
网站素材
前端模板
关于我们 免责申明 举报中心 意见反馈 讲师合作 广告合作 最新更新
php中文网:公益在线php培训,帮助PHP学习者快速成长!
关注服务号 技术交流群
PHP中文网订阅号
每天精选资源文章推送

Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号