Продвинутые обработчики и фильтры

Проблема: Вы хотите, чтобы команда /admin работала только для админов, сообщения с фото обрабатывались отдельно от текста, а команда /sum 1 2 3 автоматически суммировала числа. Без фильтров придётся писать проверки в каждом обработчике — код раздувается, логика дублируется, тестировать невозможно.

Решение: Фильтры выносят условия обработки в декларативную форму. Обработчик получает только те сообщения, которые ему нужны. Это делает код чище, упрощает тестирование и позволяет переиспользовать логику.

#Обработчики: основы

Обработчик (handler) — асинхронная функция, которая вызывается при совпадении обновления с определёнными условиями.

💡 Зачем это нужно: Обработчики — это точки входа для вашей логики. Каждый сценарий (команда, кнопка, файл) имеет свой обработчик. Правильное разделение обработчиков упрощает поддержку: изменили одну команду — не сломали другие.

from aiogram import Router, F
from aiogram.types import Message
from aiogram.filters import Command

router = Router()


@router.message(Command('start'))
async def cmd_start(msg: Message):
    """
    Обработчик команды /start.
    
    Зачем: Это первая точка контакта с пользователем. 
    Здесь можно проверить, зарегистрирован ли пользователь, 
    и показать приветственное сообщение.
    """
    await msg.answer('Привет! Я бот.')

⚠️ Важно: Имя функции (cmd_start) не влияет на логику, но помогает при отладке и в логах. Используйте префиксы: cmd_ для команд, cb_ для callback, msg_ для сообщений.

#Фильтры

Фильтр определяет, должен ли обработчик реагировать на обновление.

🔗 Связь с другими темами: Кастомные фильтры выносятся в модуль bot/filters/ и используются вместе с Middleware для сквозной логики (например, проверка прав доступа).

#Встроенные фильтры: когда какие использовать

Сценарий	Фильтр	Пример использования
Пользователь ввёл команду	`Command`	`/start`, `/help`, `/admin`
Нажал кнопку в меню	`Text`	"Меню", "Профиль", "Настройки"
Прислал фото/документ	`F.photo`, `F.document`	Загрузка аватарки, импорта данных
Нужно проверить состояние	`StateFilter`	Диалог регистрации, оформления заказа
Только для определённого чата	`ChatTypeFilter`	Админ-команды только в личных сообщениях
Сложное условие	Magic Filter `F`	Комбинация нескольких проверок

#Command и аргументы

Проблема: Пользователь вводит /sum 1 2 3 4, и вы хотите получить сумму чисел. Как извлечь аргументы команды?

Решение: Используйте CommandObject для доступа к аргументам и magic для валидации.

from aiogram.filters import Command, CommandObject


@router.message(Command('echo', magic=F.args.func(lambda x: len(x) > 0)))
async def cmd_echo(msg: Message, command: CommandObject):
    """
    Эхо-команда с аргументами.
    
    Зачем: Проверка наличия аргументов до вызова обработчика.
    Если аргументов нет — обработчик не сработает.
    """
    # command.args — аргументы после команды
    await msg.answer(f'Вы ввели: {command.args}')


@router.message(Command('sum', magic=F.args.func(lambda x: x.strip())))
async def cmd_sum(msg: Message, command: CommandObject):
    """
    Суммирует числа: /sum 1 2 3 4
    
    Зачем: Автоматическая валидация входных данных.
    Фильтр отсечёт команды без аргументов.
    """
    try:
        numbers = list(map(int, command.args.split()))
        result = sum(numbers)
        await msg.answer(f'Сумма: {result}')
    except ValueError:
        await msg.answer('Пожалуйста, введите числа')

⚠️ Антипаттерн: Не делайте сложную валидацию внутри обработчика. Если аргументы обязательны — используйте magic для проверки до вызова функции.

#Фильтр Text

Проблема: Пользователь нажал кнопку "Меню" в reply-клавиатуре. Как отличить это от обычного сообщения с текстом "Меню"?

Решение: Фильтр Text точно совпадает с текстом кнопки.

from aiogram.filters import Text


