254 lines
8.0 KiB
Python
254 lines
8.0 KiB
Python
"""
|
||
Фильтры для обработки 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
|