From 8170d7a58856da09341dc9f0bb8297d959ff755b Mon Sep 17 00:00:00 2001 From: Verum Date: Mon, 23 Feb 2026 14:37:52 +0700 Subject: [PATCH] =?UTF-8?q?=D0=A4=D0=B8=D0=BB=D1=8C=D1=82=D1=80=20=D0=BE?= =?UTF-8?q?=D0=B1=D1=80=D0=B0=D0=B1=D0=BE=D1=82=D0=BA=D0=B8=20callback-?= =?UTF-8?q?=D0=B7=D0=B0=D0=BF=D1=80=D0=BE=D1=81=D0=BE=D0=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- bot/filters/callback.py | 253 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 253 insertions(+) create mode 100644 bot/filters/callback.py diff --git a/bot/filters/callback.py b/bot/filters/callback.py new file mode 100644 index 0000000..ad52fdc --- /dev/null +++ b/bot/filters/callback.py @@ -0,0 +1,253 @@ +""" +Фильтры для обработки callback-запросов +""" +import re +from typing import Union + +from aiogram.filters import BaseFilter +from aiogram.types import CallbackQuery + +from middleware.loggers import logger + +__all__ = ( + 'CallbackStartsWith', + 'CallbackEndsWith', + 'CallbackContains', + 'CallbackMatches', + 'CallbackIn' +) + + +class CallbackStartsWith(BaseFilter): + """ + Проверяет, начинается ли callback_data с указанного префикса. + + Attributes: + prefix: Префикс для проверки (строка или список строк) + ignore_case: Игнорировать регистр + + Example: + ```python + # Один префикс + @router.callback_query(CallbackStartsWith("menu:")) + async def menu_handler(callback: CallbackQuery): + await callback.answer("Меню") + + # Несколько префиксов + @router.callback_query(CallbackStartsWith(["admin:", "mod:"])) + async def admin_handler(callback: CallbackQuery): + await callback.answer("Админ панель") + ``` + """ + + def __init__(self, prefix: Union[str, list[str]], ignore_case: bool = True): + """ + Args: + prefix: Префикс или список префиксов + ignore_case: Игнорировать регистр букв + """ + self.prefixes = [prefix] if isinstance(prefix, str) else prefix + self.ignore_case = ignore_case + + if self.ignore_case: + self.prefixes = [p.lower() for p in self.prefixes] + + async def __call__(self, callback: CallbackQuery) -> Union[bool, dict]: + if not callback.data: + return False + + data = callback.data.lower() if self.ignore_case else callback.data + + for prefix in self.prefixes: + if data.startswith(prefix): + # Извлекаем данные после префикса + value = callback.data[len(prefix):] + + logger.debug( + f"Callback с префиксом '{prefix}': {callback.data}", + log_type='CALLBACK' + ) + + return { + 'matched': True, + 'prefix': prefix, + 'value': value, + 'full_data': callback.data + } + + return False + + +class CallbackEndsWith(BaseFilter): + """ + Проверяет, заканчивается ли callback_data на указанный суффикс. + + Example: + ```python + @router.callback_query(CallbackEndsWith(":confirm")) + async def confirm_handler(callback: CallbackQuery, matched: dict): + action = matched['value'] + await callback.answer(f"Подтверждение: {action}") + ``` + """ + + def __init__(self, suffix: Union[str, list[str]], ignore_case: bool = True): + """ + Args: + suffix: Суффикс или список суффиксов + ignore_case: Игнорировать регистр букв + """ + self.suffixes = [suffix] if isinstance(suffix, str) else suffix + self.ignore_case = ignore_case + + if self.ignore_case: + self.suffixes = [s.lower() for s in self.suffixes] + + async def __call__(self, callback: CallbackQuery) -> Union[bool, dict]: + if not callback.data: + return False + + data = callback.data.lower() if self.ignore_case else callback.data + + for suffix in self.suffixes: + if data.endswith(suffix): + # Извлекаем данные до суффикса + value = callback.data[:-len(suffix)] + + return { + 'matched': True, + 'suffix': suffix, + 'value': value, + 'full_data': callback.data + } + + return False + + +class CallbackContains(BaseFilter): + """ + Проверяет, содержит ли callback_data указанную подстроку. + + Example: + ```python + @router.callback_query(CallbackContains("delete")) + async def delete_handler(callback: CallbackQuery): + await callback.answer("Удаление...") + ``` + """ + + def __init__(self, substring: Union[str, list[str]], ignore_case: bool = True): + """ + Args: + substring: Подстрока или список подстрок + ignore_case: Игнорировать регистр букв + """ + self.substrings = [substring] if isinstance(substring, str) else substring + self.ignore_case = ignore_case + + if self.ignore_case: + self.substrings = [s.lower() for s in self.substrings] + + async def __call__(self, callback: CallbackQuery) -> Union[bool, dict]: + if not callback.data: + return False + + data = callback.data.lower() if self.ignore_case else callback.data + + for substring in self.substrings: + if substring in data: + return { + 'matched': True, + 'substring': substring, + 'full_data': callback.data + } + + return False + + +class CallbackMatches(BaseFilter): + """ + Проверяет callback_data по regex паттерну. + + Example: + ```python + # Паттерн: user_123, user_456 и т.д. + @router.callback_query(CallbackMatches(r"^user_(\d+)$")) + async def user_handler(callback: CallbackQuery, matched: dict): + user_id = matched['groups'] + await callback.answer(f"Пользователь {user_id}") + ``` + """ + + def __init__(self, pattern: Union[str, re.Pattern], flags: int = 0): + """ + Args: + pattern: Regex паттерн (строка или скомпилированный Pattern) + flags: Флаги для regex (например, re.IGNORECASE) + """ + if isinstance(pattern, str): + self.pattern = re.compile(pattern, flags) + else: + self.pattern = pattern + + async def __call__(self, callback: CallbackQuery) -> Union[bool, dict]: + if not callback.data: + return False + + match = self.pattern.match(callback.data) + + if match: + logger.debug( + f"Callback соответствует паттерну {self.pattern.pattern}: {callback.data}", + log_type='CALLBACK' + ) + + return { + 'matched': True, + 'pattern': self.pattern.pattern, + 'groups': match.groups(), + 'groupdict': match.groupdict(), + 'full_data': callback.data + } + + return False + + +class CallbackIn(BaseFilter): + """ + Проверяет, находится ли callback_data в списке разрешенных значений. + + Example: + ```python + @router.callback_query(CallbackIn(["yes", "no", "cancel"])) + async def choice_handler(callback: CallbackQuery): + choice = callback.data + await callback.answer(f"Выбрано: {choice}") + ``` + """ + + def __init__(self, values: list[str], ignore_case: bool = True): + """ + Args: + values: Список разрешенных значений + ignore_case: Игнорировать регистр букв + """ + self.values = values + self.ignore_case = ignore_case + + if self.ignore_case: + self.values = [v.lower() for v in values] + + async def __call__(self, callback: CallbackQuery) -> Union[bool, dict]: + if not callback.data: + return False + + data = callback.data.lower() if self.ignore_case else callback.data + + if data in self.values: + return { + 'matched': True, + 'value': callback.data + } + + return False