如何使用Pyrogram实现Telegram机器人的对话式交互？

核心思路 #

结合telethon（模拟真实用户操作）和python-telegram-bot（运行待测试机器人），通过异步流程实现发送消息、监听回复、按钮点击等完整测试逻辑。

步骤与代码示例 #

1. 安装依赖 #

pip install python-telegram-bot telethon pytest asyncio

2. 准备测试资源 #

• 机器人Token（从@BotFather获取）
• 测试用户的API ID和API Hash（从Telegram开发者平台获取）
• 测试用户手机号（用于Telethon登录验证）
• 机器人与测试用户的聊天ID（可通过机器人接收/start消息后从更新日志中提取）

3. 待测试机器人示例 #

假设机器人逻辑：收到/start后发送带按钮的消息，点击按钮返回指定内容。

# bot.py
from telegram import InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import ApplicationBuilder, CommandHandler, CallbackQueryHandler

async def start(update, context):
    keyboard = [[InlineKeyboardButton("点击测试", callback_data='test_button')]]
    reply_markup = InlineKeyboardMarkup(keyboard)
    await update.message.reply_text('欢迎使用测试机器人，请点击按钮：', reply_markup=reply_markup)

async def button_callback(update, context):
    query = update.callback_query
    await query.answer()
    await query.edit_message_text(text=f"你点击了按钮，回调数据：{query.data}")

def main():
    application = ApplicationBuilder().token("YOUR_BOT_TOKEN").build()
    application.add_handler(CommandHandler("start", start))
    application.add_handler(CallbackQueryHandler(button_callback))
    application.run_polling()

if __name__ == '__main__':
    main()

4. 自动化测试代码 #

# test_bot.py
import asyncio
from telethon import TelegramClient, events
from bot import main as run_bot

# 配置信息
BOT_TOKEN = "YOUR_BOT_TOKEN"
USER_API_ID = YOUR_USER_API_ID
USER_API_HASH = "YOUR_USER_API_HASH"
USER_PHONE = "+1234567890"
BOT_USERNAME = "@your_bot_username"

async def test_bot_interaction():
    # 后台启动待测试机器人
    bot_task = asyncio.create_task(run_bot())
    
    # 初始化Telethon客户端（模拟真实用户）
    client = TelegramClient('test_session', USER_API_ID, USER_API_HASH)
    await client.start(USER_PHONE)
    
    try:
        # 发送/start命令给机器人
        await client.send_message(BOT_USERNAME, '/start')
        
        # 监听机器人初始回复
        @client.on(events.NewMessage(from_users=BOT_USERNAME))
        async def handle_initial_reply(event):
            print(f"收到机器人初始回复：{event.raw_text}")
            # 检测到按钮则模拟点击
            if event.message.reply_markup:
                await event.message.click(0)
        
        # 监听按钮点击后的回复并验证
        @client.on(events.NewMessage(from_users=BOT_USERNAME))
        async def handle_button_reply(event):
            reply_text = event.raw_text
            assert "你点击了按钮" in reply_text, "按钮交互回复不符合预期"
            print(f"按钮点击后回复：{reply_text}")
        
        # 等待测试流程完成
        await asyncio.sleep(10)
    finally:
        # 清理资源
        await client.disconnect()
        bot_task.cancel()

if __name__ == '__main__':
    asyncio.run(test_bot_interaction())

5. 关键逻辑说明 #

• 真实用户模拟：Telethon客户端完全模拟真实用户行为，支持发送消息、点击Inline按钮等操作，避免模拟请求的局限性。
• 异步并行控制：通过asyncio同时运行机器人和测试逻辑，确保回复监听与操作触发的时序正确性。
• 回复验证：通过断言直接验证机器人回复内容，确保功能符合预期。

注意事项 #

• 首次运行Telethon客户端需完成手机号验证码验证，后续会话会自动保存到test_session.session文件。
• 确保测试用户与机器人已互相关注，否则无法发送消息。
• 可基于pytest封装多个测试用例，覆盖不同场景下的交互逻辑。

内容的提问来源于stack exchange，提问作者João Pedro Martins de Paula