Фильтры⚓︎

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

• Command — фильтр для обработки команд вида /команда [аргументы].
• MagicFilter — универсальный декларативный фильтр (из библиотеки magic-filter) для проверки полей события.

Эти два подхода часто используются вместе: Command проверяет форму команды и выделяет аргументы, а MagicFilter — накладывает дополнительные условия на аргументы или другие поля сообщения.

Command⚓︎

Command — удобный фильтр для парсинга и обработки команд в тексте сообщения.

Command(*commands: str | Pattern[str],
        prefix: str = "/",
        ignore_case: bool = False,
        magic: MagicFilter | None = None)

Фильтр Command работает следующим образом:

1. Проверяет, что сообщение — это Message, и что оно имеет тип PLAIN_MESSAGE.
2. Проверяет, начинается ли текст с указанного префикса (по умолчанию /).
3. Делит строку на команду и аргументы (если есть).
4. Сравнивает команду с заданными (включая поддержку re.Pattern).
5. Если передан magic, применяется MagicFilter к CommandObject.
6. Если magic-фильтр сработал, возвращает словарь {"command": command_obj}.

CommandObject⚓︎

Если фильтр срабатывает, в хендлер можно получить объект CommandObject (передаётся как параметр command):

@dataclass(frozen=True)
class CommandObject:
    prefix: str
    command: str
    args: str | None
    regexp_match: Match[str] | None = None
    magic_result: Any | None = None

Примеры⚓︎

Простейший хендлер:

@r.message(Command("ping"))
async def handle_ping(message: Message):
    await bot.send_message(chat_id=message.chat_id, text="pong")

Несколько команд:

@r.message(Command("info", "about", "whoami"))
async def handle_info(message: Message, command: CommandObject):
    await bot.send_message(chat_id=message.chat_id, text=f"Used /{command.command}")

Команда с префиксом:

@r.message(Command(re.compile(r"echo_\d+")))
async def handle_echo_numbered(message: Message, command: CommandObject):
    await bot.send_message(chat_id=message.chat_id, text=f"Echo: {command.command} {command.args or ''}")

MagicFilter⚓︎

MagicFilter — декларативный, цепочечный фильтр из пакета magic-filter. Позволяет выражать проверки над полями события в виде цепочек и операторов. Вместо того чтобы вручную проверять поля апдейта внутри обработчика, условия можно задать прямо в декораторе.

Фильтр работает «лениво»: при объявлении обработчика сохраняется сама цепочка проверок, а не её результат. Проверка выполняется только в момент, когда приходит новое событие, поэтому фильтры легко комбинируются и остаются читаемыми. Такой подход делает код короче и понятнее: сразу видно, какие именно апдейты пройдут через конкретный обработчик.

Идея MagicFilter проста: вы описываете цепочку атрибутов и условие, а затем применяете её к объекту. Представьте, что у вас есть объект с вложенными полями. Вместо ручной проверки вида if obj.foo.bar.baz == "spam": ... можно собрать фильтр декларативно:

F.foo.bar.baz == "spam"

Получившийся фильтр — это не мгновенная проверка, а объект, который «запоминает» условие. При обработке апдейта этот фильтр автоматически применяется к объекту (роутер сам выполняет проверку под капотом). Технически для этого используется метод .resolve(obj), но напрямую вызывать его в коде не требуется — достаточно описать условие в декораторе, и оно будет выполнено при маршрутизации.

@r.message(F.text == "ping")
async def ping_handler(message):
    await message.answer("pong")

Здесь фильтр F.text == "ping" будет автоматически проверен для каждого входящего сообщения. Если условие совпадает, сработает обработчик.

Примеры⚓︎

Объект MagicFilter поддерживает базовые логические операции над атрибутами объектов.

Проверка существования:

F.photo  # message.photo

Равенство:

F.text == "hello"        # message.text == "hello"
F.from_user.id == 42     # message.from_user.id == 42
F.text != "spam"         # message.text != "spam"

Принадлежность множеству:

# query.from_user.id in {42, 1000, 123123}
F.from_user.id.in_({42, 1000, 123123})

# query.data in {"foo", "bar", "baz"}
F.data.in_({"foo", "bar", "baz"})       

Содержит:

F.text.contains("foo")  # "foo" in message.text

Начинается/заканчивается строкой:

F.text.startswith("foo")  # message.text.startswith("foo")
F.text.endswith("bar")    # message.text.endswith("bar")

Регулярные выражения:

F.text.regexp(r"Hello, .+")  # re.match(r"Hello, .+", message.text)

Пользовательская функция:

# (lambda chat: chat.id == -42)(message.chat)
F.chat.func(lambda chat: chat.id == -42)  

Инверсия результата:

~F.text                     # not message.text
~F.text.startswith("spam")  # not message.text.startswith("spam")

Комбинирование условий:

