main.py - master - KKleinikov/FastAPI-GigaChatAI

main.py
368 строк12 KB
Konstantin-Kleinikov
1.04 Change Docker file
06 дек 2025, 22:06
06 дек 2025, 22:06c0f0109
import logging
import os
import pathlib

from contextlib import asynccontextmanager
from typing import Any

from dotenv import load_dotenv
from fastapi import Depends, FastAPI, Form, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from gigachat import GigaChat
from gigachat.exceptions import GigaChatException
from gigachat.models import Chat
from sqlalchemy.orm import Session

from db import Base, ChatHistory, SessionLocal, engine

# Настройка логирования
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s | %(levelname)s | %(message)s',
    handlers=[
        logging.FileHandler('app.log', encoding='utf-8'),
        logging.StreamHandler(),
    ],
)
logger = logging.getLogger(__name__)

# Загружаем переменные из .env
load_dotenv()


def _check_credentials() -> tuple[str, bool]:
    """
    Проверяет и возвращает учётные данные GigaChat и флаг проверки SSL-сертификатов.

    Загружает переменные окружения:
    - GIGACHAT_CREDENTIALS: обязательная строка для аутентификации в GigaChat API.
    - GIGACHAT_VERIFY_SSL: необязательный флаг (по умолчанию `False`).
        Если `False` — отключает проверку SSL-сертификатов.

    Returns:
        Кортеж из:
        - credentials (str): значение GIGACHAT_CREDENTIALS.
        - verify_ssl (bool): значение GIGACHAT_VERIFY_SSL, преобразованное в bool.

    Raises:
        ValueError: Если GIGACHAT_CREDENTIALS не установлена в .env файле.
    """
    credentials = os.getenv('GIGACHAT_CREDENTIALS')
    if not credentials:
        raise ValueError('GIGACHAT_CREDENTIALS не установлена в .env файле')

    # Получаем и парсим флаг проверки SSL
    verify_ssl_raw = os.getenv('GIGACHAT_VERIFY_SSL', 'False').strip().lower()
    verify_ssl = verify_ssl_raw in ('true', '1', 'yes', 'on')

    return credentials, verify_ssl


@asynccontextmanager
async def lifespan(app: FastAPI) -> Any:
    """
    Управление жизненным циклом приложения.

    Выполняет создание всех таблиц в базе данных при запуске приложения.

    Args:
        app: Экземпляр FastAPI приложения (игнорируется, но требуется сигнатурой).

    Yields:
        None
    """
    Base.metadata.create_all(bind=engine)
    logger.info('Все таблицы созданы')
    yield


app = FastAPI(
    title='GigaChat API',
    lifespan=lifespan,
)


# Настройка CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=['*'],
    allow_credentials=True,
    allow_methods=['*'],
    allow_headers=['*'],
)


