Фильтры для обработчиков сообщений в AIOgram 3

В процессе создания Telegram-бота с использованием библиотеки aiogram неизбежно встают два основных этапа: прописывание "точки входа" и создание обработчиков (handlers) сообщений. Однако, чтобы aiogram смог правильно направить запрос пользователя в нужный обработчик, он использует фильтры. Именно о фильтрах и их работе мы и поговорим в этом посте.

Что такое фильтры?

Фильтры, как и следует из их названия, выполняют функцию своеобразного сито: они пропускают через себя данные и проверяют, соответствуют ли они заданным условиям. Представьте себе корпоративную парковку с несколькими заездами (разные обработчики), каждый из которых предназначен для сотрудников определённого отдела (фильтр на обработчике). Когда сотрудник из отдела "B" подъезжает к парковке, он проверяет каждый заезд (фильтр), показывая свой бейдж (данные сообщения), пока нужный шлагбаум не поднимется и его не пустят на парковку.

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

О чём пойдёт речь в посте.

Прежде чем углубиться в тему фильтров, хочу сразу обозначить рамки этого поста. В aiogram обработчики могут реагировать не только на сообщения пользователей (message), но и на самые разнообразные события: нажатия inline-клавиш (callback_query), обработку платежей (pre_checkout_query), изменение статуса участников чата (chat_member), возникновение ошибок (errors) и многое другое.

С некоторыми из них мы уже знакомы по прошлым постам:

• message - Почти в каждом посте про AIOgram.
• callback_query - На стримах посвящённых боту для управления сервером.
• pre_checkout_query - В посте про подключение оплаты Telegram Stars.
• chat_member - В посте про реагирование бота на вступление пользователей в чат.

Тем не менее, в этом посте мы сосредоточимся на фильтрах для сообщений (message). Остальные фильтры будут рассмотрены в других постах.

Виды фильтров.

Фильтры в aiogram можно условно разделить на три основные категории:

• Встроенные фильтры — Эти фильтры созданы разработчиками aiogram для удобства и широкого применения. Они охватывают стандартные сценарии и могут использоваться в проектах любого уровня сложности. В большинстве своём они представлены в виде дополнений к "магическим" фильтрам.

@router.message(Command(commands="help"))

• "Магические фильтры" — Эти фильтры не столь очевидны, но иногда оказываются чрезвычайно полезными, особенно в небольших проектах, где требуется быстрая и простая настройка без создания сложной логики.

@router.message(F.text == "help")

• Собственные фильтры — Пользовательские фильтры, которые вы создаёте сами. Они необходимы в крупных проектах, где требуется тонкая настройка логики бота и возможность легко расширять функционал без значительных изменений в коде.

@router.message(HelpFilter())

Эти типы фильтров могут использоваться в одном проекте и даже комбинироваться для достижения нужной реакции бота на действия пользователя.

Регистрация обработчиков и подключение фильтров.

В AIOgram существует два основных способа регистрации обработчиков:

1. Прямой способ регистрации.

Прямой способ регистрации обработчиков предполагает их непосредственное добавление в диспетчер (Dispatcher). Этот метод прост и удобен на начальных этапах разработки, когда количество обработчиков невелико. Однако по мере роста проекта и увеличения числа обработчиков может возникнуть путаница, поскольку все они будут зарегистрированы в одном месте.

Пример:

dp = Dispatcher()

# Регистрация функции-обработчика
dp.message.register(<Функция-обработчик без ()>, <Фильтр>)

Обратите внимание на два ключевых момента:

1. Функция-обработчик указывается без вызова, то есть без круглых скобок в конце. Это важно, поскольку мы передаём саму функцию, а не результат её выполнения.
2. Если не указать фильтр при регистрации, то этот обработчик будет срабатывать на все события, связанные с сообщениями. Это может быть полезно, если нужно обрабатывать каждое сообщение от пользователя, однако в этом случае необходимо продумать "иерархию" обработчиков. Например, общий обработчик без фильтра должен быть зарегистрирован последним, иначе более специфичные обработчики с фильтрами не будут проверяться, и все сообщения будут обрабатываться только этим универсальным обработчиком.