(F.from_user.id == 42) & (F.text == "admin")

F.text.startswith("a") | F.text.endswith("b")

(F.from_user.id.in_({42, 777, 911})) & (F.text.startswith("!") | F.text.startswith("/")) & F.text.contains("ban")   

Модификаторы атрибутов (строки):

F.text.lower() == "test"           # message.text.lower() == "test"
F.text.upper().in_({"FOO", "BAR"}) # message.text.upper() in {"FOO", "BAR"}
F.text.len() == 5                  # len(message.text) == 5

Создание своих фильтров⚓︎

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

Требования к реализации:

• Классовая структура: Собственный фильтр обязательно должен быть оформлен в виде класса. Это позволяет сохранять состояние (например, настройки фильтра) при его инициализации.
• Метод __call__: Внутри класса обязательно должен быть реализован метод call. Без него объект класса не будет считаться «вызываемым», и фильтр не сработает.
• Возвращаемое значение: Метод call должен возвращать True (если событие подходит под условия) или False (если событие должно быть проигнорировано данным хендлером).

from typing import Any

class MyFilter:
    """
    Фильтр проверяет, состоит ли пользователь 
    в указанной группе через API сервера.
    """

    def __init__(self, target_group: str):
        # Сохраняем нужную группу при создании экземпляра
        self.target_group = target_group

    async def __call__(self, event: Any) -> bool:
        user = getattr(event, "from_user", None)
        if not user:
            return False

        user_id = user.id
        # Выполняем проверку (например, через API)
        user_info: dict = await server_api.get_user(user_id)
        groups = user_info.get("groups", [])

        # Обязательно возвращаем True или False
        return any(group.get("id") == self.target_group for group in groups)

После того как класс фильтра создан, его можно использовать в декораторах хендлеров точно так же, как стандартные фильтры Command или F. При регистрации хендлера просто передайте экземпляр вашего класса в качестве аргумента.

from aiogram import Router, Message
# Импортируйте ваш класс фильтра
from my_filters import MyFilter 

router = Router()

# Подключаем фильтр аналогично Command или F
@router.message(MyFilter(target_group="0032"))
async def my_handler(message: Message):
    await message.answer("Доступ разрешен: вы состоите в нужной группе!")

При получении сообщения бот создаст экземпляр вашего класса, вызовет метод __call__ и, если тот вернет True, выполнит код хендлера.

Объединение фильтров⚓︎

В python-trueconf-bot вы можете комбинировать несколько фильтров в одном хендлере. По умолчанию, если вы перечисляете фильтры через запятую, они работают по логике AND (И) — хендлер сработает только в том случае, если все фильтры вернут True.

Command и кастомный фильтр

Вы можете ограничить доступ к конкретной команде, используя свой фильтр. Хендлер сработает, только если сообщение является указанной командой И ваш фильтр вернул True.

# Сработает, если пришла команда /admin
# И пользователь в группе "0032"
@r.message(Command("admin"), InGroupFilter(target_group="0032"))
async def admin_panel(message: Message):
    await message.answer("Добро пожаловать в админ-панель!")

MagicFilter (F) и кастомный фильтр

Комбинируйте проверку содержимого сообщения с вашими правилами доступа. Например, реагировать на конкретное слово в тексте только для определенных групп:

# Сработает, если в тексте есть " help"
# И пользователь в группе "support"
@r.message(F.text.contains("help"), InGroupFilter(target_group="0003"))
async def support_handler(message: Message):
    await message.answer("Система поддержки уведомлена о вашем запросе.")

Связка Command и аргументов команды с MagicFilter⚓︎

Комбинирование Command и MagicFilter — частая и рекомендованная практика: Command парсит команду и формирует CommandObject, а magic позволяет наложить дополнительные условия на args или другие части CommandObject.

Фильтрация команды /echo только если есть аргументы:

@r.message(Command("echo", magic=F.args.is_not(None)))
async def handle_echo(message: Message, command: CommandObject):
    await bot.send_message(chat_id=message.chat_id, text=command.args)

Дополнительная проверка длины аргумента:

@r.message(Command("upper", magic=F.args.len() > 3))
async def handle_upper(message: Message, command: CommandObject):
    await bot.send_message(chat_id=message.chat_id, text=(command.args or "").upper())

Проверка с func — единичное слово:

@r.message(Command("repeat", magic=F.args.func(lambda x: isinstance(x, str) and len(x.split()) == 1)))
async def handle_repeat(message: Message, command: CommandObject):
    await bot.send_message(chat_id=message.chat_id, text=f"{command.args} {command.args}")

Комбинирование regexp и magic:

@r.message(Command(re.compile(r"echo_\\d+"), magic=F.args))
async def handle_special_echo(message: Message, command: CommandObject):
    await bot.send_message(chat_id=message.chat_id, text=f"Special: {command.command} -> {command.args}")