mirror of
https://github.com/serty2005/rmser.git
synced 2026-02-04 19:02:33 -06:00
68 lines
2.5 KiB
Python
68 lines
2.5 KiB
Python
import os
|
||
import time
|
||
import logging
|
||
import requests
|
||
from typing import Optional
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
IAM_TOKEN_URL = "https://iam.api.cloud.yandex.net/iam/v1/tokens"
|
||
|
||
class YandexAuthManager:
|
||
def __init__(self):
|
||
self.oauth_token = os.getenv("YANDEX_OAUTH_TOKEN")
|
||
|
||
# Кэширование IAM токена
|
||
self._iam_token = None
|
||
self._token_expire_time = 0
|
||
|
||
if not self.oauth_token:
|
||
logger.warning("YANDEX_OAUTH_TOKEN not set. Yandex services will be unavailable.")
|
||
|
||
def is_configured(self) -> bool:
|
||
return bool(self.oauth_token)
|
||
|
||
def reset_token(self):
|
||
"""Сбрасывает кэшированный токен, заставляя получить новый при следующем вызове."""
|
||
self._iam_token = None
|
||
self._token_expire_time = 0
|
||
|
||
def get_iam_token(self) -> Optional[str]:
|
||
"""
|
||
Получает IAM-токен. Если есть живой кэшированный — возвращает его.
|
||
Если нет — обменивает OAuth на IAM.
|
||
"""
|
||
current_time = time.time()
|
||
|
||
# Если токен есть и он "свежий" (с запасом в 5 минут)
|
||
if self._iam_token and current_time < self._token_expire_time - 300:
|
||
return self._iam_token
|
||
|
||
if not self.oauth_token:
|
||
logger.error("OAuth token not available.")
|
||
return None
|
||
|
||
logger.info("Obtaining new IAM token from Yandex...")
|
||
try:
|
||
response = requests.post(
|
||
IAM_TOKEN_URL,
|
||
json={"yandexPassportOauthToken": self.oauth_token},
|
||
timeout=10
|
||
)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
self._iam_token = data["iamToken"]
|
||
|
||
# Токен обычно живет 12 часов, но мы будем ориентироваться на поле expiresAt если нужно,
|
||
# или просто поставим таймер. Для простоты берем 1 час жизни кэша.
|
||
self._token_expire_time = current_time + 3600
|
||
|
||
logger.info("IAM token received successfully.")
|
||
return self._iam_token
|
||
except Exception as e:
|
||
logger.error(f"Failed to get IAM token: {e}")
|
||
return None
|
||
|
||
# Глобальный инстанс
|
||
yandex_auth = YandexAuthManager() |