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

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

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() 即可。

微信 WeLM
微信 WeLM

WeLM不是一个直接的对话机器人,而是一个补全用户输入信息的生成模型。

微信 WeLM33
查看详情 微信 WeLM

获取 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
最新问题
开源免费商场系统广告
热门教程
更多>
最新下载
更多>
网站特效
网站源码
网站素材
前端模板
关于我们 免责申明 意见反馈 讲师合作 广告合作 最新更新 English
php中文网:公益在线php培训,帮助PHP学习者快速成长!
关注服务号 技术交流群
PHP中文网订阅号
每天精选资源文章推送
PHP中文网APP
随时随地碎片化学习
PHP中文网抖音号
发现有趣的

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