Как мне не допустить множественное нажатие на кнопку в телеграмм боте на питоне? (Aiogram)
Пишу новостного телеграмм бота на питоне. Пользователь выбирает сайт, а затем выбирает номер новости. Бот предлагает на выбор 3 сайта, при нажатии на кнопку, бот начинает парсить сайт. Если же пользователь многократно нажмёт на кнопку, то бот ломается. Ниже приведён пример той самой кнопки, не судите строго, первый раз пишу на питоне)
@dp.message_handler(Text(equals="SecurityLab"))
async def News(message: types.Message):
await message.reply("Сейчас что-нибудь поищу.", reply_markup=types.ReplyKeyboardRemove())
global browser
browser = webdriver.Chrome()
SecurityLab_News.parsing_Secure_lab(browser)
n = 1
for i in range(0, len(SecurityLab_News.news_title) - 11):
await message.answer(str(n) + ')' + ' ' + SecurityLab_News.news_title[i].text)
n += 1
keyboard = types.ReplyKeyboardMarkup(resize_keyboard=True)
keyboard.add(*buttons_Seclab)
await message.answer("Введите номер новости.", reply_markup=keyboard)
Как мне избежать многократного нажатия на кнопку в aiogram?
Ответы (1 шт):
Автор решения: Сергей Солдатов
→ Ссылка
class ThrottlingMiddleware(BaseMiddleware):
def __init__(self, limit=DEFAULT_RATE_LIMIT, key_prefix='antiflood_'):
self.rate_limit = limit
self.prefix = key_prefix
super(ThrottlingMiddleware, self).__init__()
async def throttle(self, target: Union[types.Message, types.CallbackQuery]):
handler = current_handler.get()
dispatcher = Dispatcher.get_current()
if not handler:
return
limit = getattr(handler, 'throttling_rate_limit', self.rate_limit)
key = getattr(handler, 'throttling_key', f"{self.prefix}_{handler.__name__}")
try:
await dispatcher.throttle(key, rate=limit)
except Throttled as t:
await self.target_throttled(target, t, dispatcher, key)
raise CancelHandler()
@staticmethod
async def target_throttled(target: Union[types.Message, types.CallbackQuery],
throttled: Throttled, dispatcher: Dispatcher, key: str):
msg = target.message if isinstance(target, types.CallbackQuery) else target
delta = throttled.rate - throttled.delta
await asyncio.sleep(delta)
if throttled.exceeded_count == 2 :
await msg.reply('Слишком Часто! Давай не так быстро')
return
elif throttled.exceeded_count == 3:
await msg.reply(f'⚠ Всё. Больше не отвечу, пока не пройдет {round(delta, 3)} секунд')
return
thr = await dispatcher.check_key(key)
if thr.exceeded_count == throttled.exceeded_count:
await msg.reply("⚠ Все, теперь отвечаю.")
async def on_process_message(self, message, data):
await self.throttle(message)
async def on_process_callback_query(self, call, data):
await self.throttle(call)
def rate_limit(limit: int, key=None):
def decorator(func):
setattr(func, "throttling_rate_limit", limit)
if key:
setattr(func, "throttling_key", key)
return func
return decorator
Должно работать, просто устанавливая декоратор над функцией и передавая сколько секунд нужно подождать, если нужна более точная настройка, можно передавать в декоратор второй параметр key='your_param' и уже в функции target_throttled строить свою логику
@rate_limit(limit=5)
@dp.message_handler(Text(equals="SecurityLab"))
async def News(message: types.Message):
await message.reply("Сейчас что-нибудь поищу.", reply_markup=types.ReplyKeyboardRemove())
global browser
browser = webdriver.Chrome()
SecurityLab_News.parsing_Secure_lab(browser)
n = 1
for i in range(0, len(SecurityLab_News.news_title) - 11):
await message.answer(str(n) + ')' + ' ' + SecurityLab_News.news_title[i].text)
n += 1
keyboard = types.ReplyKeyboardMarkup(resize_keyboard=True)
keyboard.add(*buttons_Seclab)
await message.answer("Введите номер новости.", reply_markup=keyboard)