Как создать антиспам-систему в aiogram с использованием middleware?

Как можно реализовать антиспам систему с использованием middleware в aiogram? В интернете представлено недостаточно информации на эту тему. Хотя в документации aiogram имеется пример кода для антиспама, из-за перехода с aiogram 2 на aiogram 3 этот код утратил свою актуальность.
  • 2 ноября 2024 г. 17:32
Ответы на вопрос 1
Создание антиспам-системы в `aiogram` с использованием middleware может быть полезной для защиты вашего бота от нежелательных сообщений. Приведем пример, как это можно реализовать на основе `aiogram` версии 3.

### Основные шаги:

1. **Создание middleware**: Создайте класс-мидлвар для обработки сообщений.
2. **Хранение состояния пользователя**: Используйте структуру данных, чтобы отслеживать время отправки сообщений пользователями.
3. **Определение лимитов**: Установите лимиты на количество сообщений, которые пользователь может отправить в определенный период времени.
4. **Обработка сообщений и определение антиспама**: В мидлваре проверяйте и обрабатывайте сообщения.

### Пример кода

```python
from aiogram import Bot, Dispatcher, types
from aiogram.contrib.fsm_storage.memory import MemoryStorage
from aiogram.utils import executor
from asyncio import sleep
from datetime import datetime, timedelta

API_TOKEN = 'YOUR_API_TOKEN'

# Создаем бота и диспетчер
bot = Bot(token=API_TOKEN)
storage = MemoryStorage()
dp = Dispatcher(bot, storage=storage)

# Настройки антиспама
class AntiSpamMiddleware:
    def __init__(self, limit: int, period: int):
        self.limit = limit  # Максимальное количество сообщений
        self.period = period  # Период в секундах
        self.user_messages = {}

    async def on_pre_process_message(self, message: types.Message, data: dict):
        user_id = message.from_user.id
        now = datetime.now()

        if user_id not in self.user_messages:
            self.user_messages[user_id] = []

        # Удаляем старые сообщения
        self.user_messages[user_id] = [
            timestamp for timestamp in self.user_messages[user_id]
            if now - timestamp < timedelta(seconds=self.period)
        ]

        self.user_messages[user_id].append(now)

        # Проверяем лимит сообщений
        if len(self.user_messages[user_id]) > self.limit:
            await message.answer("Вы отправляете слишком много сообщений. Пожалуйста, подождите.")
            # Отменяем дальнейшую обработку сообщения
            return False

# Инициализация мидлвара с лимитом 5 сообщений за 10 секунд
dp.middleware.setup(AntiSpamMiddleware(limit=5, period=10))

@dp.message_handler(commands=['start', 'help'])
async def send_welcome(message: types.Message):
    await message.reply("Привет! Я бот, который защищен от спама.")

@dp.message_handler()
async def echo(message: types.Message):
    await message.answer(message.text)

if __name__ == '__main__':
    executor.start_polling(dp, skip_updates=True)
```

### Объяснение кода:

1. **Создание класса `AntiSpamMiddleware`**: Этот класс реализует логику антиспам-системы. Он использует словарь для хранения временных меток сообщений от пользователей.
   
2. **Метод `on_pre_process_message`**: Этот метод вызывается перед обработкой каждого сообщения. В нем мы обновляем список временных меток сообщений, удаляя старые записи. Если количество записей превышает заданный лимит, бот отправляет предупреждение и прекращает дальнейшую обработку сообщения.

3. **Добавление мидлвара**: Мидлвар добавляется в диспетчер с помощью метода `setup`.

4. **Обработчики сообщений**: Пример простых обработчиков для команды `/start` и других сообщений.

### Советы:
- Настройте лимиты и период в зависимости от потребностей вашего бота и целевой аудитории.
- Вы можете расширить функциональность, добавив логику для временной блокировки пользователей, которые систематически нарушают правила.

Этот пример дает основу для создания антиспам-системы, но может быть расширен или изменен в зависимости от особенностей вашего проекта.
Похожие вопросы