2. Регистрация через роутеры.

Другой способ регистрации обработчиков — использование роутеров. В этом случае создаётся отдельный класс роутера, а функции-обработчики "подключаются" к нему с помощью специальных декораторов. Этот метод особенно удобен в больших проектах, поскольку позволяет логически группировать обработчики, что делает код более структурированным и легко поддерживаемым.

Пример:

# Создание роутера
router = Router()

# Подключение функции-обработчика к роутеру с указанием фильтра
@router.message(<фильтр>)
async def функция_обработчик(...):
    # Логика обработчика
    ...

# Регистрация роутера
dp = Dispatcher()
dp.include_router(router)

Встроенные фильтры.

Разработчики aiogram позаботились о том, чтобы мы имели под рукой набор встроенных фильтров, которые покрывают большинство стандартных сценариев. Мы не будем рассматривать все фильтры, так как некоторые из них довольно специфичны или рассчитаны на использование в "магических" фильтрах.

CommandStart.

Фильтр CommandStart основан на базовом фильтре Command, о котором мы поговорим чуть дальше. Как следует из названия, его основная задача — обрабатывать команду /start, которая обычно запускается при первом взаимодействии пользователя с ботом.

Команда /start вызывается, когда пользователь нажимает кнопку "Запустить" в интерфейсе бота в Telegram. Этот момент часто используется для инициализации: регистрации пользователя в системе, получения данных из базы данных или просто приветствия. Первоначальное взаимодействие с пользователем может быть уникальным для каждого проекта.

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

# Пользователь нажимает "Запустить"

# Срабатывает фильтр 👇
@router.message(CommandStart())
async def start_handler(message: Message) -> None:
# Вызывается обработчик 👆
    await message.answer(text="Добро пожаловать!")

Может возникнуть вопрос: "Зачем выделять целый класс для простой команды?" Всё не так просто, как кажется. Telegram поддерживает так называемый диплинк (deep-linking), который позволяет передавать дополнительные данные вместе со ссылкой на бота. Это могут быть ключевые слова для конкретных действий бота или идентификаторы пользователей, например, для реферальной программы. Применений у диплинков множество.

Представим, что у нас есть бот с ссылкой https://t.me/press_any_button_bot, и мы хотим, чтобы при запуске бота по этой ссылке пользователь получил персонализированное приветственное сообщение от того, кто поделился ссылкой: https://t.me/press_any_button_bot?start=Petya.

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

# Пользователь переходит по ссылке и нажимает "Запустить"

# Срабатывает фильтр 👇
@router.message(CommandStart())
async def start_handler(message: Message, command: CommandObject) -> None:
# Вызывается обработчик 👆
    # 👇 Проверяем наличие переданных аргументов
    if command.args:
        await message.answer(text=f"{command.args} передаёт привет!")
    else:
        await message.answer(text="Добро пожаловать!")

В аргументы обработчика передаётся объект command, который содержит информацию о вызванной команде и переданных аргументах. В нашем примере, при переходе по ссылке пользователь отправляет команду /start Petya, где /start — это команда, а Petya — аргумент. Мы можем получить этот аргумент через поле args у объекта command. Аргументы могут состоять из нескольких слов, разделённых, например, нижним подчёркиванием _. Чтобы разделить такие аргументы на отдельные части, можно использовать args = command.args.split("_").

Command.

Фильтр Command является основным для обработки команд в aiogram. Он предоставляет гибкость в обработке команд, позволяя задавать конкретные команды, которые нужно отфильтровать, а также настраивать параметры фильтрации.

Особенность фильтра Command заключается в его универсальности — вы можете указать любую команду, например, /menu или /get_file, и настроить поведение фильтра с помощью дополнительных аргументов.

Вот основные аргументы фильтра Command:

