139 lines
6.4 KiB
Python
139 lines
6.4 KiB
Python
import httpx
|
|
import asyncio
|
|
from app.core import config
|
|
from app.core.database import add_session
|
|
from app.core.localizer import localizer
|
|
|
|
async def login(login, password):
|
|
"""
|
|
Отправляет запрос на аутентификацию и в случае успеха сохраняет сессию.
|
|
|
|
:param login: Логин пользователя
|
|
:param password: Пароль пользователя
|
|
:return: Кортеж (успех: bool, сообщение: str)
|
|
"""
|
|
url = f"{config.BASE_URL}/v1/auth/login"
|
|
|
|
try:
|
|
async with httpx.AsyncClient(http2=True) as client:
|
|
response = await client.post(url, json={"login": login, "password": password})
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get("status") == "fine":
|
|
token_data = data["data"]
|
|
add_session(
|
|
login=login,
|
|
access_token=token_data["access_token"],
|
|
refresh_token=token_data["refresh_token"]
|
|
)
|
|
return True, localizer.translate("Успешный вход")
|
|
else:
|
|
return False, data.get("detail", localizer.translate("Неизвестная ошибка ответа"))
|
|
|
|
elif response.status_code in [401]:
|
|
error_data = response.json()
|
|
return False, error_data.get("detail", localizer.translate("Неверный логин или пароль"))
|
|
|
|
elif response.status_code in [403]:
|
|
error_data = response.json()
|
|
return False, error_data.get("detail", localizer.translate("Учетная запись пользователя отключена"))
|
|
|
|
elif response.status_code == 422:
|
|
return False, localizer.translate("Некорректные данные для входа")
|
|
|
|
else:
|
|
return False, f"{localizer.translate('Ошибка сервера')}: {response.status_code}"
|
|
|
|
except httpx.RequestError as e:
|
|
return False, f"{localizer.translate('Ошибка сети')}: {e}"
|
|
except Exception as e:
|
|
return False, f"{localizer.translate('Произошла ошибка')}: {e}"
|
|
|
|
async def register(login, password, invite=None):
|
|
"""
|
|
Отправляет запрос на регистрацию нового пользователя.
|
|
|
|
:param login: Логин
|
|
:param password: Пароль
|
|
:param invite: Инвайт-код (опционально)
|
|
:return: Кортеж (успех: bool, сообщение: str)
|
|
"""
|
|
url = f"{config.BASE_URL}/v1/auth/register"
|
|
payload = {"login": login, "password": password}
|
|
if invite:
|
|
payload["invite"] = invite
|
|
|
|
try:
|
|
async with httpx.AsyncClient(http2=True) as client:
|
|
response = await client.post(url, json=payload)
|
|
|
|
if response.status_code == 201:
|
|
return True, localizer.translate("Регистрация прошла успешно!")
|
|
|
|
error_data = response.json()
|
|
error_message = error_data.get("detail", localizer.translate("Произошла ошибка"))
|
|
|
|
if response.status_code == 409:
|
|
return False, localizer.translate("Этот логин уже занят.")
|
|
elif response.status_code == 400:
|
|
return False, localizer.translate("Неверный инвайт-код.")
|
|
elif response.status_code == 403:
|
|
return False, localizer.translate("Регистрация в данный момент отключена.")
|
|
elif response.status_code == 422:
|
|
return False, localizer.translate("Данные не прошли валидацию. Проверьте длину логина и пароля.")
|
|
else:
|
|
return False, f"{localizer.translate('Ошибка сервера')} ({response.status_code}): {error_message}"
|
|
|
|
except httpx.RequestError as e:
|
|
return False, f"{localizer.translate('Ошибка сети')}: {e}"
|
|
except Exception as e:
|
|
return False, f"{localizer.translate('Произошла ошибка')}: {e}"
|
|
|
|
|
|
async def get_user_role(access_token: str):
|
|
"""
|
|
Получает роль и права пользователя по токену доступа.
|
|
|
|
:param access_token: Токен доступа пользователя
|
|
:return: Кортеж (успех: bool, данные: UserRoleData | str)
|
|
"""
|
|
url = f"{config.BASE_URL}/v1/user/role"
|
|
headers = {"Authorization": f"Bearer {access_token}"}
|
|
|
|
try:
|
|
async with httpx.AsyncClient(http2=True) as client:
|
|
response = await client.get(url, headers=headers)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get("status") == "fine":
|
|
# Здесь можно добавить валидацию через Pydantic, если необходимо
|
|
# from app.core.models.user_models import UserRoleData
|
|
# user_role_data = UserRoleData(**data['data'])
|
|
return True, data['data']
|
|
else:
|
|
return False, data.get("detail", localizer.translate("Неизвестная ошибка ответа"))
|
|
|
|
elif response.status_code == 401:
|
|
return False, localizer.translate("Токен недействителен или истек")
|
|
elif response.status_code == 404:
|
|
return False, localizer.translate("Пользователь не найден")
|
|
else:
|
|
return False, f"{localizer.translate('Ошибка сервера')}: {response.status_code}"
|
|
|
|
except httpx.RequestError as e:
|
|
return False, f"{localizer.translate('Ошибка сети')}: {e}"
|
|
except Exception as e:
|
|
return False, f"{localizer.translate('Произошла ошибка')}: {e}"
|
|
|
|
|
|
# Пример использования (для тестирования)
|
|
async def main():
|
|
# Замените на реальные данные для теста
|
|
success, message = await login("testuser", "testpassword")
|
|
print(f"Результат входа: {success}, Сообщение: {message}")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|