147 lines
5.6 KiB
Python
147 lines
5.6 KiB
Python
"""
|
||
Новый сервис авторизации с использованием системы разрешений
|
||
Соблюдает принцип открытости/закрытости (OCP)
|
||
"""
|
||
from typing import Optional
|
||
from services.infrastructure.database import DatabaseService
|
||
from services.permissions import get_permission_checker
|
||
from services.infrastructure.logger import get_logger
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
|
||
class AuthService:
|
||
"""
|
||
Сервис авторизации, использующий систему разрешений.
|
||
Соблюдает принцип открытости/закрытости (OCP).
|
||
"""
|
||
|
||
def __init__(self, database: DatabaseService, config_provider):
|
||
self.database = database
|
||
self.config = config_provider
|
||
|
||
def is_admin(self, user_id: int) -> bool:
|
||
"""
|
||
Проверка, является ли пользователь администратором
|
||
|
||
Args:
|
||
user_id: ID пользователя в Telegram
|
||
|
||
Returns:
|
||
True если пользователь администратор, False иначе
|
||
"""
|
||
return user_id in self.config.ADMINS
|
||
|
||
async def is_superuser(self, user_id: int) -> bool:
|
||
"""
|
||
Проверка, является ли пользователь суперпользователем
|
||
|
||
Args:
|
||
user_id: ID пользователя в Telegram
|
||
|
||
Returns:
|
||
True если пользователь суперпользователь, False иначе
|
||
"""
|
||
try:
|
||
user = await self.database.get_user(user_id)
|
||
return user.is_superuser if user else False
|
||
except Exception:
|
||
return False
|
||
|
||
async def get_user_role(self, user_id: int) -> str:
|
||
"""
|
||
Получение роли пользователя
|
||
|
||
Args:
|
||
user_id: ID пользователя в Telegram
|
||
|
||
Returns:
|
||
Роль пользователя: 'admin', 'superuser' или 'user'
|
||
"""
|
||
if self.is_admin(user_id):
|
||
return 'admin'
|
||
elif await self.is_superuser(user_id):
|
||
return 'superuser'
|
||
else:
|
||
return 'user'
|
||
|
||
async def has_permission(self, user_id: int, permission: str) -> bool:
|
||
"""
|
||
Проверка наличия разрешения у пользователя через систему разрешений
|
||
|
||
Args:
|
||
user_id: ID пользователя в Telegram
|
||
permission: Название разрешения
|
||
|
||
Returns:
|
||
True если у пользователя есть разрешение, False иначе
|
||
"""
|
||
checker = get_permission_checker()
|
||
if not checker:
|
||
logger.error("Проверщик разрешений не инициализирован")
|
||
return False
|
||
|
||
return await checker.has_permission(user_id, permission)
|
||
|
||
async def has_any_permission(self, user_id: int, permissions: list[str]) -> bool:
|
||
"""
|
||
Проверка наличия хотя бы одного из разрешений у пользователя
|
||
|
||
Args:
|
||
user_id: ID пользователя в Telegram
|
||
permissions: Список названий разрешений
|
||
|
||
Returns:
|
||
True если у пользователя есть хотя бы одно разрешение, False иначе
|
||
"""
|
||
checker = get_permission_checker()
|
||
if not checker:
|
||
logger.error("Проверщик разрешений не инициализирован")
|
||
return False
|
||
|
||
return await checker.has_any_permission(user_id, permissions)
|
||
|
||
async def has_all_permissions(self, user_id: int, permissions: list[str]) -> bool:
|
||
"""
|
||
Проверка наличия всех разрешений у пользователя
|
||
|
||
Args:
|
||
user_id: ID пользователя в Telegram
|
||
permissions: Список названий разрешений
|
||
|
||
Returns:
|
||
True если у пользователя есть все разрешения, False иначе
|
||
"""
|
||
checker = get_permission_checker()
|
||
if not checker:
|
||
logger.error("Проверщик разрешений не инициализирован")
|
||
return False
|
||
|
||
return await checker.has_all_permissions(user_id, permissions)
|
||
|
||
async def get_user_permissions(self, user_id: int) -> list[str]:
|
||
"""
|
||
Получение списка всех разрешений пользователя
|
||
|
||
Args:
|
||
user_id: ID пользователя в Telegram
|
||
|
||
Returns:
|
||
Список названий разрешений пользователя
|
||
"""
|
||
checker = get_permission_checker()
|
||
if not checker:
|
||
logger.error("Проверщик разрешений не инициализирован")
|
||
return []
|
||
|
||
# Получаем все доступные разрешения
|
||
registry = checker.registry
|
||
all_permissions = registry.get_all()
|
||
|
||
user_permissions = []
|
||
for permission_name in all_permissions.keys():
|
||
if await checker.has_permission(user_id, permission_name):
|
||
user_permissions.append(permission_name)
|
||
|
||
return user_permissions
|