@router.message(Text(text='Меню', ignore_case=True))
async def menu_handler(msg: Message):
    """
    Обработчик нажатия кнопки 'Меню'.
    
    Зачем: ignore_case=True позволяет ловить 'меню', 'МЕНЮ', 'Меню'.
    Полезно, когда пользователь вводит текст вручную.
    """
    await msg.answer('Открываю меню...')


@router.message(Text(startswith='Настройка'))
async def settings_handler(msg: Message):
    """
    Обработчик для раздела настроек.
    
    Зачем: startswith позволяет ловить 'Настройка профиля', 
    'Настройка уведомлений' и т.д. одним фильтром.
    """
    await msg.answer('Раздел настроек')


@router.message(Text(endswith='помощь'))
async def help_handler(msg: Message):
    """
    Обработчик помощи.
    
    Зачем: endswith ловит 'Мне нужна помощь', 'Вызвать помощь'.
    """
    await msg.answer('Чем помочь?')

💡 Совет: Для inline-кнопок используйте callback_data вместо Text — это надёжнее (см. тему "Inline-клавиатуры").

#Magic Filter (F)

Проблема: Нужно проверить, что сообщение от админа, содержит текст больше 10 символов и отправлено в будний день. Писать три проверки в обработчике?

Решение: Magic Filter позволяет комбинировать условия декларативно.

#Базовый синтаксис

from aiogram import F
from aiogram.types import Message


# Проверка атрибута
@router.message(F.text == 'Привет')
async def hello_handler(msg: Message):
    """
    Зачем: Точное совпадение текста.
    Когда: Для команд меню, ответов на вопросы.
    """
    await msg.answer('Привет!')


# Проверка вложенного атрибута
@router.message(F.from_user.id == 123456789)
async def admin_handler(msg: Message):
    """
    Зачем: Проверка конкретного пользователя.
    Когда: Для админ-команд, whitelist-доступа.
    
    ⚠️ Антипаттерн: Не хардкодьте ID в фильтрах. 
    Используйте кастомный фильтр is_admin (см. ниже).
    """
    await msg.answer('Привет, админ!')


# Проверка наличия атрибута
@router.message(F.photo)
async def photo_handler(msg: Message):
    """
    Зачем: Обработка только фото.
    Когда: Загрузка аватарок, модерация контента.
    """
    await msg.answer('Вижу фото!')


# Отрицание
@router.message(~F.text.startswith('/'))
async def not_command_handler(msg: Message):
    """
    Зачем: Игнорировать команды.
    Когда: Обработка обычных сообщений в чате.
    """
    await msg.answer('Это не команда')

#Операторы Magic Filter

from aiogram import F


# Логические операторы
@router.message(F.text == 'Да' | F.text == 'Нет')
async def yes_no_handler(msg: Message):
    """
    Зачем: Один обработчик для нескольких вариантов.
    Когда: Подтверждение/отмена, да/нет.
    """
    await msg.answer('Выбор сделан')


# Комбинирование
@router.message(F.text.len() > 5 & F.text.len() < 100)
async def medium_text_handler(msg: Message):
    """
    Зачем: Ограничение длины текста.
    Когда: Валидация отзывов, комментариев.
    
    ⚠️ Антипаттерн: Не используйте сложные условия 
    в декораторе — вынесите в кастомный фильтр.
    """
    await msg.answer('Текст средней длины')


# Методы строк
@router.message(F.text.startswith('/'))
@router.message(F.text.endswith('?'))
@router.message(F.text.contains('спасибо'))
async def text_handlers(msg: Message):
    """
    Зачем: Быстрая проверка содержимого.
    """
    await msg.answer('Получено')


# Методы списков
@router.message(F.animation.func(lambda x: x.file_size < 1024 * 1024))
async def small_gif_handler(msg: Message):
    """
    Зачем: Ограничение размера файла.
    Когда: Экономия трафика, места на диске.
    """
    await msg.answer('Лёгкий GIF')


