如何让Telegram Bot的类变量为每个用户独立存储？

你当前的核心问题是每个用户的selected_criteria没有独立存储：原代码中每次用户发送/start时创建一个Handler实例，但全局注册的dp.callback_query_handler会绑定到最后创建的实例上，导致所有用户的回调都共用同一个实例的变量，出现数据串用的情况。

以下是两种可行的优化方案：

一、方案1：使用Aiogram FSM（有限状态机）管理用户状态 #

FSM是Aiogram官方推荐的用户交互流程管理工具，能天然隔离不同用户的状态和数据，适合多步骤交互场景。

完整代码示例 #

from aiogram import Bot, Dispatcher, executor, types
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
from aiogram.dispatcher.filters.state import State, StatesGroup
from aiogram.dispatcher import FSMContext
import os

# 动态生成对应步骤的键盘
def get_step_keyboard(step: int):
    select_btn = InlineKeyboardButton(text=f"Criteria{step}", callback_data=f"criteria_{step}")
    send_btn = InlineKeyboardButton(text="Selected Criteria", callback_data="send_photos")
    if step == 4:
        return InlineKeyboardMarkup().add(send_btn)
    return InlineKeyboardMarkup().add(select_btn, send_btn)

# 定义交互状态
class PhotoSelection(StatesGroup):
    waiting_criteria1 = State()
    waiting_criteria2 = State()
    waiting_criteria3 = State()
    selection_completed = State()

# 全局配置
BOT_TOKEN = "你的Bot令牌"
SOURCE_DIR = "你的照片存储文件夹路径"

bot = Bot(token=BOT_TOKEN)
dp = Dispatcher(bot)

@dp.message_handler(commands=['start', 'help'])
async def welcome(message: types.Message):
    await message.reply(f"Welcome chat_id= {message.chat.id}", reply_markup=get_step_keyboard(1))
    await PhotoSelection.waiting_criteria1.set()

@dp.callback_query_handler(state=PhotoSelection.waiting_criteria1)
async def handle_criteria1(call: types.CallbackQuery, state: FSMContext):
    if call.data == "criteria_1":
        await state.update_data(selected_criteria="1")
        await call.message.reply("Choose 2nd Criteria", reply_markup=get_step_keyboard(2))
        await PhotoSelection.waiting_criteria2.set()
    elif call.data == "send_photos":
        await send_target_photos(call.message, state)
    await call.answer()

@dp.callback_query_handler(state=PhotoSelection.waiting_criteria2)
async def handle_criteria2(call: types.CallbackQuery, state: FSMContext):
    if call.data == "criteria_2":
        user_data = await state.get_data()
        await state.update_data(selected_criteria=user_data["selected_criteria"] + "2")
        await call.message.reply("Choose 3rd Criteria", reply_markup=get_step_keyboard(3))
        await PhotoSelection.waiting_criteria3.set()
    elif call.data == "send_photos":
        await send_target_photos(call.message, state)
    await call.answer()

@dp.callback_query_handler(state=PhotoSelection.waiting_criteria3)
async def handle_criteria3(call: types.CallbackQuery, state: FSMContext):
    if call.data == "criteria_3":
        user_data = await state.get_data()
        await state.update_data(selected_criteria=user_data["selected_criteria"] + "3")
        await call.message.reply("End Selection", reply_markup=get_step_keyboard(4))
        await PhotoSelection.selection_completed.set()
    elif call.data == "send_photos":
        await send_target_photos(call.message, state)
    await call.answer()

@dp.callback_query_handler(state=PhotoSelection.selection_completed)
async def handle_completed(call: types.CallbackQuery, state: FSMContext):
    if call.data == "send_photos":
        await send_target_photos(call.message, state)
    await call.answer()