• commands — список строк с командами, которые будут обрабатываться фильтром. Вы можете указать одну или несколько команд.
• prefix — строка с префиксами, которые будут поддерживаться фильтром. По умолчанию это слэш /, но можно задать и свои, например, ~ или =, что позволит фильтру срабатывать на команды вроде ~menu или =menu.
• ignore_case — параметр, определяющий, будет ли фильтр игнорировать регистр команд. По умолчанию установлено значение True, что означает, что команды /menu, /mEnU и /MENU будут считаться одной и той же командой.

Эти настройки позволяют создавать гибкие и настраиваемые команды, которые можно использовать в различных ситуациях. Рассмотрим несколько примеров.

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

# Пользователь выполняет команду /menu
# Срабатывает фильтр 👇
@router.message(Command(commands=["menu"]))
async def menu_handler(message: Message) -> None:
    # Вызывается обработчик 👆
    await message.answer(text="Текст меню")


# Пользователь выполняет команду ~menu
# Срабатывает фильтр при вызове альтернативной команды 👇
@router.message(Command(commands=["menu"], prefix="~"))
async def alternate_menu_handler(message: Message) -> None:
    await message.answer(text="Текст секретного меню")


# Пользователь выполняет команду /GET_FILE
# Срабатывает фильтр только на команду в верхнем регистре 👇
@router.message(Command(commands=["GET_FILE"], ignore_case=False))
async def get_file_handler(message: Message) -> None:
    await message.answer(text="Текст для пользователя")

Этот фильтр особенно полезен, когда вам нужно обрабатывать различные команды с одинаковой логикой, но с разными префиксами или учитывать регистр команд. Например, если ваш бот поддерживает несколько форматов команд (с разными префиксами или в верхнем и нижнем регистре), вы можете легко настроить обработку всех вариантов.

StateFilter.

Ещё один весьма полезный фильтр, предназначенный для отслеживания состояний. Мы не будем погружаться в работу Машины Состояний (FSM), рассмотрим только работу фильтра.

Данный фильтр, срабатывает, когда мы устанавливаем для пользователя конкретное состояние, например, у нас есть состояние WAIT_CITY и два обработчика:

# Класс состояния
class GetCitySteps(StatesGroup):  
    WAIT_CITY = State()

# Обработчик приглашающий к вводу города
@router.message(Command(commands=["weather"]))
async def request_city_handler(message: Message, state: FSMContext) -> None:
    await message.answer("Введите город")
    await state.set_state(state=GetCitySteps.WAIT_CITY)

# Обработчик реакции на ввод города
@router.message(GetCitySteps.WAIT_CITY)
@router.message(StateFilter(GetCitySteps.WAIT_CITY))
async def show_city_handler(message: Message, state: FSMContext) -> None:
    await message.answer(f"Вы ввели {message.text}")

В примере выше, во втором обработчике вы можете наблюдать два варианта написания фильтра.

Первый вариант не использует специализированный StateFilter, а просто фильтрует "под капотом" по состоянию. Такой подход применим, когда нет необходимости в дополнительных фильтрах.

Второй вариант использует StateFilter. В обычном случае, когда нет необходимости комбинировать фильтры, нет необходимости в нём. В случае, когда нужно скомбинировать два и более фильтра, без него не обойтись. Об этом ниже.

"Магические" фильтры.

"Магические" или ещё известные как F-фильтры, позволяют получить доступ к репрезентации объекта сообщения и путём сравнения указанных условий, пропустить сообщение в обработчик или отклонить. 

Как работают F-фильтры?

Они представляют F-объект, который является экземпляром сообщения от пользователя. Ему доступны те же методы, что и для объекта Message или CallbackQuery. После чего, к этом объекту применяется сравнение с пользовательским условием.

Самый простой пример, отслеживание конкретных сообщений в чате:

@router.message(F.text.lower().contains("привет"))
async def hello_handler(message: Message) -> None:
    await message.answer(text="И тебе привет от бота!")

