Telegram 机器人跟踪费用 - 如何创建按钮和子菜单按钮

Question

我正在使用Python Telegram Bot

我想创建一个机器人来跟踪我的开支，该机器人将从我的 Google 表格中读取类别，并在另一个 Google 表格中跟踪用户的响应。该类别有 3 个级别：level1（收入、支出）； 2 级（工资、奖金等）和 3 级（A、B、C）。 我希望机器人以按钮的形式向我显示类别，用户单击 level1：收入，然后机器人向我显示收入下 level2 中的所有子类别，然后单击子类别，机器人向我显示 level2 下的所有 level3 类别。 机器人将记录所有答案，然后询问金额和简短描述（字符串）。

我开发了代码，但我一直在尝试让机器人在用户单击 level1（收入或费用按钮）后向我显示 level2 类别。我知道鳕鱼知道我是点击“收入”还是“支出”，但我无法让机器人根据我点击的 level1 按钮向我显示下一组按钮。

我尝试了很多选择，但不知道现在该去哪里，还尝试了 ChatGPT 的一些建议，但没有成功

这是代码：

import asyncio
import logging
import telegram
from telegram.constants import ParseMode
from telegram.ext import Application, MessageHandler, filters, CallbackQueryHandler, CommandHandler
from telegram import InlineKeyboardButton, InlineKeyboardMarkup

import gspread
from oauth2client.service_account import ServiceAccountCredentials

loop = asyncio.get_event_loop()

# Telegram bot token
TELEGRAM_BOT_TOKEN = 'myToken'
bot = telegram.Bot(token=TELEGRAM_BOT_TOKEN)

# Google Sheets credentials
GOOGLE_SHEET_ID = 'MyGoogleSheetID'
SHEET_NAMEIn = 'MySheetAnswers'
SHEET_NAME = 'MyCategoryList'
SCOPE = ['https://spreadsheets.google.com/feeds', 'https://www.googleapis.com/auth/drive']
CREDS_JSON = 'MyPathToJsonFile'  

# Authenticate with Google Sheets
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()

# Define categories globally
categories = []
for category in categories_data:
    level1 = category.get("level1", "")
    level2 = category.get("level2", "")
    level3 = category.get("level3", "")

    # Find or create level 1
    parent_level1 = next((c for c in categories if c["name"] == level1), None)
    if not parent_level1:
        parent_level1 = {"id": str(len(categories) + 1), "name": level1, "subcategories": []}
        categories.append(parent_level1)

    # Find or create level 2 under level 1
    parent_level2 = next((c for c in parent_level1["subcategories"] if c["name"] == level2), None)
    if not parent_level2:
        parent_level2 = {"id": str(len(categories) + 1), "name": level2, "subcategories": []}
        parent_level1["subcategories"].append(parent_level2)

    # Create level 3 under level 2
    level3_category = {"id": str(len(categories) + 1), "name": level3, "subcategories": []}
    parent_level2["subcategories"].append(level3_category)

# Dictionary to store user responses
user_responses = {}

# Function to start the conversation
async def start(update, context):
    await update.message.reply_text("Welcome! Let's get started.")
    await ask_category(update, context, categories)

# Function to ask for category
async def ask_category(update, context, categories):
    # Access the global categories variable
    # global categories

    # Create keyboard with categories
    keyboard = [[InlineKeyboardButton(category["name"], callback_data=f'category_{category["id"]}') for category in categories]]
    reply_markup = InlineKeyboardMarkup(keyboard)

    # Check if the update object and its message attribute are not None
    if update and update.message:
        await update.message.reply_text("Please select a category:", reply_markup=reply_markup)
    else:
        print("Update or message is None. Could not send the reply1.")
        # You can log the error, send a default message, or take other actions as needed

# Function to ask for subcategory
async def ask_subcategory(update, context, category_id):
    # Access the global categories variable
    global categories
    
    category = next((c for c in categories if c["id"] == category_id), None)
    if category:
        # Create keyboard with subcategories
        subcategory_buttons = [
            InlineKeyboardButton(subcategory["name"], callback_data=f'subcategory_{category_id}_{i+1}') 
            for i, subcategory in enumerate(category["subcategories"])
        ]
        reply_markup = InlineKeyboardMarkup([subcategory_buttons])

        # Check if the update object and its message attribute are not None
        if update and update.message:
            await update.message.reply_text(f"Please select a subcategory for {category['name']}:", reply_markup=reply_markup)
        else:
            print("Update or message is None. Could not send the reply2.")
            # You can log the error, send a default message, or take other actions as needed

# Function to ask for level3 subcategory
async def ask_level3_subcategory(update, context, category_id, subcategory_index):
    # Access the global categories variable
    global categories
    
    category = next((c for c in categories if c["id"] == category_id), None)
    if category and subcategory_index <= len(category["subcategories"]):
        selected_subcategory = category["subcategories"][subcategory_index - 1]
        # Create keyboard with level3 subcategories
        level3_buttons = [
            InlineKeyboardButton(level3_subcategory["name"], callback_data=f'level3_subcategory_{category_id}_{subcategory_index}_{i+1}') 
            for i, level3_subcategory in enumerate(selected_subcategory.get("subcategories", []))
        ]
        reply_markup = InlineKeyboardMarkup([level3_buttons])
        await update.message.reply_text(f"Please select a level3 subcategory for {selected_subcategory['name']}:", reply_markup=reply_markup)

# Function to ask for amount and description
async def ask_amount_and_description(update):
    await update.message.reply_text("Great! Now, please provide the amount and description.")

