Throttling и анти-спам

Учим «Цыплёнка-помощника» не сходить с ума, когда кто-то лупит по кнопке двадцать раз в секунду: ставим вежливый «турникет» на входе, который пропускает не больше одного запроса в пару секунд.

Throttling (от англ. throttle — «придушить, ограничить») — это ограничение частоты запросов от одного пользователя: если он шлёт сообщения слишком часто, лишние мы аккуратно отбрасываем и просим не частить.

Хук: друг, который заспамил твоего бота

Представь: ты выкатил Цыплёнка в чат своего игрового клана, все радостно тыкают команду /weather. И тут самый весёлый из друзей решает поприкалываться — зажимает кнопку и отправляет /weather сорок раз за пять секунд. Что произойдёт?

Каждое его сообщение — это отдельный запрос к сервису погоды (помнишь урок про хэндлеры сообщений? там каждый хэндлер срабатывает на каждое сообщение). Сорок сообщений — сорок походов в сеть. Сервис погоды от такого может временно забанить твой бот за «слишком много запросов». А сам Telegram, если бот отвечает чересчур часто, начнёт притормаживать его ответы для всех — и обычные люди в чате будут ждать ответ по минуте.

Один шутник способен испортить бота всему клану. И это не злой умысел — иногда палец просто соскользнул, иногда интернет тормозит и человек жмёт кнопку повторно. Нам нужен «турникет» на входе: пропустил один запрос — на пару секунд закрылся, а лишние нажатия вежливо отбил. К концу урока ты напишешь такой турникет и повесишь его на своего Цыплёнка. Вот к чему придём — бот, который на спам отвечает спокойно: «Не части, подожди секунду :)» — и не падает.

Что такое middleware: вахтёр на входе

Прежде чем строить турникет, надо понять, куда его ставить. И тут нам нужна новая важная штука — middleware.

Вспомни, как устроен бот сейчас. Приходит обновление от Telegram (новое сообщение или нажатие кнопки), Dispatcher смотрит, какой хэндлер для него подходит, и вызывает этот хэндлер. Прямая дорога: обновление → хэндлер.

А теперь представь, что у бота есть «здание», и на входе в это здание сидит вахтёр. Любой посетитель (обновление) сначала проходит мимо вахтёра, и только потом попадает к нужному кабинету (хэндлеру). Вахтёр может посмотреть на посетителя и решить: пропустить дальше, развернуть на входе или что-то записать в журнал. Вот этот вахтёр и есть middleware.

Middleware — это прослойка, через которую проходят все обновления до того, как попадут в хэндлер. Она удобна, чтобы делать что-то общее для всех сообщений: вести лог, проверять права, считать статистику и — наш случай — ограничивать частоту запросов.

Главная фишка middleware: её код выполняется для каждого обновления автоматически, и тебе не надо вставлять одну и ту же проверку в каждый хэндлер вручную. Написал турникет один раз, повесил на диспетчер — и он работает для всех команд сразу. Это как поставить вахтёра на входе в здание, а не сажать отдельного охранника в каждый кабинет.

Как middleware выглядит в aiogram

В aiogram 3.x middleware — это класс с одним обязательным методом __call__. Звучит страшно, но смысл простой: aiogram вызывает твой объект как функцию и передаёт ему три вещи. Посмотри на скелет, пока без логики:

from aiogram import BaseMiddleware
from aiogram.types import TelegramObject

class MyMiddleware(BaseMiddleware):
    async def __call__(self, handler, event, data):
        # тут можно посмотреть на event и решить, пускать ли дальше
        print("Пришло обновление, проверяю...")
        # вызвать хэндлер = «пропустить посетителя дальше»
        return await handler(event, data)

Результат: сам по себе этот middleware ничего не ограничивает — он просто печатает строчку в консоль и пропускает обновление дальше к хэндлеру. Но в нём уже видны три ключевых участника. Разберём их, потому что на них держится всё:

• handler — это и есть «дверь в кабинет»: сам хэндлер, который сработает дальше. Если ты вызовешь await handler(event, data) — посетитель прошёл. Если не вызовешь — посетитель развёрнут на входе, хэндлер не сработает.
• event — само обновление (например, объект Message). Из него можно узнать, кто прислал сообщение: event.from_user.id.
• data — словарь со служебными данными, который aiogram передаёт по цепочке. Пока он нам не нужен, просто пробрасываем дальше.

Запомни главное правило: вызвал handler — пропустил, не вызвал — отбил. Вся идея throttling в том, чтобы иногда «не вызывать».

Идея throttling: турникет с задержкой

Теперь сама идея. Турникет должен помнить, когда каждый пользователь в последний раз присылал сообщение. Приходит новое сообщение — мы смотрим: «сколько секунд прошло с прошлого раза?». Прошло достаточно (например, больше секунды) — пропускаем. Прошло слишком мало — значит человек частит, и мы не пускаем его к хэндлеру.

Чтобы помнить время последнего сообщения для каждого пользователя, заведём обычный словарь: ключ — id пользователя, значение — момент времени его последнего сообщения. Время в Python удобно мерить функцией time.monotonic() — она возвращает число секунд, которое всегда только растёт.

Сниппет: считаем, частит человек или нет

Прежде чем писать middleware, прочувствуем самую сердцевину — расчёт «прошло ли достаточно времени». Вот чистый Python без всякого бота, его можно запустить и в голове проследить:

last_seen = {}      # id пользователя -> время последнего сообщения
RATE = 2.0          # минимум 2 секунды между сообщениями

def allow(user_id, now):
    last = last_seen.get(user_id)
    if last is not None and now - last < RATE:
        return False            # частит — не пускаем
    last_seen[user_id] = now    # запоминаем время
    return True

# имитируем три сообщения от пользователя 777 в моменты 0, 1, 5 секунд
print(allow(777, 0.0))
print(allow(777, 1.0))
print(allow(777, 5.0))

Вывод:

True
False
True

Смотри, как это читается. В момент 0.0 пользователь пишет впервые — last ещё None, пускаем (True) и запоминаем время. Через секунду (момент 1.0) он пишет снова, но прошла всего 1 секунда из нужных 2 — это спам, отбиваем (False), и время не обновляем. На пятой секунде уже прошло достаточно с последнего пропущенного сообщения — снова пускаем. Вот и весь мозг турникета. Осталось переселить эту логику в middleware и научить её мерить настоящее время.

Разбор на примерах

Пример 1. Простейший анти-спам middleware

Перенесём логику из сниппета в настоящий middleware и добавим его в наш bot.py. Разберём по строчкам:

import time
from aiogram import BaseMiddleware
from aiogram.types import Message

class ThrottlingMiddleware(BaseMiddleware):
    def __init__(self, rate=2.0):
        self.rate = rate          # минимум секунд между сообщениями
        self.last_seen = {}       # id пользователя -> время

    async def __call__(self, handler, event: Message, data):
        user_id = event.from_user.id
        now = time.monotonic()
        last = self.last_seen.get(user_id)

        if last is not None and now - last < self.rate:
            # человек частит — НЕ вызываем handler, просто выходим
            await event.answer("Не части, подожди секунду :)")
            return

        self.last_seen[user_id] = now
        return await handler(event, data)

# вешаем турникет на все сообщения
dp.message.middleware(ThrottlingMiddleware(rate=2.0))

Результат: если писать боту команды спокойно, раз в пару секунд, он отвечает как обычно. Но если зажать кнопку и спамить — на первое сообщение бот ответит нормально, а на каждое следующее в течение 2 секунд напишет «Не части, подожди секунду :)» и не выполнит саму команду. Разберём ключевые места:

