基于Pyrogram的Telegram Bot内联按钮无响应问题求助

问题描述 #

开发基于Pyrogram的Telegram Bot，通过REST API提供视频讲座链接。发送/lecture指令后，Bot能正常发送带内联按钮的消息，但点击按钮后无任何响应。添加打印语句后日志无输出、无报错，先后尝试正则匹配回调数据、startswith处理回调查询，问题依旧。

相关代码 #

最初使用正则匹配的代码 #

import re
import requests
from pyrogram import Client, filters
from pyrogram.types import InlineKeyboardButton, InlineKeyboardMarkup

from bot import Bot

@Bot.on_message(filters.command('lecture'))
async def lectures_command(client, message):
    # 创建科目选择内联键盘
    keyboard = [
        [
            InlineKeyboardButton("Physics", callback_data="subject_physics"),
            InlineKeyboardButton("Maths", callback_data="subject_maths"),
        ],
        [
            InlineKeyboardButton("Organic Chemistry", callback_data="subject_organic"),
            InlineKeyboardButton("Inorganic Chemistry", callback_data="subject_inorganic"),
        ],
        [
            InlineKeyboardButton("Physical Chemistry", callback_data="subject_physical"),
        ]
    ]
    reply_markup = InlineKeyboardMarkup(keyboard)
    await message.reply_text("Choose a subject from below please:", reply_markup=reply_markup)

@Bot.on_callback_query(filters.regex(r'^subject_.*'))
async def subject_callback(_, query):
    print('ok')
    subject = query.data.split("_")[1]
    print('subject:', subject)
    # 从API获取数据
    response = requests.get(f"https://MY-REST-API/teachers?subject={subject}")
    if response.status_code == 200:
        teachers_data = response.json()
        buttons = [InlineKeyboardButton(teacher, callback_data=f"teacher_{subject}_{teacher}") for teacher in teachers_data[0]["teachers"]]
        buttons.append(InlineKeyboardButton("Choose a subject", callback_data="subject_choice"))
        reply_markup = InlineKeyboardMarkup([buttons])
        await query.message.edit_text(f"Choose a teacher for {subject}:", reply_markup=reply_markup)
    else:
        await query.message.edit_text("Failed to fetch data from the API. Please try again later.")

@Bot.on_callback_query(filters.regex(r'^teacher_.*'))
async def teacher_callback(_, query):
    data_parts = re.split(r'_', query.data)
    subject = data_parts[1]
    teacher_name = data_parts[2]
    response = requests.get(f"https://MY-REST-API/chapters?subject={subject}&teacher={teacher_name}")
    if response.status_code == 200:
        chapters_data = response.json()
        buttons = [InlineKeyboardButton(chapter, callback_data=f"chapter_{subject}_{teacher_name}_{chapter}") for chapter in chapters_data["chapters"]]
        reply_markup = InlineKeyboardMarkup([buttons])
        await query.message.edit_text("Please choose a chapter from below buttons:", reply_markup=reply_markup)
    else:
        await query.message.edit_text("Failed to fetch data from the API. Please try again later.")

@Bot.on_callback_query(filters.regex(r'^chapter_.*'))
async def chapter_callback(_, query):
    data_parts = re.split(r'_', query.data)
    subject = data_parts[1]
    teacher_name = data_parts[2]
    chapter_name = data_parts[3]
    response = requests.get(f"https://MY-REST-API/lecture?subject={subject}&teacher={teacher_name}&ch={chapter_name}")
    if response.status_code == 200 and response.json()["success"]:
        lecture_link = response.json()["link"]
        await query.message.reply_text(f"Here's the lecture link for {chapter_name} by {teacher_name}: {lecture_link}")
    else:
        await query.message.reply_text("Failed to fetch lecture link. Please try again later.")

改用startswith处理的代码 #

import re
import requests
from pyrogram import Client, filters
from pyrogram.types import InlineKeyboardButton, InlineKeyboardMarkup

from bot import Bot

