The command prefix is what the message content must contain initially to have a command invoked. This prefix could either be a string to indicate what the prefix should be, or a callable that takes in the bot as its first parameter and nextcord.Message as its second parameter and returns the prefix.
def prefix_func(bot, message) -> str:
, которая будет вызываться для каждого сообщения, и возвращать желаемую строку префикса. Это не только позволяет менять префикс динамически, не трогая бота (бот как дёргал функцию, так и будет дёргать), но и позволяет делать вещи типа "свой префикс у разных серверов".\
) в строковых литералах Питона (как и во многих языках) имеет специальное значение, зависящее от следующего символа. Например, \n
означает символ перевода строки, \t
- символ табуляции, и т.д. Последовательность \u
позволяет задать символ юникода по его коду, например, символ \u00A7
это значок параграфа (§). Но это также разделитель каталогов в пути на платформе Windows. А у тебя в пути есть последовательность "C:\Users\777\Desktop\Новая папка\67.png".\\
- она означает просто символ \
без специального поведения.my_path = r'C:\Users\'
. В таких литералах символ \
не имеет специального поведения, это просто символ./
. Винда это позволяет. p = multiprocessing.Process(target=gen,args=(lst,))
p.start()
p.join()
), когда он завершится. Это мало чем отличается от просто вызова gen() в твоём коде, безо всякого мультипроцессинга.Сам словарь полученный от телеграмм мне нужно его в нужном моменте сохранять в бд, а позже вытаскивать обратно превращая назад в dict, но все идет по наклонной
'entities': [<telebot.types.MessageEntity object at 0x7770137fd0>]
<telebot.types.MessageEntity object at 0x7770137fd0>
явно сериализации не поддаётся, и ни сохранить, ни восстановить его не выйдет. Или убирать ключ entities
, если он тебе не нужен, или разбираться, что это за объект и сериализовывать/восстанавливать его вручную. list1 = list('abcdef')
list2 = list('klmnop')
list3 = list('uvwxyz')
for item1, item2, item2 in zip(list1, list2, list3):
print(item1, item2, item2)
import itertools
list1 = list('abcdef')
list2 = list('klmnop')
list3 = list('uvwxyz')
for item in itertools.chain(list1, list2, list3):
print(item)
import itertools
list1 = list('abcdef')
list2 = list('klmnop')
list3 = list('uvwxyz')
for item1, item2, item3 in itertools.product(list1, list2, list3):
print(item1, item2, item3)
cd C:\my_project\
.venv\bin\python3.exe main.py
import typing
import asyncio
import functools
def throttled(delay: float, measure: typing.Literal['end_to_start', 'start_to_start'] = 'start_to_start'):
def decorator(actual_func: typing.Coroutine) -> typing.Coroutine:
queue = None
task = None
async def _single_query(future, args, kwargs):
try:
result = await actual_func(*args, **kwargs) # тут делаем асинхронное обращение к сервису
except BaseException as err:
future.set_exception(err) # была ошибка - теперь await future выкинет исключение
else:
future.set_result(result) # полуен результат - await future вернёт его
async def _work_loop():
nonlocal queue
nonlocal task
while True:
try:
# ждем, пока не придёт запрос, или пока не закончится таймаут
future, args, kwargs = await asyncio.wait_for(queue.get(), delay)
except asyncio.TimeoutError: # новые запросы долго не приходят, сворачиваем работу, чтобы не тратить ресурсы
queue = None
task = None
return
single_task = _single_query(future, args, kwargs)
if measure == 'start_to_start':
asyncio.create_task(single_task)
else:
await single_task
queue.task_done() # каждому успешному get() соответствует task_done()
await asyncio.sleep(delay)
@functools.wraps(actual_func)
async def query(*args, **kwargs):
nonlocal queue # обращение к переменной выше уровнем, но не глобальной
nonlocal task
future = asyncio.Future() # Future просигналит, когда наш запрос будет обслужен
if task is None: # либо это первый запрос, либо запросы долго не приходили, и мы свернули работу
queue = asyncio.Queue()
task = asyncio.create_task(_work_loop())
await queue.put((future, args, kwargs))
return await future
return query
return decorator
# delay - минимальный интервал между запросами в секундах
# measure - как мерять интервалы между запросами: начало-начало или конец-начало
@throttled(delay=5.0, measure='start_to_start')
async def my_coroutine(*args, **kwargs) -> ReturnValue:
...
зашитая в кнопку команда объектом энтри игнорируется