Фильтр обработки callback-запросов
This commit is contained in:
253
bot/filters/callback.py
Normal file
253
bot/filters/callback.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user