# 抽离发送照片的逻辑
async def send_target_photos(message: types.Message, state: FSMContext):
    user_data = await state.get_data()
    selected_criteria = user_data.get("selected_criteria", "")
    
    await message.reply(f"Selected criteria: {selected_criteria}")
    if not selected_criteria:
        await message.reply("No criteria selected yet!")
        return
    
    try:
        with os.scandir(SOURCE_DIR) as entries:
            for entry in entries:
                if entry.is_file() and entry.name.startswith(selected_criteria):
                    with open(entry.path, "rb") as photo_file:
                        await message.answer_photo(photo=photo_file)
    except Exception as e:
        await message.reply(f"Error loading photos: {str(e)}")

if __name__ == "__main__":
    executor.start_polling(dp)

二、方案2：用全局字典存储用户独立数据 #

如果不想引入FSM，可以用全局字典以用户ID为键，存储每个用户的筛选条件，实现数据隔离。

完整代码示例 #

from aiogram import Bot, Dispatcher, executor, types
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
import os

# 全局字典：key为用户chat_id，value为该用户的筛选条件
user_selections = {}

# 定义固定键盘
btn_criteria1 = InlineKeyboardButton(text="Criteria1", callback_data="criteria_1")
btn_criteria2 = InlineKeyboardButton(text="Criteria2", callback_data="criteria_2")
btn_criteria3 = InlineKeyboardButton(text="Criteria3", callback_data="criteria_3")
btn_send = InlineKeyboardButton(text="Selected Criteria", callback_data="send_photos")

keyboard_step1 = InlineKeyboardMarkup().add(btn_criteria1, btn_send)
keyboard_step2 = InlineKeyboardMarkup().add(btn_criteria2, btn_send)
keyboard_step3 = InlineKeyboardMarkup().add(btn_criteria3, btn_send)
keyboard_done = InlineKeyboardMarkup().add(btn_send)

# 全局配置
BOT_TOKEN = "你的Bot令牌"
SOURCE_DIR = "你的照片存储文件夹路径"

bot = Bot(token=BOT_TOKEN)
dp = Dispatcher(bot)

@dp.message_handler(commands=['start', 'help'])
async def welcome(message: types.Message):
    # 初始化当前用户的筛选条件
    user_selections[message.chat.id] = ""
    await message.reply(f"Welcome chat_id= {message.chat.id}", reply_markup=keyboard_step1)

@dp.callback_query_handler(text=["criteria_1", "criteria_2", "criteria_3", "send_photos"])
async def handle_callback(call: types.CallbackQuery):
    chat_id = call.message.chat.id
    # 确保用户有初始化记录
    if chat_id not in user_selections:
        user_selections[chat_id] = ""

    if call.data == "criteria_1":
        user_selections[chat_id] = "1"
        await call.message.reply("Choose 2nd Criteria", reply_markup=keyboard_step2)
    elif call.data == "criteria_2":
        user_selections[chat_id] += "2"
        await call.message.reply("Choose 3rd Criteria", reply_markup=keyboard_step3)
    elif call.data == "criteria_3":
        user_selections[chat_id] += "3"
        await call.message.reply("End Selection", reply_markup=keyboard_done)
    elif call.data == "send_photos":
        selected_criteria = user_selections[chat_id]
        await call.message.reply(selected_criteria)
        try:
            with os.scandir(SOURCE_DIR) as entries:
                for entry in entries:
                    if entry.is_file() and entry.name.startswith(selected_criteria):
                        with open(entry.path, "rb") as photo_file:
                            await call.message.answer_photo(photo=photo_file)
        except Exception as e:
            await call.message.reply(f"Error loading photos: {str(e)}")

    await call.answer()

if __name__ == "__main__":
    executor.start_polling(dp)

原代码的其他问题修复说明 #

1. 移除了原代码中错误的await handler.criteria_selection()调用（该方法需要call参数，直接调用会报错）；
2. 将source_dir改为全局配置变量，无需每个用户实例存储；
3. 优化了callback_data的命名，使其更直观，便于后续维护。

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