TelegramBot授权

Question

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

Answer 1

您可以添加列 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 除外）。

TelegramBot授权

TelegramBot authorization

python

sqlalchemy

aiogram