forked from HSE_team/BetterCallPraskovia
107 lines
4.1 KiB
Python
107 lines
4.1 KiB
Python
|
||
import aiohttp
|
||
from datetime import datetime
|
||
from typing import Optional
|
||
from tg_bot.config.settings import settings
|
||
|
||
|
||
class User:
|
||
"""Модель пользователя для телеграм-бота"""
|
||
def __init__(self, data: dict):
|
||
self.user_id = data.get("user_id")
|
||
self.telegram_id = data.get("telegram_id")
|
||
self.role = data.get("role")
|
||
created_at_str = data.get("created_at")
|
||
if created_at_str:
|
||
try:
|
||
created_at_str = created_at_str.replace("Z", "+00:00")
|
||
self.created_at = datetime.fromisoformat(created_at_str)
|
||
except (ValueError, AttributeError):
|
||
self.created_at = None
|
||
else:
|
||
self.created_at = None
|
||
|
||
premium_until_str = data.get("premium_until")
|
||
if premium_until_str:
|
||
try:
|
||
premium_until_str = premium_until_str.replace("Z", "+00:00")
|
||
self.premium_until = datetime.fromisoformat(premium_until_str)
|
||
except (ValueError, AttributeError):
|
||
self.premium_until = None
|
||
else:
|
||
self.premium_until = None
|
||
|
||
self.is_premium = data.get("is_premium", False)
|
||
self.questions_used = data.get("questions_used", 0)
|
||
|
||
|
||
class UserService:
|
||
"""Сервис для работы с пользователями через API бэкенда"""
|
||
|
||
def __init__(self):
|
||
self.backend_url = settings.BACKEND_URL
|
||
|
||
async def get_user_by_telegram_id(self, telegram_id: int) -> Optional[User]:
|
||
"""Получить пользователя по Telegram ID"""
|
||
try:
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.get(
|
||
f"{self.backend_url}/users/telegram/{telegram_id}"
|
||
) as response:
|
||
if response.status == 200:
|
||
data = await response.json()
|
||
return User(data)
|
||
return None
|
||
except Exception as e:
|
||
print(f"Error getting user: {e}")
|
||
return None
|
||
|
||
async def get_or_create_user(
|
||
self,
|
||
telegram_id: int,
|
||
username: str = "",
|
||
first_name: str = "",
|
||
last_name: str = ""
|
||
) -> User:
|
||
"""Получить или создать пользователя"""
|
||
user = await self.get_user_by_telegram_id(telegram_id)
|
||
if not user:
|
||
try:
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.post(
|
||
f"{self.backend_url}/users",
|
||
json={"telegram_id": str(telegram_id), "role": "user"}
|
||
) as response:
|
||
if response.status in [200, 201]:
|
||
data = await response.json()
|
||
return User(data)
|
||
except Exception as e:
|
||
print(f"Error creating user: {e}")
|
||
raise
|
||
return user
|
||
|
||
async def update_user_questions(self, telegram_id: int) -> bool:
|
||
"""Увеличить счетчик использованных вопросов"""
|
||
try:
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.post(
|
||
f"{self.backend_url}/users/telegram/{telegram_id}/increment-questions"
|
||
) as response:
|
||
return response.status == 200
|
||
except Exception as e:
|
||
print(f"Error updating questions: {e}")
|
||
return False
|
||
|
||
async def activate_premium(self, telegram_id: int, days: int = 30) -> bool:
|
||
"""Активировать premium статус"""
|
||
try:
|
||
async with aiohttp.ClientSession() as session:
|
||
async with session.post(
|
||
f"{self.backend_url}/users/telegram/{telegram_id}/activate-premium",
|
||
params={"days": days}
|
||
) as response:
|
||
return response.status == 200
|
||
except Exception as e:
|
||
print(f"Error activating premium: {e}")
|
||
return False
|