В примере выше, внутри @router.message() мы указали условие, что слово привет присутствует в тексте сообщения от пользователя приведённое в нижний регистр.

Поскольку .text представляет собой строковый тип данных, мы можем применять методы строк, в данном случае сперва понизили регистр текста методом .lower(), затем воспользовались методом .contains() для проверки вхождения слова "привет" в строку.
Данная запись аналогична lambda-функции: lambda message: 'foo' in message.text

Поддерживаемые типы.

Обращаться можно не только к .text у сообщения, а также и ко всем другим полям объекта.

Мы бы хотели опубликовать список всех магических фильтров, но в настоящий момент документация не содержит даже всех, перечисленных в этом списке, а работоспособность других ставится нами под сомнения.

Основные магические фильтры

• F.text - Отслеживание текстовых сообщений.
• F.photo - отслеживание отправленного изображения.
• F.document - отслеживания отправленных файлов.
• F.voice - отслеживания отправленных голосовых сообщений, а также видео-сообщений в "кружочках".
• F.data - отслеживание данных нажатой inline-клавиши.
• F.from_user - отслеживание отправленных сообщений от конкретного пользователя.
• F.chat - отслеживание сообщений отправленных в конкретном чате.

Остальные магические фильтры

• F.new_chat_members - Отслеживание новых участников чата.
• F.left_chat_member - Отслеживание выхода участника из чата.
• F.new_chat_photo - Отслеживание изменения фото чата.
• F.new_chat_title - Отслеживание изменения названия чата.
• F.web_app_data - Отслеживание данных, отправленных из веб-приложения.
• F.connected_website - Отслеживание сообщений, связанных с авторизацией через веб-сайт.
• F.sticker - Отслеживание отправленных стикеров.
• F.contact - Отслеживание отправленных контактов.
• F.poll - Отслеживание отправленных опросов.
• F.dice - Отслеживание отправленных эмодзи-кубиков.

И многие другие.

Поддерживаемые операции к одиночному фильтру.

К одному фильтру можно применить базовые операции сравнения, например:

• F.text == "меню" - проверяем, что сообщение от пользователя это "меню".
• F.chat.id != 123456 - проверяем, что сообщение не из чата с id 123456.

Также доступен специальный символ инверсии ~, например:

• ~F.from_user.id == 123456 - фильтр будет срабатывать только на сообщения от пользователя с id 123456.

Комбинирование фильтров.

С комбинированием всё куда интереснее. Можно совместить два и более фильтра для одного обработчика, что позволит добавить несколько уровней фильтрации в логику работы. И для этого есть несколько способов и поддерживаемых операций.

В булевой логике присутствует три основных оператора AND - &, OR - | и NOT - !, в русском языке они известны как И, ИЛИ, НЕ. Вы наверняка знакомы с ними и активно их применяете при построении логических условий, что касается комбинации фильтров, то тут точно такой же принцип.

Оператор AND.

Данный оператор возвращает True, только в том случае, если все участвующие в условии операнды положительны. В Python, он является "ленивым", т.е. если левый операнд отрицательный, то и правый проверять нет смысла.

Применяется он двумя способами:

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

#            Первый фильтр 👇     Второй фильтр 👇
@router.message(F.text.lower() == "привет", F.from_user.id == 123456)
async def hello_handler(message: Message) -> None:
    await message.answer(text="И тебе привет от бота!")

В примере выше к обработчику подключены два фильтра, первый проверяет, что пользователь отправил слово "Привет", второй, что это конкретный пользователь. Например, тут можно указать ID администратора чата и когда он приходит в чат и пишет "Привет", срабатывает некое событие.

2. Помимо запятой, их можно обрабатывать через символ амперсанда &:

#               Первое условие 👇            Второе условие 👇
@router.message(F.text.lower() == "привет" & F.from_user.id == 123456)
async def hello_handler(message: Message) -> None:
    await message.answer(text="И тебе привет от бота!")

