如何用Telethon库为Telegram Bot添加菜单及多用户数据隔离

一、存储用户选择菜单前的输入值 #

用全局字典以用户ID为键存储输入内容，Bot运行在单线程异步环境中，普通字典即可满足基础需求（若需更高并发安全性，可搭配asyncio.Lock避免读写冲突）。当用户输入值后，先将值存入字典，再发送操作菜单供用户选择。

二、确保多用户操作结果隔离 #

1. 绑定用户ID执行操作：所有操作携带用户唯一ID，返回结果仅发送给对应ID的用户；
2. 异步处理耗时任务：将同步耗时函数转为异步执行，或用asyncio.to_thread把同步逻辑放入线程池运行，避免阻塞Bot事件循环，保证多用户请求并行处理；
3. 回调事件关联用户：通过菜单回调事件获取用户ID，从状态字典中取出该用户的专属输入值，执行对应操作后返回结果。

修改后的完整代码 #

#!/usr/bin/env python3.8
import asyncio
api_id = 12345
api_hash = 'XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX'
bot_token = 'XXXXXXXXXX:XXXXXXXXXXXXXXXXXXXXXXXXXXXXXX'

from telethon.sync import TelegramClient, events, Button
from telethon import functions, types

# 存储用户输入状态：键为用户ID，值为用户输入的内容
user_inputs = {}

async def first_function(value):
    # 替换为实际耗时操作逻辑
    await asyncio.sleep(5)
    return f"操作1执行完成，输入值：{value}"

async def second_function(value):
    # 替换为实际耗时操作逻辑
    await asyncio.sleep(10)
    return f"操作2执行完成，输入值：{value}"

bot = TelegramClient('bot', api_id, api_hash).start(bot_token=bot_token)

with bot:
    # 处理用户文本输入：接收值并发送菜单
    @bot.on(events.NewMessage(pattern=r'^[^/].*'))  # 匹配非命令的普通文本
    async def handle_user_input(event):
        user_id = event.message.peer_id.user_id
        user_input = event.message.message
        # 存储当前用户的输入内容
        user_inputs[user_id] = user_input
        # 发送Inline操作菜单
        await event.reply(
            "请选择要执行的操作：",
            buttons=[
                [Button.inline("操作1", b"opt1")],
                [Button.inline("操作2", b"opt2")]
            ]
        )

    # 处理菜单回调事件
    @bot.on(events.CallbackQuery)
    async def handle_menu_select(event):
        user_id = event.peer_id.user_id
        # 检查用户是否已输入内容
        if user_id not in user_inputs:
            await event.answer("请先输入内容再选择操作！", alert=True)
            return
        
        input_value = user_inputs[user_id]
        # 根据选择的选项执行对应函数
        if event.data == b"opt1":
            result = await first_function(input_value)
        elif event.data == b"opt2":
            result = await second_function(input_value)
        else:
            result = "无效的操作选项"
        
        # 将操作结果返回给当前用户
        await event.edit(result)
        # 可选：执行完成后清除该用户的输入缓存
        del user_inputs[user_id]

    bot.run_until_disconnected()

代码说明 #

• user_inputs字典：专门存储每个用户的输入内容，利用用户ID作为唯一标识，确保数据隔离；
• 异步函数改造：将原同步函数改为async函数，模拟耗时操作，若实际业务为同步逻辑，可使用await asyncio.to_thread(your_sync_function, value)实现异步执行；
• 菜单与回调分离：用户输入文本后触发存储逻辑并发送菜单，点击菜单触发回调事件，根据回调数据执行对应操作，全程绑定用户ID；
• 结果返回：通过event.edit()更新菜单消息为操作结果，仅对当前操作的用户可见，完全避免多用户结果混淆。

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