Так как Вы часто создаете и удаляете токены, то лучше всех их хранить структурировано в базе данных. Поэтому я предлагаю Вам использовать стандартный модуль shelve
(рус.), или же sqlite3
, или же mongodb
.
Связь в таблицах будет между токеном и временем его жизни (Временная метка - timestamp). Мы можем использовать токен, если время больше, чем текущее время, иначе удаляем токен из базы данных. Функция delete_old_tokens
удаляет устаревшие токены каждые 10 минут ссылка. И чтобы не попасться на устаревший токен, то мы при работе с токеном проверяем его функцией is_valid_token
.
Также для генерации токенов лучше использовать стандартный модуль secrets
(функция generate_token
).
Ещё я заметил, что Вы проверяете: может ли пользователь воспользоваться командой. Это можно вынести в отдельный декоратор @only_kings_id
(по id) или @only_kings_role
(по роли) ссылка. Если пользователь не прошел проверку, то декоратор поднимает ошибку NoKingPermissions
, которую можно перехватить или в on_message_error
, или в @<ваша-функция>.error
. ссылка
Предлагаю использовать такой код:
config.py
ds_token = 'ваш-токен'
owner_ids = {ваш-id}
kings_ids = {id-королей}
king_role_id = id-роли-для-королей
bot_prefix = '!' # префикс для команд
tokens = './tokens' # путь до базы данных (для shelve и sqlite3)
index.py
(Версия с shelve.)
from __future__ import annotations
from discord.ext import commands
from discord.ext import tasks
import secrets
import config
import string
import shelve
import time
bot = commands.Bot(config.bot_prefix, owner_ids=config.owner_ids)
@bot.event
async def on_ready():
print('Bot is ready!')
def convert_to_seconds(pattern: str) -> int:
seconds_per_unit = {"s": 1, "m": 60, "h": 3600, "d": 86400, "w": 604800}
try:
return int(pattern[:-1]) * seconds_per_unit[pattern[-1].lower()]
except (ValueError, TypeError, KeyError):
return 0
def generate_token(size: int) -> str:
pattern = string.ascii_letters + string.digits + '.-'
return ''.join(secrets.choice(pattern) for _ in range(size))
def is_valid_token(token: str) -> bool:
pattern = set(string.ascii_letters + string.digits + '.-')
if set(token) - pattern:
return False
with shelve.open(config.tokens) as tokens:
if token not in tokens:
return False
elif tokens[token] < int(time.time()): # type: ignore # compare valid time and current time
del tokens[token] # delete old token
return False
else:
return True
# https://discordpy.readthedocs.io/en/stable/ext/tasks/index.html
@tasks.loop(minutes=10) # every 10 minutes execute this function
async def delete_old_tokens():
with shelve.open(config.tokens) as tokens:
for token, timeout in tokens:
if timeout < int(time.time()):
del tokens[token]
# https://discordpy.readthedocs.io/en/rewrite/ext/commands/commands.html#checks
class NoKingPermissions(commands.CheckFailure): ...
def only_kings_id():
async def predicate(ctx) -> bool:
if ctx.author.id not in config.kings_ids:
raise NoKingPermissions
return True
return commands.check(predicate)
# or
def only_kings_role():
return commands.has_role(config.king_role_id)
@bot.command()
@only_kings_id() # or @only_kings_role()
async def create_token(ctx, timeout: convert_to_seconds): # type: ignore
if timeout <= 0:
return await ctx.send(f'Упс, укажи правильно время!')
with shelve.open(config.tokens) as tokens:
valid_time = int(time.time() + timeout)
tokens[token := generate_token(18)] = valid_time
await ctx.send(token)
# https://discordpy.readthedocs.io/en/rewrite/ext/commands/commands.html#error-handling
@create_token.error
async def create_token_error(ctx, error):
if isinstance(error, commands.CheckFailure):
return await ctx.send('Я не подчинюсь таким как ты!')
raise
@bot.command()
@only_kings_id() # or @only_kings_role()
async def use_token(ctx, token: str):
if not is_valid_token(token):
return await ctx.send(f'Токен устарел или не существует!')
await ctx.send('Работа с токеном...')
...
bot.run(config.ds_token, reconnect=True)
delete_old_tokens.start()
index.py
(Версия с sqlite3.)
from __future__ import annotations
from discord.ext import commands
from discord.ext import tasks
import sqlite3
import secrets
import config
import string
bot = commands.Bot(config.bot_prefix, owner_ids=config.owner_ids)
cxn = sqlite3.connect(config.tokens + '.db', check_same_thread=False)
cur = cxn.cursor()
@bot.event
async def on_ready():
cur.executescript(
'''
CREATE TABLE IF NOT EXISTS tokens (
token VARCHAR(18) PRIMARY KEY,
valid_time TIMESTAMP
);
DELETE FROM tokens WHERE valid_time < CURRENT_TIMESTAMP;
'''
)
cxn.commit()
print('Bot is ready!')
def convert_to_seconds(pattern: str) -> str | None:
try:
units = {'s': 'SECOND', 'm': 'MINUTE', 'h': 'HOUR', 'd': 'DAY', 'w': 'WEEK'}
if (number := int(pattern[:-1])) > 0 :
return f'{number} {units[pattern[-1]]}'
return None
except (ValueError, TypeError, KeyError):
return None
def generate_token(size: int) -> str:
pattern = string.ascii_letters + string.digits + '.-'
return ''.join(secrets.choice(pattern) for _ in range(size))
def is_valid_token(token: str) -> bool:
pattern = set(string.ascii_letters + string.digits + '.-')
if set(token) - pattern:
return False
cur.execute('DELETE FROM tokens WHERE valid_time < CURRENT_TIMESTAMP') # delete old tokens
cxn.commit()
cur.execute('SELECT 1 FROM tokens WHERE token == ?', [token])
return cur.fetchone() is not None
# https://discordpy.readthedocs.io/en/stable/ext/tasks/index.html
@tasks.loop(minutes=10) # every 10 minutes execute this function
async def delete_old_tokens():
cur.execute('DELETE FROM tokens WHERE valid_time < CURRENT_TIMESTAMP')
cxn.commit()
# https://discordpy.readthedocs.io/en/rewrite/ext/commands/commands.html#checks
class NoKingPermissions(commands.CheckFailure): ...
def only_kings_id():
async def predicate(ctx) -> bool:
if ctx.author.id not in config.kings_ids:
raise NoKingPermissions
return True
return commands.check(predicate)
# or
def only_kings_role():
return commands.has_role(config.king_role_id)
@bot.command()
@only_kings_id() # or @only_kings_role()
async def create_token(ctx, timeout: convert_to_seconds): # type: ignore
if timeout is None:
return await ctx.send(f'Упс, укажи правильно время!')
cur.execute(
'INSERT INTO tokens VALUES (?, datetime(CURRENT_TIMESTAMP, ?))',
[token := generate_token(18), timeout]
)
cxn.commit()
await ctx.send(token)
# https://discordpy.readthedocs.io/en/rewrite/ext/commands/commands.html#error-handling
@create_token.error
async def create_token_error(ctx, error):
if isinstance(error, commands.CheckFailure):
return await ctx.send('Я не подчинюсь таким как ты!')
raise
@bot.command()
@only_kings_id() # or @only_kings_role()
async def use_token(ctx, token: str):
if not is_valid_token(token):
return await ctx.send(f'Токен устарел или не существует!')
await ctx.send('Работа с токеном...')
...
bot.run(config.ds_token, reconnect=True)
delete_old_tokens.start()