Ищу документацию про мут. Если человек шлет более 3 сообщений за 5 секунд, как сделать проверку таких таймингов сообщений?
2 ответа
Все что тебе понадобится так это библиотека datetime и telebot.
У меня есть одно решение мы просто будем получать дату отправки сообщений message.date()
получаем время в unix формате затем его переводим в обыкновенно(человеческое), достаем от туда секунды затем проверяем количество сообщений отправленных в определенный промежуток времени и если за пять секунд у нас более чем три сообщения то мутим его с помощью:bot.restrict_chat_member(message.chat.id, message.from_user.id, until_date= [сюда время мута])
,
вот как я вижу решение этой задачки.
-
чтобы получить секунды в рф времени можно просто написать
from datetime import datetime import pytz seconds = datetime.now(pytz.timezone('Europe/Moscow')).strftime('%S') print(seconds)
27 авг 2021 в 6:43
В aiogram решили данную задачу через добавление middleware с leaky bucket.
Готовый для использования код: https://github.com/aiogram/aiogram/blob/dev-2.x/examples/middleware_and_antiflood.py
import asyncio
from aiogram import Bot, Dispatcher, executor
from aiogram.contrib.fsm_storage.redis import RedisStorage2
from aiogram.dispatcher import DEFAULT_RATE_LIMIT
from aiogram.dispatcher.handler import CancelHandler, current_handler
from aiogram.dispatcher.middlewares import BaseMiddleware
from aiogram.types import Message
from aiogram.utils.exceptions import Throttled
TOKEN = "BOT_TOKEN_HERE"
bot = Bot(token=TOKEN)
storage = RedisStorage2()
dp = Dispatcher(bot, storage=storage)
def rate_limit(limit: int, key=None):
"""Decorator for configuring rate limit and key."""
def decorator(func):
setattr(func, "throttling_rate_limit", limit)
if key:
setattr(func, "throttling_key", key)
return func
return decorator
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 on_process_message(self, message: Message, data: dict):
handler = current_handler.get()
dispatcher = Dispatcher.get_current()
if handler:
limit = getattr(handler, "throttling_rate_limit", self.rate_limit)
key = getattr(
handler, "throttling_key", f"{self.prefix}_{handler.__name__}"
)
else:
limit = self.rate_limit
key = f"{self.prefix}_message"
try:
await dispatcher.throttle(key, rate=limit)
except Throttled as t:
await self.message_throttled(message, t)
raise CancelHandler()
async def message_throttled(self, message: Message, throttled: Throttled):
handler = current_handler.get()
dispatcher = Dispatcher.get_current()
if handler:
key = getattr(
handler, "throttling_key", f"{self.prefix}_{handler.__name__}"
)
else:
key = f"{self.prefix}_message"
delta = throttled.rate - throttled.delta
if throttled.exceeded_count <= 2:
await message.reply("Too many requests! ")
await asyncio.sleep(delta)
thr = await dispatcher.check_key(key)
if thr.exceeded_count == throttled.exceeded_count:
await message.reply("Unlocked.")
@dp.message_handler(commands=["start"])
@rate_limit(5, "start")
async def cmd_test(message: Message):
await message.reply("Test passed! You can use this command every 5 seconds.")
if __name__ == "__main__":
dp.middleware.setup(ThrottlingMiddleware())
executor.start_polling(dp)