В данном случае, если левый будет отрицательным, правый не будет проверяться и всё условие вернёт False, что не позволит сработать обработчику.

Их также можно объединить, применив и разделение запятой и амперсанд:

#                            Первый фильтр 👇                         Второй фильтр 👇
@router.message(F.text.lower() == "привет" & F.from_user.id == 123456, F.chat.id == 654321)
async def hello_handler(message: Message) -> None:
    await message.answer(text="И тебе привет от бота!")

Оператор OR.

Данный оператор возвращает True, если хотя бы один из операндов положительный. В коде он обозначается символом |:

#               Первое условие 👇            Второе условие 👇
@router.message(F.text.lower() == "привет" | F.from_user.id == 123456)
async def hello_handler(message: Message) -> None:
    await message.answer(text="И тебе привет от бота!")

В примере выше, обработчик сработает если одно из условий будет положительным.

Комбинирование AND и OR.

Можно также выстраивать сложные логические конструкции, комбинируя операции:

@router.message(  
    (F.text.lower() == "привет" & F.from_user.id == 123456) | (F.text.lower() == "привет" & F.chat.id == 654321)  
)
async def hello_handler(message: Message) -> None:
    await message.answer(text="И тебе привет от бота!")

В примере выше мы проверяем, что конкретный пользователь отправил слово "Привет" ИЛИ слово "Привет" было отправлено в указанном чате. Это простейший пример, но он отражает принцип построения сложной логики.

Специальные функции и методы в фильтрах.

Помимо логических операторов, есть ещё специальные функции:

• and_f() - работает аналогично оператору AND.
• or_f() - работает аналогично оператору OR.
• invert_f() - работает аналогично оператору NOT или символу ~.

А также методы:

• F.text.in_() - проверяет, что отправленный пользователем текст присутствует в коллекции.
• F.text.startswith и F.text.endswith - проверяет, что отправленный пользователем текст начинается или заканчивается на указанные строки.
• .as_() - передаёт результат работы фильтра в аргументы обработчика, пример из документации:

@router.message(F.text.regexp(r"^(\d+)$").as_("digits"))
async def any_digits_handler(message: Message, digits: Match[str]) -> None:
    await message.answer(html.quote(str(digits)))

В примере выше фильтр проверяет, состоит ли сообщение пользователя из чисел и если да, записывает его в переменную digits.

Ещё один пример:

admin_ids = [123456, 678901]  

@router.message(F.text.lower() == "привет", F.from_user.id.in_(admin_ids).as_("is_admin"))  
async def any_digits_handler(message: Message, is_admin: bool) -> None:  
    await message.answer(str(is_admin))

В примере выше у нас есть список с идентификаторами администраторов. В первом фильтре проверяем, что отправленное пользователем сообщение равно "привет", Далее во втором фильтре проверяем, есть ли идентификатор пользователя в списке администраторов и присваиваем результату имя is_admin для передачи её в обработчик.

Обратите внимание, в данном случае, даже если "привет" напишет НЕ АДМИНИСТРАТОР, обработчик всё равно начнёт работу, поскольку при присвоении имени фильтр считается успешно выполненным.

Собственные фильтры.

Собственные или "кастомные" фильтры, позволяют вынести логику фильтров в отдельный модуль. Как и было упомянуто ранее, "магические" фильтры отлично подойдут для маленьких проектов, но когда обработчиков становится много, а логика взаимодействия с пользователем усложняется, логичным шагом является вынесение логики фильтров в отдельный модуль. Это позволит более гибко контролировать фильтрации, а также даёт дополнительные возможности.

Пример из документации:

class MyFilter(Filter):
    def __init__(self, my_text: str) -> None:
        self.my_text = my_text

    async def __call__(self, message: Message) -> bool | dict[str, Any]:
        return message.text == self.my_text

Собственный фильтр представляет собой класс, унаследованный от Filter. 
Тело фильтра состоит из двух dunder-методов:

