构建Telegram多级按钮菜单与状态管理教程

本教程旨在指导开发者如何使用 `python-telegram-bot` 库创建具有多级交互式按钮菜单的telegram机器人，并有效管理用户会话状态，特别适用于如费用追踪等需要引导用户完成多步操作的场景。核心内容将聚焦于 `conversationhandler` 的应用，以实现流畅、逻辑清晰的用户交互流程。

引言：构建交互式Telegram机器人的挑战

在开发Telegram机器人时，尤其当需要引导用户完成一系列选择或输入时，创建一个直观且响应迅速的用户界面至关重要。例如，一个费用追踪机器人可能需要用户首先选择“收入”或“支出”，然后进入二级分类（如“工资”、“餐饮”），再到三级分类（如“基本工资”、“早餐”），最后输入金额和描述。这种多级选择的交互模式，如果处理不当，很容易导致代码混乱、状态管理困难，甚至用户体验不佳。

原有的实现尝试通过全局变量和手动判断 callback_data 来管理流程，但随着交互深度的增加，这种方法变得难以维护且容易出错，尤其是在处理并发用户请求时。核心问题在于缺乏一种机制来追踪特定用户在对话中的当前“状态”。

解决方案：使用 ConversationHandler 进行状态管理

python-telegram-bot 库提供了一个强大的工具 ConversationHandler，它专门用于处理有状态的、多步的对话流程，也被称为有限状态机（FSM）。ConversationHandler 允许开发者定义一系列“状态”，并为每个状态指定相应的处理器，从而清晰地管理用户在对话中的每一步。

ConversationHandler 的核心概念

Entry Points (入口点): 启动对话的触发器，通常是命令（如 /start）。States (状态): 对话中的不同阶段，每个状态都有其对应的处理器来响应用户输入。状态可以是自定义的常量（例如 SELECT_LEVEL1, SELECT_LEVEL2, ENTER_AMOUNT_DESCRIPTION）。Handlers (处理器): 针对特定更新类型（如 CommandHandler, CallbackQueryHandler, MessageHandler）在特定状态下执行的函数。Fallbacks (回退点): 当用户输入不符合任何当前状态的预期时，用于结束或重置对话的处理器，例如 /cancel 命令。

改造费用追踪机器人

我们将使用 ConversationHandler 来重构费用追踪机器人，使其能够顺畅地引导用户完成三级分类选择，并最终记录金额和描述。

1. 定义对话状态

首先，定义机器人可能处于的各个状态。这些状态将指导 ConversationHandler 如何响应用户的输入。

表单大师AI

一款基于自然语言处理技术的智能在线表单创建工具，可以帮助用户快速、高效地生成各类专业表单。

74 查看详情

# 定义对话状态SELECT_LEVEL1, SELECT_LEVEL2, SELECT_LEVEL3, ENTER_AMOUNT_DESCRIPTION = range(4)

2. 数据结构优化

为了更好地与多级菜单交互，原始的类别数据结构需要优化。我们可以将其转换为一个嵌套的字典或列表，以便于查找子类别。

# 假设 categories_data 是从 Google Sheet 获取的原始数据# 优化后的类别结构示例# {#     "Income": {#         "id": "1",#         "subcategories": {#             "Sueldo": {#                 "id": "101",#                 "subcategories": {#                     "Salario": {"id": "1011"},#                     "Propinas": {"id": "1012"},#                     # ...#                 }#             },#             # ...#         }#     },#     "Expense": {#         # ...#     }# }def build_nested_categories(raw_data):    nested_categories = {}    for item in raw_data:        l1_name = item.get("level1")        l2_name = item.get("level2")        l3_name = item.get("level3")        item_id = str(item.get("id"))        if l1_name and not l2_name and not l3_name: # Level 1 category            if l1_name not in nested_categories:                nested_categories[l1_name] = {"id": item_id, "subcategories": {}}        elif l2_name and not l3_name: # Level 2 category            for l1_key in nested_categories:                if l1_key == l1_name: # Find its parent                    if l2_name not in nested_categories[l1_key]["subcategories"]:                        nested_categories[l1_key]["subcategories"][l2_name] = {"id": item_id, "subcategories": {}}                    break            else: # If no explicit L1 parent in data, try to infer or handle                # This part might need more robust logic if L1 is not always explicit                pass        elif l3_name: # Level 3 category            for l1_key in nested_categories:                if l1_key == l1_name:                    for l2_key in nested_categories[l1_key]["subcategories"]:                        if l2_key == l2_name:                            nested_categories[l1_key]["subcategories"][l2_key]["subcategories"][l3_name] = {"id": item_id}                            break                    break    return nested_categories# 假设 categories_data 已经从 Google Sheet 读取# nested_categories = build_nested_categories(categories_data)# 为了简化示例，我们假设 nested_categories 是一个全局变量或通过 context 传递

