api/main.py - master - YoungFreddy/FinalWork

api/main.py
165 строк5 KB
Igor
Initial commit: Q&A LLM service
23 июл 2026, 18:02
23 июл 2026, 18:02e5e37db
"""FastAPI-эндпоинты: /, /ask, /cache/status."""

import json
import logging
import time
from typing import Optional

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field

from config.loader import load_config
from services.qa_service import QAService

logging.basicConfig(
    level=logging.INFO,
    format="%(message)s",
    handlers=[
        logging.FileHandler("qa_service.log", encoding="utf-8"),
        logging.StreamHandler(),
    ],
)

app = FastAPI(title="Q&A Service API", description="Сервис Вопрос-Ответ на базе LLM")

_ERROR_MESSAGES = {
    "string_too_short": "должно содержать хотя бы {min_length} символ(а)",
    "string_too_long": "не может превышать {max_length} символов",
    "missing": "обязательное поле",
    "json_invalid": "невалидный JSON в теле запроса",
}


@app.exception_handler(RequestValidationError)
def handle_validation_error(request, exc):
    """Обрабатывает ошибки валидации Pydantic и возвращает 422 с русским описанием."""
    first = exc.errors()[0]
    field = " -> ".join(str(part) for part in first["loc"][1:]) if first["loc"] else "body"
    err_type = first.get("type", "")
    msg = first.get("msg", "Ошибка валидации")
    ctx = first.get("ctx", {})

    template = _ERROR_MESSAGES.get(err_type)
    if template:
        detail = f"Поле '{field}' {template.format(**ctx)}"
    else:
        detail = f"Поле '{field}': {msg}"

    logging.getLogger(__name__).error(
        json.dumps(
            {
                "event": "validation_error",
                "timestamp": time.time(),
                "field": field,
                "error_type": err_type,
                "detail": detail,
            },
            ensure_ascii=False,
        )
    )

    return JSONResponse(
        status_code=422,
        content={"error": "validation_error", "detail": detail},
    )


_service: Optional[QAService] = None


def get_service() -> QAService:
    """Возвращает синглтон QAService (ленивая инициализация)."""
    global _service
    if _service is None:
        cfg = load_config()
        _service = QAService(config=cfg)
    return _service


class QuestionRequest(BaseModel):
    question: str = Field(
        ..., min_length=1, max_length=2000, description="Вопрос пользователя"
    )


class AnswerResponse(BaseModel):
    answer: str


class CacheStatusResponse(BaseModel):
    cache_size: int


@app.get("/")
def read_root():
    """Корневой эндпоинт — информация о сервисе."""
    return {
        "service": "Q&A Service",
        "version": "1.0.0",
        "description": "Сервис Вопрос-Ответ на базе LLM",
    }


@app.post("/ask", response_model=AnswerResponse)
def ask(request: QuestionRequest):
    """Принимает вопрос пользователя и возвращает ответ от LLM."""
    svc = get_service()
    start_time = time.time()
    try:
        response_text = svc.answer_question(request.question)
        elapsed = time.time() - start_time
        logging.getLogger(__name__).info(
            json.dumps(
                {
                    "event": "api_response",
                    "timestamp": time.time(),
                    "elapsed_seconds": round(elapsed, 3),
                    "status": "success",
                },
                ensure_ascii=False,
            )
        )
        return AnswerResponse(answer=response_text)
    except ValueError as e:
        elapsed = time.time() - start_time
        logging.getLogger(__name__).error(
            json.dumps(
                {
                    "event": "api_error",
                    "timestamp": time.time(),
                    "elapsed_seconds": round(elapsed, 3),
                    "error": str(e),
                    "status": "validation_error",
                },
                ensure_ascii=False,
            )
        )
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        elapsed = time.time() - start_time
        logging.getLogger(__name__).error(
            json.dumps(
                {
                    "event": "api_error",
                    "timestamp": time.time(),
                    "elapsed_seconds": round(elapsed, 3),
                    "error": str(e),
                    "error_type": type(e).__name__,
                    "status": "internal_error",
                },
                ensure_ascii=False,
            )
        )
        raise HTTPException(
            status_code=503,
            detail="Сервис временно недоступен, попробуйте позже",
        )


@app.get("/cache/status")
def cache_status():
    """Возвращает текущий размер кэша."""
    svc = get_service()
    return CacheStatusResponse(cache_size=len(svc.cache._cache))