# Доступ по индексу
@router.message(F.entities[0].type == 'bot_command')
async def command_entity_handler(msg: Message):
    """
    Зачем: Проверка типа entity.
    Когда: Обработка команд в тексте.
    
    ⚠️ Антипаттерн: Доступ по индексу [0] упадёт, 
    если entities пуст. Используйте F.entities.first()
    или проверку len(F.entities) > 0.
    """
    await msg.answer('Обнаружена команда')

#Кастомные функции в F

from aiogram import F
from datetime import datetime


def is_weekend(dt: datetime) -> bool:
    """Проверяет, выходной ли день."""
    return dt.weekday() >= 5


@router.message(F.date.func(is_weekend))
async def weekend_handler(msg: Message):
    """
    Зачем: Разная логика для будней/выходных.
    Когда: Рабочий бот, уведомления только в рабочее время.
    """
    await msg.answer('Сейчас выходной!')


# Лямбда-функции
@router.message(F.text.func(lambda x: x.isdigit()))
async def digit_handler(msg: Message):
    """
    Зачем: Проверка формата.
    Когда: Ввод возраста, количества, кода.
    """
    await msg.answer(f'Число: {msg.text}')

💡 Совет: Если функция используется в нескольких местах — вынесите в кастомный фильтр-класс.

#Кастомные фильтры

Проблема: Вы проверяете user_id in admin_ids в пяти обработчиках. Изменили список админов — правите пять файлов.

Решение: Кастомный фильтр инкапсулирует логику проверки. Изменили в одном месте — работает везде.

#Фильтр-функция

📁 Структура проекта: Сохраняйте фильтры в bot/filters/ для переиспользования.

# bot/filters/is_admin.py
from aiogram.types import Message, CallbackQuery
from bot.config import settings


async def admin_filter(msg: Message | CallbackQuery) -> bool:
    """
    Проверяет, является ли пользователь админом.
    
    Зачем: Централизованная проверка прав.
    Когда: Админ-команды, доступ к статистике, модерация.
    """
    user_id = (msg.from_user.id if hasattr(msg, 'from_user')
               else msg.from_user.id)
    return user_id in settings.admin_ids


# Использование
@router.message(admin_filter)
async def admin_handler(msg: Message):
    await msg.answer('Привет, админ!')

⚠️ Антипаттерн: Не вызывайте admin_filter() вручную в обработчике. Используйте как декоратор — aiogram сам вызовет фильтр.

#Фильтр-класс

# bot/filters/is_subscribed.py
from aiogram import BaseFilter
from aiogram.types import Message, CallbackQuery
from bot.services.api_client import APIClient


class IsSubscribedFilter(BaseFilter):
    """
    Проверяет подписку пользователя на канал.
    
    Зачем: Ограничение доступа для неподписанных.
    Когда: Контент только для подписчиков, закрытые клубы.
    
    🔗 Связь с другими темами: Для массовой проверки используйте 
    кэширование (Redis) для снижения нагрузки на API Telegram.
    """

    def __init__(self, channel_id: int):
        self.channel_id = channel_id

    async def __call__(self, event: Message | CallbackQuery) -> bool:
        user_id = event.from_user.id
        # Проверка подписки через get_chat_member
        return await self.check_subscription(user_id)

    async def check_subscription(self, user_id: int) -> bool:
        # Логика проверки (через бота или API)
        ...


# Использование
@router.message(IsSubscribedFilter(channel_id=-1001234567890))
async def subscribed_handler(msg: Message):
    await msg.answer('Спасибо за подписку!')

#Фильтр с зависимостями

🔗 Связь с другими темами: Этот фильтр использует Dependency Injection для доступа к БД. См. тему "Архитектурные паттерны".

# bot/filters/is_premium_user.py
from aiogram import BaseFilter
from aiogram.types import Message
from sqlalchemy.ext.asyncio import AsyncSession
from bot.repositories.user import UserRepository


