时间戳

首页 > 解决方案 > TelegramBot 授权

python - TelegramBot 授权

问题描述

我正在使用 aiogram 和 SQLAlchemy 创建 TelegramBot 以与数据库交互。我想从授权开始。所有用户的电话号码都存储在 SQLite 数据库中。当用户运行 Bot 时，它应该要求用户输入电话号码。用户完成后，Bot 会检查数据库中是否存在此号码。如果是这样，用户可以继续。如果不是，Bot 应该拒绝访问。你能解释一下如何在代码中实现它吗？

标签： pythonsqlalchemyaiogram

解决方案

您可以添加列logged并创建/login phone以设置它True并/logout设置它False。然后其他命令必须检查是否logged是True或False

您还可以让它像在网页上一样，并在用户登录时保持时间，并在离开 5 分钟时阻止访问。如果剩下不到 5 分钟，那么命令可以将其更新为当前时间。

from aiogram import Bot, Dispatcher, executor, types
import os
import datetime

database = {
    1079414868: {'phone': '1234', 'logged': datetime.datetime(1970, 1, 1)}
}

# --- functions ---

five_minutes = datetime.timedelta(minutes=1)

def do_login(message):
    user = message.from_user.id
    print('[do_login] user:', user)
    
    if user not in database:
        return False
        
    user_data = database[user]

    current_time = datetime.datetime.now()

    #phone = message.get_args()
    args = message.get_args().split(' ')
    phone = args[0]
    
    print('[do_login] phone:', phone)

    # check new phone
    if phone and user_data['phone'] == phone:
        user_data['logged'] = current_time
        return True
    
    return False

def do_logout(message):
    user = message.from_user.id
    
    if user not in database:
        return True
        
    user_data = database[user]
    user_data['logged'] = datetime.datetime(1970, 1, 1)

    return True
    
def check_login(message, check_phone=False):

    user = message.from_user.id
    
    if user not in database:
        return False
        
    user_data = database[user]

    current_time = datetime.datetime.now()

    # check current session    
    if current_time - user_data['logged'] < five_minutes:
        user_data['logged'] = current_time
        return True
        
    return False
    
# --- init ---

TOKEN = os.getenv('TELEGRAM_TOKEN')

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

# --- commands ---
    
@dp.message_handler(commands=['login'])
async def login(message: types.Message):
    if do_login(message):
        await message.answer("Login: OK")
    else:
        await message.answer("Login: Problem")

@dp.message_handler(commands=['logout'])
async def logout(message: types.Message):
    if do_logout(message):
        await message.answer("Logout: OK")
    else:
        await message.answer("Logout: Problem")

@dp.message_handler(commands=['test'])
async def test(message: types.Message):

    if not check_login(message):
        await message.answer('You have to use "/login phone"')
        return
    
    await message.answer("I'm a bot, please talk to me!")

# --- start ---

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

顺便说一句：也许您甚至可以创建装饰器@check_login以使其更简单地分配给命令。或者您应该使用一些事件来自动测试每个命令的登录（除了/login）。

推荐阅读