1. Конструктор класса __init__(self, ...) - Является необязательным методом. Принимает аргументы при создании экземпляра класса.
2. dunder-метод __call__(self, ...)- Срабатывает при вызове экземпляра класса как функцию. Принимает в аргументы выполняемое событие, например объект класса Message или CallbackQuery. Возвращает bool (True/False) или словарь данных.

Как работают собственные фильтры?

Разберём на примере фильтра администратора:

Создаём класс:

class AdminFilter(Filter):  
    async def __call__(self, message: Message) -> bool | dict[str, Any]:  
        return message.from_user.id in config.admin_ids

В теле класса прописываем только __call__(). Затем возвращаем результат проверки, что идентификатор пользователя присутствует в списке администраторов.

Добавляем его к обработчику:

@router.message(Command(commands="admin_menu"), AdminFilter())
async def send_admin_menu(message: Message) -> None:
    await message.answer(
        text="Меню администратора:,
        reply_markup=await get_admin_keyboard()
    )

В примере выше мы скомбинировали вызов команды /admin_menu и собственный фильтр.

Теперь, когда пользователь отправит команду /admin_menu, сперва AIOgram найдёт обработчик для этой команды, после чего будет вызван второй фильтр для проверки на админа. Если обе проверки будут положительными, пользователю, т.е. в нашем случае уже администратору, будет выведено меню.

Принцип DRY

Пример выше не кажется каким-то особенным, однако, представим ситуацию, что у нас есть 10 обработчиков которые должны быть доступны только администратору. Как это будет выглядеть в F-фильтрах:

@router.message(F.from_user.id.in_(admin_ids)))
...

И так в каждом обработчике. В нашем же случае, будет всего один класс и он будет везде одинаковым:

@router.message(AdminFilter())

И происходит ситуация, когда необходимо изменить условия фильтрации админа. Для этого надо будет пройтись по всем обработчикам и исправить условия, тогда как в собственном фильтре это достаточно сделать всего раз и оно будет применено ко всем.

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

Если в случае с F-фильтрами мы можем возвращать в обработчик только ограниченные данные, то собственные фильтры куда более гибче в этом плане. Ситуаций применения масса, например, мы отслеживаем отправку изображений в бота и нам нужно это изображение сохранить в S3-хранилище. Для этого необходимо получить BytesIO-объект. Это занимает три строки, которые в обработчике смотрятся инородно, так почему бы получение файла не вынести в фильтр? Который в свою очередь вернёт в обработчик готовый к работе файл.

Пример фильтра:

class PhotoFilter(Filter):  
    async def __call__(self, message: Message) -> bool | dict[str, Any]:
        if message.photo:
            file_id = message.photo.file_id  
            file = await bot.get_file(file_id)  
            file_data: BytesIO = await bot.download_file(file.file_path)
            return {"photo_file": file_data}

В коде выше мы проверяем, что в отправленном сообщении есть изображение. Если его нет, фильтр вернёт None, что равносильное отрицательному случаю.

Внутри условия получаем идентификатор файла изображения, получаем объект с информацией о файле, затем скачиваем файл с сервера Telegram в оперативную память.

В самом конце создаём словарь, в котором по указанному ключу передаём файл.

Пример обработчика:

@router.message(PhotoFilter())
async def photo_handler(message: Message, photo_file: BytesIO) -> None:
    await save_image_to_s3(image=photo_file)
    await message.answer(text="Фотография успешно загружена.")

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

И ситуаций в которых собственный фильтр будет предпочтительнее действительно масса: Можно производить действия с базой данных, оповещать администратора в момент когда как-то пользователь вызовет определённую команду, можно подготавливать данные и многое многое другое, на что способна ваша фантазия.

Заключение.

AIOgram предоставляет поистине широкий ассортимент возможностей по фильтрации событий. Рассказанное в этой статье лишь часть из того, что можно делать. В следующих постах разберём с вами работу callback-фильтров.