class IsPremiumUser(BaseFilter):
    """
    Проверяет, есть ли у пользователя премиум-подписка.
    
    Зачем: Разделение функционала для платных/бесплатных пользователей.
    Когда: Premium-контент, приоритетная поддержка, расширенные функции.
    """

    def __init__(self, required: bool = True):
        self.required = required

    async def __call__(
        self,
        msg: Message,
        session: AsyncSession,
        user_repo: UserRepository
    ) -> bool:
        user = await user_repo.get_by_telegram_id(msg.from_user.id)
        is_premium = user and user.is_premium

        return is_premium if self.required else not is_premium


# Использование
@router.message(IsPremiumUser())
async def premium_handler(msg: Message):
    await msg.answer('Премиум-контент')


@router.message(IsPremiumUser(required=False))
async def non_premium_handler(msg: Message):
    await msg.answer('Доступно только для премиум')

#Приоритеты обработчиков

⚠️ Критично важно: Это одна из самых частых ошибок новичков.

Проблема: Вы зарегистрировали общий обработчик всех сообщений, а потом — обработчик команды /start. Команда не работает.

Почему: aiogram обрабатывает обработчики в порядке регистрации. Первый совпавший обработчик выполняется, остальные игнорируются.

#Правильный порядок регистрации

# ❌ НЕПРАВИЛЬНО: общий обработчик перехватит все сообщения
@router.message()
async def catch_all(msg: Message):
    await msg.answer('Получено')


@router.message(Command('start'))
async def cmd_start(msg: Message):
    await msg.answer('Start')  # Никогда не вызовется!


# ✅ ПРАВИЛЬНО: специфичные обработчики раньше общих
@router.message(Command('start'))
async def cmd_start(msg: Message):
    await msg.answer('Start')


@router.message(Command('help'))
async def cmd_help(msg: Message):
    await msg.answer('Help')


@router.message()  # Общий обработчик в конце
async def catch_all(msg: Message):
    await msg.answer('Получено')

💡 Правило: От специфичного к общему. Сначала команды, потом текст, потом catch-all.

#Явное управление приоритетами через роутеры

Проблема: У вас 50 обработчиков в одном файле. Порядок регистрации легко нарушить.

Решение: Группируйте обработчики по роутерам и регистрируйте роутеры в нужном порядке.

# bot/dispatcher.py
from bot.handlers import high_priority, medium_priority, low_priority

dp = Dispatcher()

# Регистрируем роутеры в порядке приоритета
dp.include_router(high_priority.router)    # Команды, callback
dp.include_router(medium_priority.router)  # Текстовые сообщения
dp.include_router(low_priority.router)     # Catch-all

🔗 Связь с другими темами: Роутеры — часть модульной архитектуры. См. тему "Архитектурные паттерны".

#Флаги (Flags)

Проблема: Нужно передать метаданные в обработчик: требует ли он авторизации, какой rate limit применять. Писать отдельные фильтры для каждого случая?

Решение: Флаги — механизм передачи метаданных через декоратор.

#Объявление и использование флагов

from aiogram import Router, F
from aiogram.types import Message
from aiogram.dispatcher.flags import get_flag

router = Router()


@router.message(F.text == 'admin', flags={'require_admin': True})
async def admin_only(msg: Message):
    """
    Зачем: Флаг указывает, что обработчик требует прав админа.
    
    ⚠️ Антипаттерн: Не полагайтесь только на флаги. 
    Флаг — это метаданные, а не защита. 
    Проверку прав делайте в middleware или обработчике.
    """
    # Извлекаем флаг внутри обработчика
    require_admin = get_flag(msg, 'require_admin', default=False)

    if require_admin:
        # Проверка прав
        if not is_admin(msg.from_user.id):
            return

    await msg.answer('Админ-панель')

#Middleware для обработки флагов

🔗 Связь с другими темами: Middleware — сквозной механизм. См. тему "Архитектурные паттерны".

# bot/middleware/flags.py
from typing import Callable, Dict, Any
from aiogram import BaseMiddleware
from aiogram.types import Update
from aiogram.dispatcher.flags import get_flag


