构建Telegram多级按钮菜单与状态管理教程-Python教程-PHP中文网

构建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

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

查看详情

# 定义对话状态
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, ConversationHandler
from telegram import InlineKeyboardButton, InlineKeyboardMarkup
import asyncio
import logging
import gspread
from 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 token
TELEGRAM_BOT_TOKEN = 'YOUR_TELEGRAM_BOT_TOKEN' # 替换为你的Bot Token

# Google Sheets credentials
GOOGLE_SHEET_ID = 'YOUR_GOOGLE_SHEET_ID' # 替换为你的Google Sheet ID
SHEET_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 Sheets
try:
    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_LEVEL2

async def select_level2(update, context):
    """处理二级分类选择，显示三级分类按钮"""
    query = update.callback_query
    await query.answer()

    data_parts = query.

登录后复制

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