• __init__ — настраиваем турникет при создании: задаём rate (сколько секунд держать паузу) и заводим пустой словарь last_seen.
• event.from_user.id — узнаём, кто прислал сообщение. У каждого пользователя свой счётчик: турникет считает паузу для каждого человека отдельно, а не один на всех.
• now - last < self.rate — то самое сравнение из сниппета: прошло ли достаточно времени.
• Если частит — мы вызываем event.answer(...) и делаем return, не вызывая handler. Это и есть «отбить на турникете»: команда не выполнится.
• dp.message.middleware(...) — вешаем турникет именно на обновления-сообщения. Создаём объект ThrottlingMiddleware(rate=2.0) один раз — он будет жить, пока живёт бот, и помнить времена всех пользователей.

Пример 2. Не спамим предупреждением в ответ

У первого варианта есть неприятная мелочь. Если друг лупит по кнопке двадцать раз, бот двадцать раз ответит «Не части, подожди секунду» — и теперь уже сам бот заспамит чат. Получилось, что мы боролись со спамом спамом. Исправим: предупреждение шлём только один раз за серию частых сообщений.

class ThrottlingMiddleware(BaseMiddleware):
    def __init__(self, rate=2.0):
        self.rate = rate
        self.last_seen = {}
        self.warned = set()       # кого мы уже предупредили в этой серии

    async def __call__(self, handler, event: Message, data):
        user_id = event.from_user.id
        now = time.monotonic()
        last = self.last_seen.get(user_id)

        if last is not None and now - last < self.rate:
            if user_id not in self.warned:
                await event.answer("Не части, подожди секунду :)")
                self.warned.add(user_id)   # больше не предупреждаем
            return

        # человек написал спокойно — сбрасываем «флажок предупреждения»
        self.warned.discard(user_id)
        self.last_seen[user_id] = now
        return await handler(event, data)

Результат: теперь при спаме бот ответит «Не части, подожди секунду :)» один раз, а остальные частые сообщения молча проглотит. Как только человек угомонится и напишет нормально (с паузой больше rate), флажок сбросится, и при следующем спаме его снова предупредят разок. Тихо и аккуратно.

Идея в множестве warned: туда мы кладём id того, кого уже предупредили. Пока он в множестве — лишние предупреждения не шлём. Команда self.warned.discard(user_id) убирает его оттуда, как только он написал по-человечески (discard не падает, даже если такого id там нет, в отличие от remove).

Пример 3. Разная строгость для разных команд

Не все команды одинаково «тяжёлые». Команде /weather, которая лезет в сеть, нужна строгая пауза. А обычная болтовня в чате может идти чаще. В aiogram middleware можно вешать не на весь dp.message, а точечно — но проще держать один турникет с настраиваемым rate и создавать несколько с разной строгостью. Покажем сам принцип на параметре:

# строгий турникет для тяжёлых команд: пауза 3 секунды
strict = ThrottlingMiddleware(rate=3.0)

# мягкий турникет: пауза всего 0.5 секунды
soft = ThrottlingMiddleware(rate=0.5)

# на все сообщения вешаем мягкий — он просто гасит совсем уж бешеный спам
dp.message.middleware(soft)

Результат: сообщения будут проходить через мягкий турникет — он отобьёт только совсем бешеный спам (чаще двух раз в секунду), а нормальную переписку пропустит. Главная мысль примера: rate — это ручка громкости. Маленькое число — мягко, большое — строго. Под разные задачи крутишь её по-разному, а сам код турникета не меняется. Для тяжёлых сетевых команд бери rate побольше, для лёгких — поменьше.

Частые ошибки и подводные камни

Вот на чём обычно спотыкаются, когда впервые делают анти-спам.

