基于aiogram的Telegram Bot用户认证实现方案咨询

核心逻辑 #

• 用户发送/start后，机器人引导选择身份（客户/员工）
• 客户身份直接解锁基础功能，员工身份需完成账号密码验证
• 用aiogram的**状态机（FSM）**管理多步交互，保证流程清晰可控

完整代码示例 #

1. 基础配置与状态定义 #

from aiogram import Bot, Dispatcher, types
from aiogram.contrib.fsm_storage.memory import MemoryStorage
from aiogram.dispatcher import FSMContext
from aiogram.dispatcher.filters.state import State, StatesGroup
from aiogram.utils import executor

# 替换为你的机器人Token
API_TOKEN = 'YOUR_TELEGRAM_BOT_TOKEN'

bot = Bot(token=API_TOKEN)
storage = MemoryStorage()
dp = Dispatcher(bot, storage=storage)

# 定义认证流程的状态
class AuthStates(StatesGroup):
    choosing_role = State()          # 选择身份阶段
    entering_emp_username = State()  # 输入员工账号阶段
    entering_emp_password = State()  # 输入员工密码阶段

2. /start命令触发初始交互 #

@dp.message_handler(commands=['start'])
async def cmd_start(message: types.Message):
    # 构建身份选择键盘
    role_keyboard = types.ReplyKeyboardMarkup(resize_keyboard=True)
    role_keyboard.add('客户', '员工')
    await message.answer('您是客户还是员工？', reply_markup=role_keyboard)
    await AuthStates.choosing_role.set()

3. 身份选择分支处理 #

@dp.message_handler(state=AuthStates.choosing_role)
async def handle_role_choice(message: types.Message, state: FSMContext):
    if message.text == '客户':
        await message.answer('欢迎您，客户！现在可以使用全部客户功能。', reply_markup=types.ReplyKeyboardRemove())
        await state.finish()
    elif message.text == '员工':
        await message.answer('请输入您的员工账号：', reply_markup=types.ReplyKeyboardRemove())
        await AuthStates.entering_emp_username.set()
    else:
        await message.answer('请选择正确的身份：客户或员工')

4. 员工账号密码验证 #

# 模拟员工数据库，实际项目替换为数据库查询逻辑
EMPLOYEE_DB = {
    'admin': 'admin123',
    'sales001': 'salespass'
}

@dp.message_handler(state=AuthStates.entering_emp_username)
async def handle_username_input(message: types.Message, state: FSMContext):
    # 暂存用户输入的账号
    await state.update_data(username=message.text)
    await message.answer('请输入您的密码：')
    await AuthStates.entering_emp_password.set()

@dp.message_handler(state=AuthStates.entering_emp_password)
async def handle_password_input(message: types.Message, state: FSMContext):
    user_data = await state.get_data()
    input_username = user_data['username']
    input_password = message.text

    if input_username in EMPLOYEE_DB and EMPLOYEE_DB[input_username] == input_password:
        await message.answer(f'验证通过！欢迎您，{input_username}，现在可以使用员工专属功能。')
        # 此处可添加：将用户ID与员工身份绑定存入数据库/缓存，用于后续权限校验
    else:
        await message.answer('账号或密码错误，请重新输入账号。')
        await AuthStates.entering_emp_username.set()
    
    await state.finish()

5. 启动机器人 #

if __name__ == '__main__':
    executor.start_polling(dp, skip_updates=True)

优化方向 #

• 用Redis存储替换MemoryStorage，支持多进程部署与状态持久化
• 添加错误次数限制，防止暴力破解
• 增加/cancel命令，允许用户随时终止认证流程
• 员工认证通过后，在数据库中标记用户身份，后续请求直接校验权限

内容的提问来源于stack exchange，提问作者Rebel In The F.D.G