def get_db() -> Session:
    """
    Зависимость для получения сессии базы данных.

    Используется в маршрутах с `Depends()` для безопасного получения и закрытия сессии.

    Yields:
        Активная сессия SQLAlchemy.
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


# Создание директорий для статики и шаблонов
pathlib.Path('static/css').mkdir(exist_ok=True, parents=True)
pathlib.Path('templates').mkdir(exist_ok=True, parents=True)

# Настройка шаблонов
app.mount('/static', StaticFiles(directory='static'), name='static')
templates = Jinja2Templates(directory='templates')


# Инициализация GigaChat
try:
    GIGACHAT_CREDENTIALS, VERIFY_SSL_CERT = _check_credentials()
    gigachat = GigaChat(
        credentials=GIGACHAT_CREDENTIALS,
        verify_ssl_certs=VERIFY_SSL_CERT,
    )
    logger.info('GigaChat успешно инициализирован')
except Exception:
    logger.exception('Ошибка инициализации GigaChat')
    gigachat = None


def get_user_ip(request: Request) -> str:
    """
    Определяет IP-адрес клиента из HTTP-заголовков.

    Поддерживает проксированные запросы через:
    - X-Forwarded-For
    - X-Real-IP
    - Прямое подключение (request.client.host)

    Args:
        request: Объект запроса FastAPI.

    Returns:
        Строка с IP-адресом клиента или 'unknown'.
    """
    if forwarded_for := request.headers.get('X-Forwarded-For'):
        return forwarded_for.split(',')[0].strip()
    if real_ip := request.headers.get('X-Real-IP'):
        return real_ip.strip()
    if request.client and request.client.host:
        return request.client.host
    return 'unknown'


@app.get('/models')
async def get_available_models() -> list[dict[str, Any]]:
    """
    Возвращает список доступных GigaChat-моделей.

    Формирует ответ в виде списка объектов с полями:
    - model: идентификатор модели
    - description: тип модели
    - capabilities: возможности (chat/embedder)

    Returns:
        Список словарей с информацией о моделях.

    Raises:
        HTTPException: Если GigaChat не инициализирован или произошла ошибка API.
    """
    if not gigachat:
        raise HTTPException(
            status_code=500, detail='GigaChat не инициализирован'
        )

    try:
        models_response = gigachat.get_models()
        logger.info('Получен ответ от /models')
        logger.debug('Тип models_response: %s', type(models_response))
        logger.debug('Содержимое models_response: %s', models_response)

        # Извлечение списка моделей
        if hasattr(models_response, 'data'):
            model_list = models_response.data
        elif isinstance(models_response, tuple) and len(models_response) > 0:
            first_item = models_response[0]
            model_list = (
                first_item.data if hasattr(first_item, 'data') else first_item
            )
        elif hasattr(models_response, '__root__'):
            model_list = models_response.__root__
        else:
            model_list = models_response

        def _raise_invalid_model_list() -> None:
            raise ValueError('Не удалось извлечь список моделей из ответа')

        if not hasattr(model_list, '__iter__'):
            _raise_invalid_model_list()

        result = []
        for model in model_list:
            model_id = getattr(model, 'id_', 'unknown')
            model_type = (
                'embedder'
                if 'Embedding' in model_id or 'Embeddings' in model_id
                else 'chat'
            )
            result.append({
                'model': model_id,
                'description': f'Model type: {model_type}',
                'capabilities': [model_type],
            })

        logger.info('Возвращено %d моделей', len(result))
        return result

    except Exception as e:
        logger.exception('Ошибка при получении моделей')
        raise HTTPException(
            status_code=500, detail=f'Ошибка получения моделей: {e!s}'
        ) from e


@app.get('/', response_class=HTMLResponse)
async def root(request: Request) -> HTMLResponse:
    """
    Главная страница приложения.

    Возвращает HTML-шаблон index.html.

    Args:
        request: Объект запроса.

    Returns:
        HTML-ответ с отрендеренным шаблоном.
    """
    return templates.TemplateResponse('index.html', {'request': request})


@app.post('/chat')
async def chat(
    prompt: str = Form(...),
    model: str = Form(...),
    user_ip: str = Depends(get_user_ip),
    db: Session = Depends(get_db),
) -> dict[str, str]:
    """
    Обрабатывает запрос к GigaChat API.

    Принимает промпт и модель, отправляет запрос к GigaChat,
    сохраняет историю в базу данных.

    Args:
        prompt: Текст запроса от пользователя.
        model: Название модели для генерации.
        user_ip: IP-адрес пользователя (внедряется через Depends).
        db: Сессия базы данных (внедряется через Depends).

    Returns:
        Словарь с ключом 'response' — текстом ответа модели.

    Raises:
        HTTPException: При ошибках GigaChat или базы данных.
    """
    if not gigachat:
        logger.error('GigaChat не инициализирован')
        raise HTTPException(
            status_code=500, detail='GigaChat не инициализирован'
        )

    try:
        payload = Chat(
            model=model,
            messages=[{'role': 'user', 'content': prompt}],
        )
        response = gigachat.chat(payload)
        assistant_message = response.choices[0].message.content
        logger.info('Сгенерирован ответ для модели %s', model)

        # Сохранение в БД
        chat_record = ChatHistory(
            user_ip=user_ip,
            prompt=prompt,
            response=assistant_message,
            model=model,
        )
        db.add(chat_record)
        db.commit()
        db.refresh(chat_record)

    except GigaChatException as e:
        error_msg = f'Ошибка GigaChat: {e!s}'
        logger.exception('Ошибка при обращении к GigaChat')
        db.add(
            ChatHistory(
                user_ip=user_ip,
                prompt=prompt,
                response=error_msg,
                model=model,
            )
        )
        db.commit()
        raise HTTPException(status_code=500, detail=error_msg) from e

    except Exception as e:
        logger.exception('Неизвестная ошибка при обработке запроса')
        db.add(
            ChatHistory(
                user_ip=user_ip,
                prompt=prompt,
                response=f'Неизвестная ошибка: {e!s}',
                model=model,
            )
        )
        db.commit()
        raise HTTPException(
            status_code=500, detail=f'Неизвестная ошибка: {e!s}'
        ) from e

    return {'response': assistant_message}


@app.get('/chat_history')
async def get_chat_history(
    db: Session = Depends(get_db),
) -> list[dict[str, Any]]:
    """
    Возвращает историю всех запросов из базы данных.

    Отображает записи в порядке убывания времени (сначала новые).

    Args:
        db: Сессия базы данных.

    Returns:
        Список словарей с полями: id, prompt, response, model, timestamp.

    Raises:
        HTTPException: При ошибке чтения из базы данных.
    """
    try:
        records = (
            db.query(ChatHistory).order_by(ChatHistory.timestamp.desc()).all()
        )
        logger.info('Запрошена история: %d записей найдено', len(records))
        result = [
            {
                'id': r.id,
                'prompt': r.prompt,
                'response': r.response,
                'model': r.model,
                'timestamp': r.timestamp.isoformat(),
            }
            for r in records
        ]
        return result

    except Exception as e:
        logger.exception('Ошибка чтения истории')
        raise HTTPException(
            status_code=500, detail=f'Ошибка чтения истории: {e!s}'
        ) from e