New-planet-api/new-planet-backend/app/services/gigachat_service.py

import os

import aiohttp
import base64
import uuid
import time
from urllib.parse import urlencode
from typing import Optional, List, Dict, Any

from dotenv import load_dotenv

from app.core.config import settings

load_dotenv()

class GigaChatService:
    def __init__(self):
        self.access_token: Optional[str] = None
        self.token_expires_at: Optional[float] = None
    
    async def _get_token(self) -> str:
        """Получить OAuth токен"""
        # Проверяем, не истек ли токен (оставляем запас 60 секунд)
        if self.access_token and self.token_expires_at:
            if time.time() < (self.token_expires_at - 60):
                return self.access_token
        
        # Проверяем наличие credentials
        client_id = os.getenv("GIGACHAT_CLIENT_ID")
        client_secret = os.getenv("GIGACHAT_CLIENT_SECRET")
        
        if not client_id or not client_secret:
            raise Exception(
                "GigaChat credentials not configured. "
                "Please set GIGACHAT_CLIENT_ID and GIGACHAT_CLIENT_SECRET in .env file"
            )
        
        # Формируем credentials и кодируем в Base64 с явным указанием UTF-8
        credentials = f"{client_id}:{client_secret}".strip().encode('utf-8')  # Обрезаем лишние символы
        encoded_credentials = base64.b64encode(credentials).decode('utf-8')
        
        headers = {
            "Authorization": f"Basic {encoded_credentials}",
            "Content-Type": "application/x-www-form-urlencoded",
            "Accept": "application/json",
            "RqUID": str(uuid.uuid4())
        }
        
        # Правильно кодируем данные формы (как в рабочем примере)
        form_data = {
            "grant_type": "client_credentials",
            "scope": "GIGACHAT_API_PERS"
        }
        
        # Отключаем проверку SSL (только для разработки!)
        # Используем ssl=False для полного отключения проверки сертификата
        connector = aiohttp.TCPConnector(ssl=False)
        async with aiohttp.ClientSession(connector=connector) as session:
            async with session.post(
                os.getenv("GIGACHAT_BASE_URL"),
                headers=headers,
                data=form_data
            ) as response:
                if response.status != 200:
                    # Получаем детали ошибки из ответа
                    try:
                        error_body = await response.text()
                        # Пытаемся распарсить как JSON, если не получается - возвращаем текст
                        try:
                            error_json = await response.json()
                            error_detail = error_json.get("error_description") or error_json.get("error") or str(error_json)
                        except:
                            error_detail = error_body
                    except:
                        error_detail = "No error details available"
                    
                    raise Exception(
                        f"Failed to get token: HTTP {response.status}. "
                        f"Error details: {error_detail}. "
                        f"Check your GIGACHAT_CLIENT_ID and GIGACHAT_CLIENT_SECRET_2 in .env file"
                    )
                
                result = await response.json()
                self.access_token = result.get("access_token")
                if not self.access_token:
                    raise Exception(f"Token not found in response: {result}")
                
                # Обрабатываем время истечения токена (может быть expires_at или expires_in)
                expires_at = result.get("expires_at")
                expires_in = result.get("expires_in")
                
                if expires_at:
                    # expires_at может быть timestamp или количество секунд
                    if expires_at > 1000000000:  # Это timestamp
                        self.token_expires_at = expires_at
                    else:  # Это количество секунд
                        self.token_expires_at = time.time() + expires_at
                elif expires_in:
                    # expires_in - это всегда количество секунд до истечения
                    self.token_expires_at = time.time() + expires_in
                else:
                    # По умолчанию 30 минут (1800 секунд)
                    self.token_expires_at = time.time() + 1800
                
                return self.access_token
    
    async def chat(
        self,
        message: str,
        context: Optional[List[Dict[str, Any]]] = None,
        model: str = None
    ) -> Dict[str, Any]:
        """Отправить сообщение в GigaChat"""
        token = await self._get_token()
        model = model or settings.GIGACHAT_MODEL_CHAT or "GigaChat"
        
        messages = context or []
        messages.append({"role": "user", "content": message})
        
        headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "messages": messages,
            "temperature": 0.7,
            "max_tokens": 2000
        }
        
        # Отключаем проверку SSL (только для разработки!)
        # Используем ssl=False для полного отключения проверки сертификата
        connector = aiohttp.TCPConnector(ssl=False)
        async with aiohttp.ClientSession(connector=connector) as session:
            async with session.post(
                f"{settings.GIGACHAT_BASE_URL}/chat/completions",
                headers=headers,
                json=payload
            ) as response:
                if response.status != 200:
                    error_text = await response.text()
                    raise Exception(f"GigaChat API error: {response.status} - {error_text}")
                
                result = await response.json()
                return result
    
    async def generate_text(
        self,
        prompt: str,
        model: str = None
    ) -> str:
        """Генерация текста по промпту"""
        result = await self.chat(prompt, model=model)
        return result.get("choices", [{}])[0].get("message", {}).get("content", "")


gigachat_service = GigaChatService()