• Забывают return после предупреждения. Если ты написал «не части», но не сделал return и всё равно вызвал handler — команда выполнится, и весь турникет бесполезен. Правило: отбил — сразу выходи из middleware через return, не доходя до handler.
• Один счётчик на всех вместо счётчика по user_id. Если хранить одно общее «время последнего сообщения», то активный человек заблокирует бота для всех остальных: написал он — и весь чат на 2 секунды в бане. Всегда веди отдельный отсчёт для каждого event.from_user.id.
• Создают новый middleware внутри хэндлера. Объект ThrottlingMiddleware и его словарь last_seen должны жить всё время работы бота — создавай его один раз при настройке dp. Если создавать заново на каждое сообщение, словарь будет каждый раз пустым и турникет ничего не вспомнит.
• Берут time.time() вместо time.monotonic(). time.time() — это «настенные часы», их могут перевести назад (синхронизация времени, смена часового пояса), и тогда разница now - last станет отрицательной и сломает логику. time.monotonic() только растёт и для измерения промежутков надёжнее.
• Ставят паузу слишком большой. Соблазн поставить rate=10 «чтоб уж точно не спамили». Но тогда живые люди будут натыкаться на «подожди» при обычном использовании и решат, что бот сломан. Начни с 1–2 секунд и подкручивай, глядя на реальное поведение.
• Думают, что throttling защищает от всего. Турникет режет частоту от одного пользователя. Если ботом завладела толпа из ста человек, каждый по разу в секунду — это уже другая история (тут помогает кэш и оптимизация запросов). Throttling — первая линия обороны, а не броня от всего.

Мини-практика: «холодок» на тяжёлую команду

Закрепим на маленьком задании. У Цыплёнка есть команда /weather, которая ходит в сеть, — именно её обиднее всего заспамить. Сделай так, чтобы у этой команды был отдельный «холодок» (cooldown): не чаще одного запроса погоды в 5 секунд от одного человека.

Твоя задача:

1. Возьми за основу ThrottlingMiddleware из Примера 2 (с одним предупреждением).
2. Создай его экземпляр со строгим rate=5.0 и повесь на сообщения через dp.message.middleware(...).
3. Проверь сам себя: отправь боту /weather два раза подряд быстро — второй раз должен прийти ответ «Не части, подожди секунду :)», а сама погода не запроситься.
4. Подбери текст предупреждения подружелюбнее — например, «Погоду спрашиваем не чаще раза в 5 секунд, дай мне выдохнуть :)».
5. Со звёздочкой: сделай так, чтобы у администратора чата (его id ты знаешь и можешь захардкодить в список) турникет не срабатывал — то есть пропускай его сообщения мимо проверки. Подсказка: в самом начале __call__ проверь if event.from_user.id in ADMINS: return await handler(event, data).

Подсказка: тебе почти ничего не нужно дописывать — только поменять rate и текст, а для «звёздочки» добавить одну проверку в начало __call__. Если справишься — у тебя в руках универсальный турникет, который можно навесить на любую команду любого будущего бота.

Итоги и что дальше

Соберём главное:

• Middleware — это «вахтёр на входе»: прослойка, через которую проходят все обновления до хэндлера. Удобна для общих задач: логов, проверки прав и нашего анти-спама.
• В aiogram 3.x middleware — это класс-наследник BaseMiddleware с методом async def __call__(self, handler, event, data).
• Главное правило: вызвал handler — пропустил обновление, не вызвал — отбил. На этом и строится throttling.
• Throttling — ограничение частоты запросов: помним время последнего сообщения каждого пользователя и не пускаем тех, кто частит.
• Счётчик ведём отдельно для каждого user_id, время мерим через time.monotonic(), а турникет создаём один раз при настройке dp.
• Предупреждаем спамера вежливо и один раз за серию — чтобы не заспамить чат в ответ.

Теперь Цыплёнок умеет держать удар: один шутник больше не положит бота всему клану, а сервисы вроде погоды не забанят его за лавину запросов. Это важный шаг от «бота, который работает, пока его не трогают» к «боту, который выживает в настоящем чате с живыми людьми». В следующем уроке мы возьмёмся за ещё один полезный паттерн — научимся ловить ошибки во всём боте разом, чтобы он не падал ни от кривого ввода, ни от сбоя сети. Турникет ты поставил — увидимся у следующего рубежа обороны!