注意： 原始 categories 列表的构建方式是线性的，不利于按层级检索。上述 build_nested_categories 示例展示了如何将其转换为嵌套结构，这对于 ConversationHandler 中的层级导航至关重要。在实际应用中，您需要根据 Google Sheet 的实际结构调整构建逻辑。

3. 编写状态处理器函数

每个状态都需要一个或多个处理器函数来生成按钮、响应用户点击并推进对话到下一个状态。

# 假设 `nested_categories` 已经从 Google Sheet 加载并处理成嵌套结构# 全局或通过 context 传递，此处简化为全局# 例如：# nested_categories = {#     "Income": {#         "id": "1",#         "subcategories": {#             "Sueldo": {"id": "101", "subcategories": {"Salario": {"id": "1011"}, "Propinas": {"id": "1012"}}},#             "Otro Ingreso": {"id": "102", "subcategories": {"Transferencia de ahorros": {"id": "1021"}}}#         }#     },#     "Expense": {#         "id": "2",#         "subcategories": {#             "Diarios": {"id": "201", "subcategories": {"Comida": {"id": "2011"}, "Restaurantes": {"id": "2012"}}},#             "Vivienda": {"id": "202", "subcategories": {"Renta": {"id": "2021"}}}#         }#     }# }async def start(update, context):    """开始对话，显示一级分类按钮"""    keyboard = []    for category_name, category_data in nested_categories.items():        # callback_data 格式: "level1_{category_name}"        keyboard.append([InlineKeyboardButton(category_name, callback_data=f"level1_{category_name}")])    reply_markup = InlineKeyboardMarkup(keyboard)    await update.message.reply_text("请选择一个一级分类:", reply_markup=reply_markup)    return SELECT_LEVEL1 # 返回下一个状态async def select_level1(update, context):    """处理一级分类选择，显示二级分类按钮"""    query = update.callback_query    await query.answer() # 必须回答回调查询    data_parts = query.data.split('_')    selected_l1_name = data_parts[1]    # 存储用户选择到 context.user_data    context.user_data['level1'] = selected_l1_name    l1_category = nested_categories.get(selected_l1_name)    if not l1_category or not l1_category.get("subcategories"):        await query.edit_message_text("此分类下无子分类。请重新开始。", reply_markup=None)        return ConversationHandler.END # 结束对话    keyboard = []    for l2_name, l2_data in l1_category["subcategories"].items():        # callback_data 格式: "level2_{l1_name}_{l2_name}"        keyboard.append([InlineKeyboardButton(l2_name, callback_data=f"level2_{selected_l1_name}_{l2_name}")])    reply_markup = InlineKeyboardMarkup(keyboard)    await query.edit_message_text(f"您选择了 '{selected_l1_name}'。请选择一个二级分类:", reply_markup=reply_markup)    return SELECT_LEVEL2 # 返回下一个状态async def select_level2(update, context):    """处理二级分类选择，显示三级分类按钮"""    query = update.callback_query    await query.answer()    data_parts = query.data.split('_')    selected_l1_name = data_parts[1]    selected_l2_name = data_parts[2]    context.user_data['level2'] = selected_l2_name    l1_category = nested_categories.get(selected_l1_name)    l2_category = l1_category["subcategories"].get(selected_l2_name)    if not l2_category or not l2_category.get("subcategories"):        await query.edit_message_text("此分类下无三级分类。请提供金额和描述。", reply_markup=None)        await query.message.reply_text("请输入金额和描述（例如：100 晚餐）。")        return ENTER_AMOUNT_DESCRIPTION # 如果没有三级分类，直接进入金额描述阶段    keyboard = []    for l3_name, l3_data in l2_category["subcategories"].items():        # callback_data 格式: "level3_{l1_name}_{l2_name}_{l3_name}"        keyboard.append([InlineKeyboardButton(l3_name, callback_data=f"level3_{selected_l1_name}_{selected_l2_name}_{l3_name}")])    reply_markup = InlineKeyboardMarkup(keyboard)    await query.edit_message_text(f"您选择了 '{selected_l2_name}'。请选择一个三级分类:", reply_markup=reply_markup)    return SELECT_LEVEL3 # 返回下一个状态async def select_level3(update, context):    """处理三级分类选择，并请求金额和描述"""    query = update.callback_query    await query.answer()    data_parts = query.data.split('_')    # selected_l1_name = data_parts[1] # 此时不再需要，已在 user_data 中    # selected_l2_name = data_parts[2]    selected_l3_name = data_parts[3]    context.user_data['level3'] = selected_l3_name    await query.edit_message_text(f"您选择了 '{selected_l3_name}'。")    await query.message.reply_text("请输入金额和描述（例如：100 晚餐）。")    return ENTER_AMOUNT_DESCRIPTION # 返回下一个状态async def enter_amount_description(update, context):    """处理金额和描述输入，并记录数据"""    text_input = update.message.text    # 简单的解析金额和描述，实际应用中可能需要更复杂的正则或验证    try:        parts = text_input.split(' ', 1)        amount = float(parts[0])        description = parts[1] if len(parts) > 1 else ""    except (ValueError, IndexError):        await update.message.reply_text("输入格式不正确。请重新输入金额和描述（例如：100 晚餐）。")        return ENTER_AMOUNT_DESCRIPTION # 停留在当前状态，等待正确输入    context.user_data['amount'] = amount    context.user_data['description'] = description    # 假设 sheetIn 是 GSpread 工作表对象    # 记录数据到 Google Sheets    # 注意：这里需要确保 context.user_data 包含所有需要记录的字段    # 例如：sheetIn.append_row([context.user_data.get('level1'), context.user_data.get('level2'), context.user_data.get('level3'), amount, description])    await update.message.reply_text("记录成功！对话结束。")    return ConversationHandler.END # 结束对话async def cancel(update, context):    """取消对话"""    await update.message.reply_text("对话已取消。")    return ConversationHandler.END # 结束对话async def unknown_command(update, context):    """处理未知命令"""    await update.message.reply_text("抱歉，我无法识别此命令。")async def handle_invalid_input(update, context):    """处理无效的按钮点击或消息，防止卡顿"""    await update.callback_query.answer()    await update.callback_query.edit_message_text("无效操作或超时，请重新开始 `/start`。")    return ConversationHandler.END