# New navigate_categories function
async def navigate_categories(update, context, category_id, level):
    global categories

    # Find the selected category object
    selected_category = next((c for c in categories if c["id"] == category_id and c["name"]), None)
    if selected_category:
        # Check if the update object and its message attribute are not None
        if update and update.message:
            print("Debug: Sending reply.")
            # Branch based on level
            if level == 1:
                # Get and display level 2 categories
                subcategory_buttons = [
                    InlineKeyboardButton(subcategory["name"], callback_data=f'subcategory_{category_id}_{i+1}') 
                    for i, subcategory in enumerate(selected_category["subcategories"])
                ]
                reply_markup = InlineKeyboardMarkup([subcategory_buttons])
                await update.message.reply_text(f"Please select a subcategory for {selected_category['name']}:", reply_markup=reply_markup)
                # Update user data and call the function again for level 2
                context.user_data["current_category"] = selected_category
                await navigate_categories(update, context, subcategory_buttons[0].callback_data.split("_")[1], 2)

            elif level == 2:
                # Get and display level 3 categories
                level3_buttons = [
                    InlineKeyboardButton(level3_subcategory["name"], callback_data=f'level3_subcategory_{category_id}_{i+1}') 
                    for i, level3_subcategory in enumerate(selected_category["subcategories"])
                ]
                reply_markup = InlineKeyboardMarkup([level3_buttons])
                await update.message.reply_text(f"Please select a third-level subcategory for {selected_category['name']}:", reply_markup=reply_markup)
                # Update user data and call the function again for level 3
                context.user_data["current_category"] = selected_category
                await navigate_categories(update, context, level3_buttons[0].callback_data.split("_")[1], 3)

            elif level == 3:
                # Get and display level 4 categories (if any)
                level4_buttons = [
                    InlineKeyboardButton(level4_subcategory["name"], callback_data=f'level4_subcategory_{category_id}_{i+1}') 
                    for i, level4_subcategory in enumerate(selected_category["subcategories"])
                ]
                reply_markup = InlineKeyboardMarkup([level4_buttons])
                await update.message.reply_text(f"Please select a fourth-level subcategory for {selected_category['name']}:", reply_markup=reply_markup)
                # Update user data and call the function again for level 4
                context.user_data["current_category"] = selected_category
                await navigate_categories(update, context, level4_buttons[0].callback_data.split("_")[1], 4)
        else:
            print("Debug: Update or message is None. Could not send the reply.")
            # You can log the error, send a default message, or take other actions as needed
    else:
        print("Debug: Selected category is None or has an empty name. Could not send the reply.")
        # You can log the error, send a default message, or take other actions as needed

# Function to handle button clicks
async def button_click(update, context):
    query = update.callback_query
    user_id = query.from_user.id

    # Parse the callback data
    data_parts = query.data.split('_')
    selection_type, *selection_values = data_parts

    # Log user selection
    if selection_type == 'category':
        category_id = selection_values[0]
        user_responses['category'] = category_id
        print(f"Debug: Selected category ID: {category_id}")
        print(f"Debug: Categories: {categories}")
        await navigate_categories(update, context, category_id, 1)
    elif selection_type == 'subcategory':
        subcategory_index = int(selection_values[1])
        user_responses['subcategory'] = selection_values[0]
        print(f"Debug: Selected subcategory ID: {user_responses['subcategory']}")
        await navigate_categories(update, context, user_responses['subcategory'], 2)
    elif selection_type == 'level3_subcategory':
        user_responses['level3_subcategory'] = selection_values[0]
        print(f"Debug: Selected level3_subcategory ID: {user_responses['level3_subcategory']}")
        await ask_amount_and_description(update)

    # Answer callback query
    await query.answer()

# Function to handle text input
async def handle_text_input(update, context):
    # Log user text input
    user_responses['text_input'] = update.message.text

    # Log all user responses to Google Sheets
    sheetIn.append_row(list(context.user_data.values()))

    # End the conversation
    await update.message.reply_text("Thank you! Conversation completed.")

# Set up handlers
start_handler = CommandHandler('start', start)
button_handler = CallbackQueryHandler(button_click)
def text_input_handler(update, context):
    if update.message.text and not update.message.is_command():
        handle_text_input(update, context)

text_input_handler = MessageHandler(filters.Text and ~filters.COMMAND, text_input_handler)

# Create and run the application
application = Application.builder().token(TELEGRAM_BOT_TOKEN).build()
application.add_handler(start_handler)
application.add_handler(button_handler)
application.add_handler(text_input_handler)
application.run_polling()

这是我的类别列表结构：

id	1级	2级	3级
1	收入
101		苏尔多
1011			萨拉里奥
1012			普罗皮纳斯
1013			奖金
1014			委员会
1015			奥特罗斯
102		奥特罗·英格雷索
1021			恐怖转移
1022			兴趣爱好
1023			股息
1024			雷加洛斯
1025			雷姆博索斯
1026			奥特罗斯
2	费用
201		日记
2011			科米达
2012			餐厅
2013			卫生个人
2014			罗帕
2015			林皮萨产品
2016			梅林达斯
2017			奥特罗斯
202		维维安达
2021			伦塔
2022			GC
2023			阿莫布拉多
2024			装饰
2025			Mantenimiento
2026			穆丹扎
2027			奥特罗斯
203		服务
2031			蜂窝网络
2032			互联网
2033			电动
2034			阿瓜
2035			煤气
2036			奥特罗斯
...	...	...	...

Answer 1

IISC，您希望机器人跟踪当前对话所处的“状态”，即您需要使用 FSM 之类的工具对更新进行有状态处理。 PTB 为此类用例提供了

ConversationHandler

，并且有 这个示例 专门针对内联按钮。

---

免责声明：我目前是

python-telegram-bot

的维护者。