1

Ищу документацию про мут. Если человек шлет более 3 сообщений за 5 секунд, как сделать проверку таких таймингов сообщений?

2 ответа 2

1

Все что тебе понадобится так это библиотека datetime и telebot. У меня есть одно решение мы просто будем получать дату отправки сообщений message.date() получаем время в unix формате затем его переводим в обыкновенно(человеческое), достаем от туда секунды затем проверяем количество сообщений отправленных в определенный промежуток времени и если за пять секунд у нас более чем три сообщения то мутим его с помощью:bot.restrict_chat_member(message.chat.id, message.from_user.id, until_date= [сюда время мута]), вот как я вижу решение этой задачки.

1
  • чтобы получить секунды в рф времени можно просто написать from datetime import datetime import pytz seconds = datetime.now(pytz.timezone('Europe/Moscow')).strftime('%S') print(seconds) 27 авг 2021 в 6:43
0

В 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)

0

Ваш ответ

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

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.