forked from HSE_team/BetterCallPraskovia
80 lines
3.6 KiB
Python
80 lines
3.6 KiB
Python
"""
|
||
Use cases для работы с пользователями
|
||
"""
|
||
from uuid import UUID
|
||
from typing import Optional
|
||
from datetime import datetime, timedelta
|
||
from src.domain.entities.user import User, UserRole
|
||
from src.domain.repositories.user_repository import IUserRepository
|
||
from src.shared.exceptions import NotFoundError, ValidationError
|
||
|
||
|
||
class UserUseCases:
|
||
"""Use cases для пользователей"""
|
||
|
||
def __init__(self, user_repository: IUserRepository):
|
||
self.user_repository = user_repository
|
||
|
||
async def create_user(self, telegram_id: str, role: UserRole = UserRole.USER) -> User:
|
||
"""Создать пользователя"""
|
||
existing_user = await self.user_repository.get_by_telegram_id(telegram_id)
|
||
if existing_user:
|
||
raise ValidationError(f"Пользователь с telegram_id {telegram_id} уже существует")
|
||
|
||
user = User(telegram_id=telegram_id, role=role)
|
||
return await self.user_repository.create(user)
|
||
|
||
async def get_user(self, user_id: UUID) -> User:
|
||
"""Получить пользователя по ID"""
|
||
user = await self.user_repository.get_by_id(user_id)
|
||
if not user:
|
||
raise NotFoundError(f"Пользователь {user_id} не найден")
|
||
return user
|
||
|
||
async def get_user_by_telegram_id(self, telegram_id: str) -> Optional[User]:
|
||
"""Получить пользователя по Telegram ID"""
|
||
return await self.user_repository.get_by_telegram_id(telegram_id)
|
||
|
||
async def update_user(self, user_id: UUID, telegram_id: str | None = None, role: UserRole | None = None) -> User:
|
||
"""Обновить пользователя"""
|
||
user = await self.get_user(user_id)
|
||
|
||
if telegram_id is not None:
|
||
user.telegram_id = telegram_id
|
||
if role is not None:
|
||
user.role = role
|
||
|
||
return await self.user_repository.update(user)
|
||
|
||
async def delete_user(self, user_id: UUID) -> bool:
|
||
"""Удалить пользователя"""
|
||
return await self.user_repository.delete(user_id)
|
||
|
||
async def list_users(self, skip: int = 0, limit: int = 100) -> list[User]:
|
||
"""Получить список пользователей"""
|
||
return await self.user_repository.list_all(skip=skip, limit=limit)
|
||
|
||
async def increment_questions_used(self, telegram_id: str) -> User:
|
||
"""Увеличить счетчик использованных вопросов"""
|
||
user = await self.user_repository.get_by_telegram_id(telegram_id)
|
||
if not user:
|
||
raise NotFoundError(f"Пользователь с telegram_id {telegram_id} не найден")
|
||
|
||
user.questions_used += 1
|
||
return await self.user_repository.update(user)
|
||
|
||
async def activate_premium(self, telegram_id: str, days: int = 30) -> User:
|
||
"""Активировать premium статус"""
|
||
user = await self.user_repository.get_by_telegram_id(telegram_id)
|
||
if not user:
|
||
raise NotFoundError(f"Пользователь с telegram_id {telegram_id} не найден")
|
||
|
||
user.is_premium = True
|
||
if user.premium_until and user.premium_until > datetime.utcnow():
|
||
user.premium_until = user.premium_until + timedelta(days=days)
|
||
else:
|
||
user.premium_until = datetime.utcnow() + timedelta(days=days)
|
||
|
||
return await self.user_repository.update(user)
|
||
|