Paul Bailey
31da896e56
- Created FastAPI application with CORS middleware - Defined Pydantic models (RoleDetectionRequest, RolesResponse, TTSRequest, ErrorResponse) - Implemented /api/health endpoint - Set up error handling middleware for validation and general exceptions - Configured logging with INFO level - Added test script to verify API functionality - Validates Requirements: 9.1, 9.2, 9.4
100 lines
3.4 KiB
Python
100 lines
3.4 KiB
Python
"""
|
|
FastAPI backend for Text to Audiobook application.
|
|
"""
|
|
import logging
|
|
from typing import List, Optional
|
|
from fastapi import FastAPI, Request, status
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import JSONResponse
|
|
from fastapi.exceptions import RequestValidationError
|
|
from pydantic import BaseModel, Field
|
|
|
|
# Configure logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Pydantic Models
|
|
class RoleDetectionRequest(BaseModel):
|
|
"""Request model for role detection endpoint."""
|
|
text: str = Field(..., description="Text to analyze for role detection")
|
|
|
|
class RoleInfo(BaseModel):
|
|
"""Model representing a detected role."""
|
|
id: str = Field(..., description="Unique identifier for the role")
|
|
label: str = Field(..., description="Display name for the role")
|
|
|
|
class RolesResponse(BaseModel):
|
|
"""Response model for role detection endpoint."""
|
|
roles: List[RoleInfo] = Field(..., description="List of detected roles")
|
|
|
|
class RoleVoiceMapping(BaseModel):
|
|
"""Model for role-to-voice mapping."""
|
|
id: str = Field(..., description="Role identifier")
|
|
voice: str = Field(..., description="Kokoro voice ID")
|
|
|
|
class TTSRequest(BaseModel):
|
|
"""Request model for TTS generation endpoint."""
|
|
text: str = Field(..., description="Text to convert to speech")
|
|
lang: str = Field(..., description="Language code (a, b, f, j)")
|
|
roles: List[RoleVoiceMapping] = Field(..., description="Role to voice mappings")
|
|
speed: Optional[float] = Field(1.0, description="Playback speed (0.5-1.5)")
|
|
|
|
class ErrorResponse(BaseModel):
|
|
"""Standard error response model."""
|
|
error: str = Field(..., description="Error type")
|
|
message: str = Field(..., description="Human-readable error message")
|
|
details: Optional[dict] = Field(None, description="Additional error details")
|
|
|
|
# Initialize FastAPI app
|
|
app = FastAPI(
|
|
title="Text to Audiobook API",
|
|
description="API for converting text to multi-voice audiobooks using Kokoro TTS",
|
|
version="1.0.0"
|
|
)
|
|
|
|
# Configure CORS
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["http://localhost:5173", "http://localhost:3000"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
# Error handling middleware
|
|
@app.exception_handler(RequestValidationError)
|
|
async def validation_exception_handler(request: Request, exc: RequestValidationError):
|
|
"""Handle validation errors with custom error response."""
|
|
logger.error(f"Validation error: {exc.errors()}")
|
|
return JSONResponse(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
content={
|
|
"error": "ValidationError",
|
|
"message": "Invalid request data",
|
|
"details": {"errors": exc.errors()}
|
|
}
|
|
)
|
|
|
|
@app.exception_handler(Exception)
|
|
async def general_exception_handler(request: Request, exc: Exception):
|
|
"""Handle general exceptions with custom error response."""
|
|
logger.error(f"Unhandled exception: {str(exc)}", exc_info=True)
|
|
return JSONResponse(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
content={
|
|
"error": "InternalServerError",
|
|
"message": "An unexpected error occurred",
|
|
"details": {"error": str(exc)}
|
|
}
|
|
)
|
|
|
|
# API Endpoints
|
|
@app.get("/api/health")
|
|
async def health_check():
|
|
"""Health check endpoint."""
|
|
logger.info("Health check requested")
|
|
return {"status": "ok"}
|