class FlagMiddleware(BaseMiddleware):
    """
    Обрабатывает флаги перед вызовом обработчика.
    
    Зачем: Централизованная обработка метаданных.
    Когда: Аутентификация, rate limiting, логирование.
    """

    async def __call__(
        self,
        handler: Callable,
        event: Update,
        data: Dict[str, Any]
    ) -> Any:
        # Извлекаем флаги из контекста
        require_auth = get_flag(event, 'require_auth', default=False)
        rate_limit = get_flag(event, 'rate_limit', default=10)

        if require_auth:
            user = data.get('user')
            if not user:
                return  # Прерываем, если нет авторизации

        # Передаем флаги в data
        data['flags'] = {
            'require_auth': require_auth,
            'rate_limit': rate_limit
        }

        return await handler(event, data)


# Использование
@router.message(
    F.text == 'profile',
    flags={'require_auth': True, 'rate_limit': 5}
)
async def profile_handler(msg: Message):
    flags = msg.flags  # Или из контекста
    await msg.answer('Профиль')

#Цепочка фильтров

Проблема: Нужно, чтобы обработчик срабатывал только для команды /register, от пользователя с языком 'ru' и не в состоянии формы.

Решение: Несколько фильтров в декораторе комбинируются через логическое И.

from aiogram import F
from aiogram.filters import Command, StateFilter
from aiogram.fsm.state import State, StatesGroup


class Form(StatesGroup):
    name = State()
    age = State()


# Все фильтры должны совпасть
@router.message(
    Command('register'),
    F.from_user.language_code == 'ru',
    ~StateFilter(Form.name)  # Не в состоянии Form.name
)
async def cmd_register(msg: Message):
    """
    Зачем: Комбинация условий для точного контроля.
    
    Когда: 
    - Command: только для команды
    - language_code: локализация
    - ~StateFilter: игнорировать, если пользователь в форме
    """
    await msg.answer('Начинаем регистрацию')

⚠️ Антипаттерн: Не перегружайте декоратор условиями. Если фильтров больше 3-4 — вынесите в кастомный фильтр-класс.

#Фильтрация по типу контента

from aiogram import F
from aiogram.types import ContentType


@router.message(F.content_type == ContentType.PHOTO)
async def photo_handler(msg: Message):
    """
    Зачем: Обработка только фото.
    Когда: Загрузка аватарок, модерация изображений.
    """
    photo = msg.photo[-1]  # Лучшее качество
    await msg.answer(f'Фото получено, file_id: {photo.file_id}')


@router.message(F.content_type == ContentType.DOCUMENT)
async def doc_handler(msg: Message):
    """
    Зачем: Обработка документов.
    Когда: Импорт данных, загрузка файлов.
    """
    await msg.answer(f'Документ: {msg.document.file_name}')


@router.message(F.content_type == ContentType.VOICE)
async def voice_handler(msg: Message):
    """
    Зачем: Обработка голосовых.
    Когда: Транскрибация, голосовые команды.
    """
    await msg.answer('Голосовое сообщение получено')


# Любой медиа-контент
@router.message(F.photo | F.document | F.video | F.audio)
async def media_handler(msg: Message):
    """
    Зачем: Обработка любого медиа.
    Когда: Универсальный загрузчик файлов.
    """
    await msg.answer('Медиа получено')

#Фильтрация по времени

from aiogram import F
from datetime import datetime, time


async def is_business_hours(msg: Message) -> bool:
    """
    Проверяет, рабочее ли время (9:00 - 18:00).
    
    Зачем: Ограничение доступа по времени.
    Когда: Рабочий бот, уведомления только в рабочее время.
    
    ⚠️ Антипаттерн: Не используйте datetime.now() 
    без учёта часового пояса пользователя.
    """
    now = datetime.now().time()
    return time(9, 0) <= now <= time(18, 0)


async def is_weekday(msg: Message) -> bool:
    """Проверяет, будний ли день."""
    return datetime.now().weekday() < 5


@router.message(is_business_hours & is_weekday)
async def business_hours_handler(msg: Message):
    await msg.answer('Мы работаем!')


@router.message(~is_business_hours | ~is_weekday)
async def off_hours_handler(msg: Message):
    await msg.answer('Мы сейчас не работаем. Оставьте сообщение.')