4. 配置 ConversationHandler

在 main 函数中，创建 ConversationHandler 实例并将其添加到 Application。

from telegram.ext import Application, CommandHandler, CallbackQueryHandler, MessageHandler, filters, ConversationHandlerfrom telegram import InlineKeyboardButton, InlineKeyboardMarkupimport asyncioimport loggingimport gspreadfrom oauth2client.service_account import ServiceAccountCredentials# 配置日志logging.basicConfig(    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO)logger = logging.getLogger(__name__)# Telegram bot tokenTELEGRAM_BOT_TOKEN = 'YOUR_TELEGRAM_BOT_TOKEN' # 替换为你的Bot Token# Google Sheets credentialsGOOGLE_SHEET_ID = 'YOUR_GOOGLE_SHEET_ID' # 替换为你的Google Sheet IDSHEET_NAMEIn = 'MySheetAnswers'SHEET_NAME = 'MyCategoryList'SCOPE = ['https://spreadsheets.google.com/feeds', 'https://www.googleapis.com/auth/drive']CREDS_JSON = 'path/to/your/credentials.json' # 替换为你的JSON凭证文件路径# Authenticate with Google Sheetstry:    creds = ServiceAccountCredentials.from_json_keyfile_name(CREDS_JSON, SCOPE)    client = gspread.authorize(creds)    sheetIn = client.open_by_key(GOOGLE_SHEET_ID).worksheet(SHEET_NAMEIn) # 用于记录答案    sheet = client.open_by_key(GOOGLE_SHEET_ID).worksheet(SHEET_NAME) # 用于读取分类    # Fetch categories from the Google Sheet    categories_data = sheet.get_all_records()    # 构建嵌套类别结构    nested_categories = {}    for category in categories_data:        level1 = category.get("level1")        level2 = category.get("level2")        level3 = category.get("level3")        item_id = str(category.get("id"))        if level1 and not level2 and not level3:            if level1 not in nested_categories:                nested_categories[level1] = {"id": item_id, "subcategories": {}}        elif level2 and not level3:            # 查找或创建一级分类            l1_parent_name = next((c.get("level1") for c in categories_data if c.get("id") == int(item_id[:1]) and c.get("level1")), None)            if l1_parent_name and l1_parent_name in nested_categories:                if level2 not in nested_categories[l1_parent_name]["subcategories"]:                    nested_categories[l1_parent_name]["subcategories"][level2] = {"id": item_id, "subcategories": {}}        elif level3:            # 查找或创建二级分类            l1_parent_name = next((c.get("level1") for c in categories_data if c.get("id") == int(item_id[:1]) and c.get("level1")), None)            l2_parent_name = next((c.get("level2") for c in categories_data if c.get("id") == int(item_id[:3]) and c.get("level2")), None)            if l1_parent_name and l2_parent_name and                l1_parent_name in nested_categories and                l2_parent_name in nested_categories[l1_parent_name]["subcategories"]:                nested_categories[l1_parent_name]["subcategories"][l2_parent_name]["subcategories"][level3] = {"id": item_id}    logger.info("Categories loaded and nested structure built.")except Exception as e:    logger.error(f"Error authenticating with Google Sheets or loading categories: {e}")    # 在生产环境中，可能需要更优雅的错误处理，例如机器人无法启动或发送错误消息# 定义对话状态SELECT_LEVEL1, SELECT_LEVEL2, SELECT_LEVEL3, ENTER_AMOUNT_DESCRIPTION = range(4)async def start(update, context):    """开始对话，显示一级分类按钮"""    keyboard = []    # 确保 nested_categories 是一个字典，且包含有效的键    if not nested_categories:        await update.message.reply_text("抱歉，未能加载分类数据。请稍后再试。")        return ConversationHandler.END    for category_name in nested_categories.keys():        # callback_data 格式: "level1_{category_name}"        keyboard.append([InlineKeyboardButton(category_name, callback_data=f"level1_{category_name}")])    reply_markup = InlineKeyboardMarkup(keyboard)    await update.message.reply_text("欢迎！让我们开始记录您的费用。请选择一个一级分类:", reply_markup=reply_markup)    return SELECT_LEVEL1 # 返回下一个状态async def select_level1(update, context):    """处理一级分类选择，显示二级分类按钮"""    query = update.callback_query    await query.answer()    data_parts = query.data.split('_')    selected_l1_name = data_parts[1]    context.user_data['level1'] = selected_l1_name    l1_category = nested_categories.get(selected_l1_name)    if not l1_category or not l1_category.get("subcategories"):        await query.edit_message_text(f"'{selected_l1_name}' 下无子分类。请提供金额和描述。", reply_markup=None)        await query.message.reply_text("请输入金额和描述（例如：100 晚餐）。")        return ENTER_AMOUNT_DESCRIPTION    keyboard = []    for l2_name in l1_category["subcategories"].keys():        # callback_data 格式: "level2_{l1_name}_{l2_name}"        keyboard.append([InlineKeyboardButton(l2_name, callback_data=f"level2_{selected_l1_name}_{l2_name}")])    reply_markup = InlineKeyboardMarkup(keyboard)    await query.edit_message_text(f"您选择了 '{selected_l1_name}'。请选择一个二级分类:", reply_markup=reply_markup)    return SELECT_LEVEL2async def select_level2(update, context):    """处理二级分类选择，显示三级分类按钮"""    query = update.callback_query    await query.answer()    data_parts = query.

以上就是构建Telegram多级按钮菜单与状态管理教程的详细内容，更多请关注创想鸟其它相关文章！