1

есть вот такой код:

@bot.command()
async def test(ctx, src):
    author = ctx.message.author
    a = str(author.id)
    if a in kinglist:
        
        symbols = 'abcdefghijklnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890.-'
        rand = (''.join(choice(symbols) for i in range(18)))
        f = open(f'token', 'a+')
        f.write(f'{rand}\n')
        await ctx.send(f'{rand}')
        tk = rand
        def dltk():
            if tk in open(f'token').read():
                f = open("token", "r")
                old_token = f.readlines()
                f.close()

                with open("token", "w") as file_obj:
                    for token in old_token:
                        token = token.strip()
                        new_token_list = []
                        if token != tk:
                            file_obj.write(token + '\n')
        
        time = src[:-1]
        mod = src[-1]
        if time == 0:
            pass
        elif mod == 'd':
            seconds = int(time)*86400
            await asyncio.sleep(seconds)
            dltk()
            await ctx.send(f'ТОКЕН УДАЛЕН!')
        elif mod == 'h':
            seconds = int(time)*3600
            await asyncio.sleep(seconds)
            dltk()
            await ctx.send(f'ТОКЕН УДАЛЕН!')
        elif mod == 'm':
            seconds = int(time)*60
            await asyncio.sleep(seconds)
            dltk()
            await ctx.send(f'ТОКЕН УДАЛЕН!')
        elif mod == 's':
            seconds = int(time)
            await asyncio.sleep(seconds)
            dltk()
            await ctx.send(f'ТОКЕН УДАЛЕН!')
        else:
            
            await ctx.send(f'Упс, укажи правильно время!')


    else:
        await ctx.send(f'Я не подчинюсь таким как ты!')

с виду все нормально, но код работает невероятно странно, поидеи он должен создавать токен, и спустя оприделенное время удалять его, но происходит все очень странно, когда я пишу команду он не создает токен, как только время кончаеться и бот пишет токен удален, создаеться токен, и уже не удаляеться, и такое ощющение как буд-то код выполняеться снизу вверх, но добавив кучу принтов понял что код выполняеться по порядку

1
  • по началу подумал что проблема в await asyncio.sleep(), ибо работал с ней впервые, но поставив обычные слипы проблема не решилась
    – user451780
    6 фев 2022 в 12:19

2 ответа 2

2
+50
        f = open(f'token', 'a+')
        f.write(f'{rand}\n')

А где закрытие файла? Судя по остальному коду вы же умеете пользоваться with, используйте его:

        with open(f'token', 'a+') as f:
            f.write(f'{rand}\n')

Вообще, вы по ходу дела много раз открываете один и тот же файл разными способами (то через with, то без него), с разными режимами ('a+', 'r', 'w'), а закрываете не везде. Не закрытый на чтение файл, открытый через if tk in open(f'token').read(): может быть и не мешает, но я не думаю, что при открытии файла на запись у вас потом нормально файл ещё раз сможет открыться. Странно, если никаких ошибок при этом не выходит. Но в любом случае, если вы сразу не закрыли файл, а смогли открыть его ещё раз, что-то прочитали-записали, а потом уже закрыли этот же файл, открытый ранее на запись, то я не берусь вообще предсказать, что вы прочитали/записали в этот файл в итоге, потому что запись в файл буферизуется и сбрасывается буфер наверняка только в момент закрытия файла. Чтение тоже буферизуется. Чтение-запись файлов не потокобезопасны! Так что сделайте нормальное атомарное чтение и запись, так, чтобы они не могли никак пересекаться. Для начала хотя бы вынесите чтение и запись файла в отдельные функции, и оформите там открытие нормально, через with open. И вызывайте в коде эти функции, а не открывайте-закрывайте файлы "вручную". Возможно, проблема и решится.

И def dltk(): тоже вынесите наружу, функция, определённая внутри другой функции, смотрится странно и запутывает код. Если только это не декоратор, где так нужно по технологии, не делайте так лучше.

1

Так как Вы часто создаете и удаляете токены, то лучше всех их хранить структурировано в базе данных. Поэтому я предлагаю Вам использовать стандартный модуль 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()
1
  • спасибо, прислушаюсь к вашим советам
    – user451780
    9 фев 2022 в 20:07

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.