💡 Совет: Для работы с часовыми поясами используйте pytz или zoneinfo.

#Комбинирование фильтров в классе

Проблема: Нужно проверить длину текста, язык пользователя, наличие премиума и время отправки. Четыре отдельных фильтра загромождают код.

Решение: Один фильтр-класс с параметрами.

# bot/filters/complex_filter.py
from aiogram import BaseFilter, F
from aiogram.types import Message
from datetime import datetime


class ComplexFilter(BaseFilter):
    """
    Сложный фильтр с несколькими условиями.
    
    Зачем: Инкапсуляция сложной логики проверки.
    Когда: Многофакторная валидация сообщений.
    """

    def __init__(
        self,
        min_text_length: int = 1,
        allowed_languages: list[str] | None = None,
        require_premium: bool = False
    ):
        self.min_text_length = min_text_length
        self.allowed_languages = allowed_languages or ['ru', 'en']
        self.require_premium = require_premium

    async def __call__(self, msg: Message) -> bool:
        # Проверка длины текста
        if not msg.text or len(msg.text) < self.min_text_length:
            return False

        # Проверка языка
        if msg.from_user.language_code not in self.allowed_languages:
            return False

        # Проверка премиума
        if self.require_premium and not msg.from_user.is_premium:
            return False

        # Проверка времени (не ночью)
        hour = datetime.now().hour
        if hour < 6 or hour > 23:
            return False

        return True


# Использование
@router.message(ComplexFilter(
    min_text_length=5,
    allowed_languages=['ru'],
    require_premium=False
))
async def filtered_handler(msg: Message):
    await msg.answer('Сообщение прошло все фильтры!')

#Отладка фильтров

import logging

logger = logging.getLogger(__name__)


async def debug_filter(msg: Message) -> bool:
    """
    Фильтр для отладки — логирует все сообщения.
    
    Зачем: Понимать, какие сообщения получает бот.
    Когда: Отладка проблем с обработкой.
    """
    logger.debug(
        f"Message from {msg.from_user.username}: {msg.text}"
    )
    return True


@router.message(debug_filter)
async def debug_handler(msg: Message):
    await msg.answer('Сообщение залогировано')

💡 Совет: Включите debug-логирование только в development. В production это создаст лишнюю нагрузку.

#Итоги

#✅ Чек-лист: готово ли ваше решение к продакшену?

Обработчики зарегистрированы в правильном порядке (специфичные → общие)
Кастомные фильтры вынесены в отдельный модуль bot/filters/
Magic Filter используются для сложных условий
Флаги применяются для метаданных (rate limit, require_auth)
Написаны unit-тесты на кастомные фильтры
В обработчиках нет сложной валидации — она вынесена в фильтры

#🔗 Связь с другими темами

Тема	Как связана
Архитектура	Фильтры выносятся в `bot/filters/`, используются в middleware
FSM	StateFilter проверяет текущее состояние диалога
Безопасность	Фильтры прав доступа (is_admin, is_subscribed)
Тестирование	Unit-тесты на кастомные фильтры

#📚 Что дальше?

→ Middleware — сквозная логика для всех обработчиков
→ Архитектурные паттерны — модульная структура проекта
→ Тестирование — проверка корректности фильтров

Сценарий

Фильтр

Пример использования

Пользователь ввёл команду

Command

/start, /help, /admin

Нажал кнопку в меню

Text

"Меню", "Профиль", "Настройки"

Прислал фото/документ

F.photo, F.document

Загрузка аватарки, импорта данных

Нужно проверить состояние

StateFilter

Диалог регистрации, оформления заказа

Только для определённого чата

ChatTypeFilter

Админ-команды только в личных сообщениях

Сложное условие

Magic Filter F

Комбинация нескольких проверок

Тема

Как связана

Архитектура

Фильтры выносятся в bot/filters/, используются в middleware

FSM

StateFilter проверяет текущее состояние диалога

Безопасность

Фильтры прав доступа (is_admin, is_subscribed)

Тестирование

Unit-тесты на кастомные фильтры