@Bot.on_message(filters.command('lecture'))
async def lectures_command(client, message):
    # 创建科目选择内联键盘
    keyboard = [
        [
            InlineKeyboardButton("Physics", callback_data="subject_physics"),
            InlineKeyboardButton("Maths", callback_data="subject_maths"),
        ],
        [
            InlineKeyboardButton("Organic Chemistry", callback_data="subject_organic"),
            InlineKeyboardButton("Inorganic Chemistry", callback_data="subject_inorganic"),
        ],
        [
            InlineKeyboardButton("Physical Chemistry", callback_data="subject_physical"),
        ]
    ]
    reply_markup = InlineKeyboardMarkup(keyboard)
    await message.reply_text("Choose a subject from below please:", reply_markup=reply_markup)

@Bot.on_callback_query()
async def handle_callback(_, query):
    if query.data.startswith("subject_"):
        print('ok')
        subject = query.data.split("_")[1]
        response = requests.get(f"https://MY-REST-API/teachers?subject={subject}")
        if response.status_code == 200:
            teachers_data = response.json()
            buttons = [InlineKeyboardButton(teacher, callback_data=f"teacher_{subject}_{teacher}") for teacher in teachers_data[0]["teachers"]]
            buttons.append(InlineKeyboardButton("Choose a subject", callback_data="subject_choice"))
            reply_markup = InlineKeyboardMarkup([buttons])
            await query.message.edit_text(f"Choose a teacher for {subject}:", reply_markup=reply_markup)
        else:
            await query.message.edit_text("Failed to fetch data from the API. Please try again later.")
    elif query.data.startswith("teacher_"):
        data_parts = query.data.split("_")
        subject = data_parts[1]
        teacher_name = data_parts[2]
        response = requests.get(f"https://MY-REST-API/chapters?subject={subject}&teacher={teacher_name}")
        if response.status_code == 200:
            chapters_data = response.json()
            buttons = [InlineKeyboardButton(chapter, callback_data=f"chapter_{subject}_{teacher_name}_{chapter}") for chapter in chapters_data["chapters"]]
            reply_markup = InlineKeyboardMarkup([buttons])
            await query.message.edit_text("Please choose a chapter from below buttons:", reply_markup=reply_markup)
        else:
            await query.message.edit_text("Failed to fetch data from the API. Please try again later.")
    elif query.data.startswith("chapter_"):
        data_parts = query.data.split("_")
        subject = data_parts[1]
        teacher_name = data_parts[2]
        chapter_name = data_parts[3]
        response = requests.get(f"https://MY-REST-API/lecture?subject={subject}&teacher={teacher_name}&ch={chapter_name}")
        if response.status_code == 200 and response.json()["success"]:
            lecture_link = response.json()["link"]
            await query.message.reply_text(f"Here's the lecture link for {chapter_name} by {teacher_name}: {lecture_link}")
        else:
            await query.message.reply_text("Failed to fetch lecture link. Please try again later.")
    else:
        print('no ok')

解决方案 #

1. 强制调用query.answer() #

Pyrogram要求处理回调查询时必须调用await query.answer()，否则Telegram服务器会判定请求未处理，后续回调会被屏蔽。即使不需要返回提示，也要添加这行代码，示例：

@Bot.on_callback_query()
async def handle_callback(_, query):
    await query.answer()  # 必须添加这行
    if query.data.startswith("subject_"):
        # 后续逻辑...

2. 修复同步IO阻塞问题 #

代码中使用的requests.get()是同步阻塞操作，会卡住异步事件循环，导致回调无法处理。改用异步HTTP库aiohttp：

# 替换requests导入
import aiohttp

# 异步请求示例
async with aiohttp.ClientSession() as session:
    async with session.get(f"https://MY-REST-API/teachers?subject={subject}") as response:
        if response.status == 200:
            teachers_data = await response.json()
            # 后续逻辑...

3. 检查Bot运行模式 #

• Polling模式：重启Bot，检查会话是否正常，避免Polling连接异常导致回调丢失。
• Webhook模式：确认Webhook地址可被Telegram服务器访问，无防火墙、反向代理拦截请求，可通过Bot API测试Webhook状态。

4. 启用DEBUG日志 #

添加日志配置，查看更详细的运行信息：

import logging
logging.basicConfig(level=logging.DEBUG)

内容的提问来源于stack exchange，提问作者Ashish Dube