70 lines
2.5 KiB
Python
70 lines
2.5 KiB
Python
"""Управление токенами GigaChat."""
|
|
import os
|
|
import time
|
|
from typing import Optional
|
|
|
|
import aiohttp
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
|
|
class TokenManager:
|
|
"""Менеджер токенов для GigaChat API."""
|
|
|
|
def __init__(
|
|
self,
|
|
client_id: Optional[str] = None,
|
|
client_secret: Optional[str] = None,
|
|
auth_url: Optional[str] = None,
|
|
):
|
|
self.client_id = client_id or os.getenv("GIGACHAT_CLIENT_ID")
|
|
self.client_secret = client_secret or os.getenv("GIGACHAT_CLIENT_SECRET")
|
|
self.auth_url = auth_url or os.getenv(
|
|
"GIGACHAT_AUTH_URL", "https://ngw.devices.sberbank.ru:9443/api/v2/oauth"
|
|
)
|
|
self._access_token: Optional[str] = None
|
|
self._expires_at: float = 0
|
|
|
|
async def get_token(self, force_refresh: bool = False) -> str:
|
|
"""
|
|
Получить актуальный токен доступа.
|
|
|
|
Args:
|
|
force_refresh: Принудительно обновить токен
|
|
|
|
Returns:
|
|
Токен доступа
|
|
"""
|
|
if not force_refresh and self._access_token and time.time() < self._expires_at:
|
|
return self._access_token
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
auth = aiohttp.BasicAuth(self.client_id, self.client_secret)
|
|
async with session.post(
|
|
self.auth_url,
|
|
auth=auth,
|
|
data={"scope": "GIGACHAT_API_PERS"},
|
|
) as response:
|
|
if response.status != 200:
|
|
error_text = await response.text()
|
|
raise Exception(f"Failed to get token: {response.status} - {error_text}")
|
|
|
|
data = await response.json()
|
|
self._access_token = data["access_token"]
|
|
# Токен обычно действителен 30 минут, обновляем за 5 минут до истечения
|
|
expires_in = data.get("expires_in", 1800)
|
|
self._expires_at = time.time() + expires_in - 300
|
|
|
|
return self._access_token
|
|
|
|
def is_token_valid(self) -> bool:
|
|
"""Проверить, действителен ли текущий токен."""
|
|
return self._access_token is not None and time.time() < self._expires_at
|
|
|
|
def clear_token(self):
|
|
"""Очистить токен (для тестирования)."""
|
|
self._access_token = None
|
|
self._expires_at = 0
|
|
|