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 app.core.config import settings
from app.services.ai_agent_client import ai_agent_client

class GigaChatService:
    """
    Сервис для работы с GigaChat через внешний AI-agent сервис.
    Все запросы к GigaChat теперь проходят через внешний сервис.
    """
    def __init__(self):
        self.access_token: Optional[str] = None
        self.token_expires_at: Optional[float] = None
    
    async def _get_token(self) -> str:
        """
        Получить OAuth токен.
        
        ВНИМАНИЕ: Этот метод больше не используется, так как все запросы
        к GigaChat теперь проходят через внешний AI-agent сервис.
        Метод оставлен для возможной обратной совместимости.
        """
        # Проверяем, не истек ли токен (оставляем запас 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 через внешний AI-agent сервис.
        Сохраняет обратную совместимость с форматом ответа GigaChat API.
        """
        try:
            # Используем внешний AI-agent сервис
            result = await ai_agent_client.chat(
                message=message,
                conversation_id=None,  # Если нужен conversation_id, его нужно передавать отдельно
                context=context
            )
            
            # Преобразуем ответ AI-agent сервиса в формат, совместимый с GigaChat API
            # Предполагаем, что ai_agent_client возвращает структуру ChatResponse или аналогичную
            if "response" in result:
                # Если ответ в формате ChatResponse, преобразуем в формат GigaChat
                return {
                    "model": result.get("model", model or settings.GIGACHAT_MODEL_CHAT or "GigaChat"),
                    "choices": [{
                        "message": {
                            "role": "assistant",
                            "content": result["response"]
                        },
                        "finish_reason": "stop"
                    }],
                    "usage": {
                        "total_tokens": result.get("tokens_used", 0),
                        "prompt_tokens": 0,
                        "completion_tokens": result.get("tokens_used", 0)
                    }
                }
            else:
                # Если ответ уже в формате GigaChat, возвращаем как есть
                return result
        except Exception as e:
            # Если внешний сервис недоступен, пробрасываем ошибку
            raise Exception(f"AI-agent service error: {str(e)}")
    
    async def generate_text(
        self,
        prompt: str,
        model: str = None
    ) -> str:
        """
        Генерация текста по промпту через внешний AI-agent сервис.
        """
        try:
            # Используем метод generate_text из ai_agent_client
            response_text = await ai_agent_client.generate_text(prompt=prompt, model=model)
            return response_text
        except Exception as e:
            # Если произошла ошибка, пробрасываем её
            raise Exception(f"AI-agent service error: {str(e)}")


gigachat_service = GigaChatService()