src/cdp/clients/gigachat_client.py - develop - pythonpractice/consultant-data-provider

src/cdp/clients/gigachat_client.py
126 строк4 KB
Кузьма W600DEV Даждъбогин
TSKCNSLTNTDTP-1. base preparing. оживил проект. прописал базовые команды для мок-запроса и для реального.
15 май 2026, 16:09
15 май 2026, 16:09cab050c
"""Реальный клиент для GigaChat (исправленная версия)."""

import httpx
import uuid
from tenacity import retry, stop_after_attempt, wait_exponential
from typing import List, Dict, Optional
from datetime import datetime, timedelta

from .base import LLMClient
from ..config import load_config


class GigaChatClient(LLMClient):
    """Клиент для работы с GigaChat через HTTP API."""

    def __init__(self):
        self.config = load_config()
        self._token: Optional[str] = None
        self._token_expires_at: Optional[datetime] = None

        # Создаём HTTP клиент
        self._client = httpx.Client(
            timeout=self.config.timeout,
            verify=False  # Для локальной разработки (отключаем SSL)
        )

    def _acquire_token(self) -> str:
        """Получить JWT токен доступа согласно спецификации GigaChat."""

        # Формируем Authorization header: Basic <authorization_key>
        auth_header = f"Basic {self.config.client_secret.get_secret_value()}"

        # Генерируем уникальный ID запроса
        rq_uid = str(uuid.uuid4())

        # Тело запроса: только scope
        data = {"scope": "GIGACHAT_API_PERS"}

        headers = {
            "Content-Type": "application/x-www-form-urlencoded",
            "Accept": "application/json",
            "RqUID": rq_uid,
            "Authorization": auth_header
        }

        response = self._client.post(
            self.config.auth_url,
            data=data,
            headers=headers
        )
        response.raise_for_status()

        token_data = response.json()
        self._token = token_data["access_token"]

        # Токен живёт 30 минут, обновляем через 25 минут
        self._token_expires_at = datetime.now() + timedelta(minutes=25)

        print(f"🔑 Токен получен, истекает в {self._token_expires_at.strftime('%H:%M:%S')}")
        return self._token

    def _is_token_valid(self) -> bool:
        """Проверить, не истёк ли токен."""
        if not self._token or not self._token_expires_at:
            return False
        return datetime.now() < self._token_expires_at

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10)
    )
    def generate(self, prompt: str) -> str:
        """Отправить запрос к GigaChat."""
        messages = [{"role": "user", "content": prompt}]
        return self.generate_with_context(messages)

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10)
    )
    def generate_with_context(self, messages: List[Dict[str, str]]) -> str:
        """Отправить диалог с историей."""
        # Обновляем токен если нужно
        if not self._is_token_valid():
            self._acquire_token()

        request_body = {
            "model": self.config.model,
            "messages": messages,
            "temperature": 0.7
        }

        headers = {
            "Authorization": f"Bearer {self._token}",
            "Content-Type": "application/json"
        }

        try:
            response = self._client.post(
                f"{self.config.api_url}/chat/completions",
                json=request_body,
                headers=headers
            )
            response.raise_for_status()

            result = response.json()
            answer = result["choices"][0]["message"]["content"]
            return answer

        except httpx.HTTPStatusError as e:
            print(f"❌ HTTP ошибка: {e.response.status_code}")
            print(f"Текст: {e.response.text}")
            raise
        except Exception as e:
            print(f"❌ Ошибка при вызове GigaChat: {e}")
            raise

    def close(self):
        """Закрыть HTTP клиент."""
        self._client.close()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()