Telegram内联机器人按钮无响应及启动方式调整问题
Telegram内联机器人对话流程修复方案
问题背景
开发Telegram内联机器人,需求是用户可在任意聊天中通过@botusername + 自定义命令触发对话流程。当前存在两个核心问题:
- ConversationHandler需通过
/start命令启动,不符合内联触发的预期; - 机器人能显示初始消息和按钮,但点击按钮无响应,状态流转失效。
核心问题分析
- 对话入口配置错误:原ConversationHandler的entry_points使用CommandHandler,无法响应内联查询;
- 函数返回值类型不匹配:
inlinequery函数定义为返回None,但实际返回状态值,导致对话状态无法正确初始化; - 用户信息获取逻辑错误:回调处理函数错误访问
update.inline_query,回调查询需从update.callback_query获取用户数据; - 全局变量多用户冲突:EmojiCode等全局变量会被多用户操作覆盖,应使用用户上下文存储专属数据;
- 重复处理器冲突:dispatcher同时添加了ConversationHandler和独立的InlineQueryHandler,导致内联查询被重复处理。
修复后的完整代码
from datetime import datetime import logging import emojihash from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update from telegram.ext import ( Updater, CommandHandler, CallbackQueryHandler, ConversationHandler, CallbackContext, InlineQueryHandler ) from telegram import InlineQueryResultArticle, InputTextMessageContent logging.basicConfig( format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO ) logger = logging.getLogger(__name__) # 对话状态定义 FIRST, SECOND = range(2) ONE, TWO, THREE = range(3) # FOUR未使用,移除 verified_message_2 = "message 2" verified_message_1 = "Message 1 " def inlinequery(update: Update, context: CallbackContext) -> int: print("Inline hit!") query_text = update.inline_query.query print(f"len(query): {len(query_text)}") if len(query_text) == 0 or query_text[-1] != ".": print("Empty query, showing guide message") results = [ InlineQueryResultArticle( id="guide", title="请输入买家用户名并以句号结尾", input_message_content=InputTextMessageContent( "正确用法:输入买家用户名后加句号,例如 @buyer." ), ) ] update.inline_query.answer(results) return ConversationHandler.END # 不符合条件时结束流程 # 处理符合规则的内联查询 buyer_username = query_text.strip(".") seller_user = update.inline_query.from_user seller_username = seller_user.username seller_id = seller_user.id # 生成用户专属emoji代码,存储到上下文 emoji_code = emojihash.eh1(f"unique password{seller_username}{seller_id}", 5) emojihash_code = emojihash.eh1(f"unique code{seller_username}{seller_id}", 5) context.user_data['emoji_code'] = emoji_code context.user_data['emojihash_code'] = emojihash_code context.user_data['buyer_username'] = buyer_username # 发送带按钮的初始消息 keyboard = [ [InlineKeyboardButton(emoji_code, callback_data=str(ONE))], [InlineKeyboardButton(emojihash_code, callback_data=str(TWO))] ] reply_markup = InlineKeyboardMarkup(keyboard) context.bot.send_message( chat_id=seller_user.id, text=verified_message_1, reply_markup=reply_markup ) return FIRST # 进入第一个对话状态 def start_over(update: Update, context: CallbackContext) -> int: query = update.callback_query logger.info("User clicked button: %s", query.data) user = query.from_user # 从上下文获取用户专属数据 emoji_code = context.user_data.get('emoji_code', emojihash.eh1(f"unique password{user.username}{user.id}", 5)) emojihash_code = context.user_data.get('emojihash_code', emojihash.eh1(f"unique code{user.username}{user.id}", 5)) query.answer() keyboard = [ [InlineKeyboardButton(emoji_code, callback_data=str(ONE))], [InlineKeyboardButton(emojihash_code, callback_data=str(TWO))] ] reply_markup = InlineKeyboardMarkup(keyboard) query.edit_message_text( text=verified_message_1, reply_markup=reply_markup ) return FIRST def one(update: Update, context: CallbackContext) -> int: query = update.callback_query logger.info("User clicked button: %s", query.data) query.answer() emoji_code = context.user_data.get('emoji_code', '') emojihash_code = context.user_data.get('emojihash_code', '') keyboard = [ [InlineKeyboardButton(emoji_code, callback_data=str(THREE))], [InlineKeyboardButton(emojihash_code, callback_data=str(TWO))] ] reply_markup = InlineKeyboardMarkup(keyboard) query.edit_message_text( text=verified_message_2, reply_markup=reply_markup ) return FIRST def two(update: Update, context: CallbackContext) -> int: query = update.callback_query logger.info("User clicked button: %s", query.data) query.answer() keyboard = [ [InlineKeyboardButton("Yes", callback_data=str(ONE))], [InlineKeyboardButton("No", callback_data=str(TWO))] ] reply_markup = InlineKeyboardMarkup(keyboard) query.edit_message_text( text="你点错了代码,要不要再试一次?", reply_markup=reply_markup ) return SECOND def three(update: Update, context: CallbackContext) -> int: query = update.callback_query logger.info("User clicked button: %s", query.data) seller_username = query.from_user.username query.answer() keyboard = [ [InlineKeyboardButton(text='Yes', url=f'https://t.me/{seller_username}')], [InlineKeyboardButton("No", callback_data=str(TWO))], [InlineKeyboardButton("Read Again", callback_data=str(ONE))] ] reply_markup = InlineKeyboardMarkup(keyboard) query.edit_message_text( text="你已确认阅读上述消息,返回和卖家的聊天?", reply_markup=reply_markup ) return SECOND def end(update: Update, context: CallbackContext) -> int: query = update.callback_query logger.info("User clicked button: %s", query.data) query.answer() query.edit_message_text(text="流程已终止") return ConversationHandler.END def main() -> None: """启动机器人""" updater = Updater("你的TOKEN") dispatcher = updater.dispatcher # 配置ConversationHandler,使用InlineQueryHandler作为入口 conv_handler = ConversationHandler( entry_points=[InlineQueryHandler(inlinequery)], states={ FIRST: [ CallbackQueryHandler(one, pattern=f'^{str(ONE)}$'), CallbackQueryHandler(two, pattern=f'^{str(TWO)}$'), CallbackQueryHandler(three, pattern=f'^{str(THREE)}$'), ], SECOND: [ CallbackQueryHandler(start_over, pattern=f'^{str(ONE)}$'), CallbackQueryHandler(end, pattern=f'^{str(TWO)}$'), ], }, fallbacks=[CommandHandler('start', inlinequery)], per_user=True # 确保每个用户的对话状态独立 ) dispatcher.add_handler(conv_handler) updater.start_polling() updater.idle() if __name__ == '__main__': main()
关键修改说明
- 入口替换:将ConversationHandler的entry_points改为
InlineQueryHandler(inlinequery),支持内联查询启动对话; - 上下文存储:用
context.user_data存储用户专属的emoji代码和买家用户名,避免全局变量冲突; - 回调信息修正:所有回调处理函数从
update.callback_query获取用户信息,不再访问inline_query; - 返回值规范:修正
inlinequery函数的返回值类型,确保正确进入对话状态; - 移除重复处理器:删除独立的InlineQueryHandler,避免重复处理内联查询。
内容的提问来源于stack exchange,提问作者DzEskobar
相关产品推荐
相关产品推荐

