init

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

@@ -0,0 +1,102 @@
+import aiohttp
+import base64
+import uuid
+import time
+from typing import Optional, List, Dict, Any
+from app.core.config import settings
+
+
+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 = f"{settings.GIGACHAT_CLIENT_ID}:{settings.GIGACHAT_CLIENT_SECRET}"
+        encoded_credentials = base64.b64encode(credentials.encode()).decode()
+        
+        headers = {
+            "Authorization": f"Basic {encoded_credentials}",
+            "Content-Type": "application/x-www-form-urlencoded",
+            "Accept": "application/json",
+            "RqUID": str(uuid.uuid4())
+        }
+        
+        data = {"scope": "GIGACHAT_API_PERS"}
+        
+        async with aiohttp.ClientSession() as session:
+            async with session.post(
+                settings.GIGACHAT_AUTH_URL,
+                headers=headers,
+                data=data
+            ) as response:
+                if response.status != 200:
+                    raise Exception(f"Failed to get token: {response.status}")
+                
+                result = await response.json()
+                self.access_token = result.get("access_token")
+                expires_in = result.get("expires_at", 1800)
+                # expires_at может быть timestamp или количество секунд
+                if expires_in > 1000000000:  # Это timestamp
+                    self.token_expires_at = expires_in
+                else:  # Это количество секунд
+                    self.token_expires_at = time.time() + expires_in
+                
+                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
+        
+        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
+        }
+        
+        async with aiohttp.ClientSession() 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()
+