bot/auth.py - master - zimaaadev/Parser

bot/auth.py
83 строки3 KB
zimaaadev
Бот работает, геокодинг тоже
18 дек 2025, 14:39
18 дек 2025, 14:39cf545bd
"""
Модуль аутентификации для SaluteSpeech (STT/TTS)
Автоматически получает и обновляет access_token.
"""
import asyncio
import time
import aiohttp
from typing import Optional
import uuid

# === Ваш ключ ===
AUTHORIZATION_KEY = "MDE5YjJiYWYtYjRjOS03YTcyLWFiZjAtZmM1Yjk1NjEzMDBlOjFiODJiNTQ4LTkzNmItNGFhMS05MDBlLTQ1ZDA5N2M0MjIzOA=="

# === URL API ===
TOKEN_URL = "https://ngw.devices.sberbank.ru:9443/api/v2/oauth"
STT_URL = "https://smartspeech.sber.ru/rest/v1/speech:recognize"
TTS_URL = "https://smartspeech.sber.ru/rest/v1/text:synthesize"

# === Глобальное состояние ===
_access_token: Optional[str] = None
_token_expiry: float = 0
_token_lock = asyncio.Lock()


async def get_access_token(session: aiohttp.ClientSession) -> str:
    """
    Возвращает валидный access_token.
    Автоматически обновляет, если токен просрочен.
    """
    global _access_token, _token_expiry

    now = time.time()
    if _access_token and now < _token_expiry - 60:
        return _access_token

    async with _token_lock:
        # Повторная проверка после захвата блокировки
        if _access_token and now < _token_expiry - 60:
            return _access_token

        try:
            async with session.post(
                TOKEN_URL,
                data={"scope": "SALUTE_SPEECH_PERS"},
                headers={
                    "Authorization": f"Basic {AUTHORIZATION_KEY}",
                    "RqUID": str(uuid.uuid4()),
                    "Content-Type": "application/x-www-form-urlencoded",
                    "Accept": "application/json"
                },
                ssl=False
            ) as resp:
                if resp.status != 200:
                    text = await resp.text()
                    raise Exception(f"❌ Ошибка токена {resp.status}: {text}")

                data = await resp.json()
                print("✅ Ответ на получение токена:", data)

                token = data.get("access_token")
                if not token:
                    raise Exception("❌ access_token отсутствует в ответе")

                # Обработка expires_at (в миллисекундах)
                expires_at_ms = data.get("expires_at")
                if expires_at_ms:
                    expires_in = (expires_at_ms / 1000) - now  # в секундах
                    if expires_in <= 0:
                        expires_in = 60  # минимальное время
                else:
                    expires_in = 3600  # fallback

                _access_token = token
                _token_expiry = now + expires_in
                print(f"⏰ Токен будет действовать: {int(expires_in)} секунд")
                return _access_token

        except Exception as e:
            print(f"❌ Ошибка получения токена: {e}")
            if _access_token:
                print("⚠️ Используем старый токен")
                return _access_token
            raise