142 lines
5.9 KiB
Python
142 lines
5.9 KiB
Python
"""
|
||
Декораторы для проверки разрешений
|
||
"""
|
||
from functools import wraps
|
||
from typing import Callable, Any, Union
|
||
from aiogram.types import Message, CallbackQuery
|
||
from services.infrastructure.logger import get_logger
|
||
from .registry import get_permission_checker
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
|
||
def require_permission(permission_name: str, error_message: str = "❌ У вас нет прав для выполнения этой команды."):
|
||
"""
|
||
Декоратор для проверки разрешения пользователя
|
||
|
||
Args:
|
||
permission_name: Имя разрешения для проверки
|
||
error_message: Сообщение об ошибке при отсутствии разрешения
|
||
"""
|
||
def decorator(func: Callable) -> Callable:
|
||
@wraps(func)
|
||
async def wrapper(*args, **kwargs):
|
||
# Извлекаем объект события (Message или CallbackQuery)
|
||
event = None
|
||
for arg in args:
|
||
if isinstance(arg, (Message, CallbackQuery)):
|
||
event = arg
|
||
break
|
||
|
||
if not event:
|
||
logger.error("Не удалось найти объект события для проверки разрешения")
|
||
return await func(*args, **kwargs)
|
||
|
||
# Получаем проверщик разрешений
|
||
checker = get_permission_checker()
|
||
if not checker:
|
||
logger.error("Проверщик разрешений не инициализирован")
|
||
return await func(*args, **kwargs)
|
||
|
||
# Проверяем разрешение
|
||
user_id = event.from_user.id
|
||
has_permission = await checker.has_permission(user_id, permission_name)
|
||
|
||
if not has_permission:
|
||
if isinstance(event, Message):
|
||
await event.answer(error_message)
|
||
elif isinstance(event, CallbackQuery):
|
||
await event.answer(error_message, show_alert=True)
|
||
return
|
||
|
||
# Выполняем оригинальную функцию
|
||
return await func(*args, **kwargs)
|
||
|
||
return wrapper
|
||
return decorator
|
||
|
||
|
||
def require_admin(error_message: str = "❌ У вас нет прав администратора."):
|
||
"""
|
||
Декоратор для проверки прав администратора
|
||
|
||
Args:
|
||
error_message: Сообщение об ошибке при отсутствии прав администратора
|
||
"""
|
||
return require_permission("admin", error_message)
|
||
|
||
|
||
def require_superuser(error_message: str = "❌ У вас нет прав суперпользователя."):
|
||
"""
|
||
Декоратор для проверки прав суперпользователя
|
||
|
||
Args:
|
||
error_message: Сообщение об ошибке при отсутствии прав суперпользователя
|
||
"""
|
||
return require_permission("superuser", error_message)
|
||
|
||
|
||
def require_admin_or_superuser(error_message: str = "❌ У вас нет прав для выполнения этой команды."):
|
||
"""
|
||
Декоратор для проверки прав администратора или суперпользователя
|
||
|
||
Args:
|
||
error_message: Сообщение об ошибке при отсутствии прав
|
||
"""
|
||
def decorator(func: Callable) -> Callable:
|
||
@wraps(func)
|
||
async def wrapper(*args, **kwargs):
|
||
# Извлекаем объект события (Message или CallbackQuery)
|
||
event = None
|
||
for arg in args:
|
||
if isinstance(arg, (Message, CallbackQuery)):
|
||
event = arg
|
||
break
|
||
|
||
if not event:
|
||
logger.error("Не удалось найти объект события для проверки разрешения")
|
||
return await func(*args, **kwargs)
|
||
|
||
# Получаем проверщик разрешений
|
||
checker = get_permission_checker()
|
||
if not checker:
|
||
logger.error("Проверщик разрешений не инициализирован")
|
||
return await func(*args, **kwargs)
|
||
|
||
# Проверяем права администратора или суперпользователя
|
||
user_id = event.from_user.id
|
||
has_permission = await checker.has_any_permission(user_id, ["admin", "superuser"])
|
||
|
||
if not has_permission:
|
||
if isinstance(event, Message):
|
||
await event.answer(error_message)
|
||
elif isinstance(event, CallbackQuery):
|
||
await event.answer(error_message, show_alert=True)
|
||
return
|
||
|
||
# Выполняем оригинальную функцию
|
||
return await func(*args, **kwargs)
|
||
|
||
return wrapper
|
||
return decorator
|
||
|
||
|
||
def require_active_user(error_message: str = "❌ Ваш аккаунт неактивен."):
|
||
"""
|
||
Декоратор для проверки активности пользователя
|
||
|
||
Args:
|
||
error_message: Сообщение об ошибке при неактивном аккаунте
|
||
"""
|
||
return require_permission("view_questions", error_message)
|
||
|
||
|
||
def require_unbanned_user(error_message: str = "❌ Ваш аккаунт заблокирован."):
|
||
"""
|
||
Декоратор для проверки, что пользователь не забанен
|
||
|
||
Args:
|
||
error_message: Сообщение об ошибке при заблокированном аккаунте
|
||
"""
|
||
return require_permission("ask_questions", error_message)
|