Paul Bailey
77437348ed
- Created RoleDetector class with colon and bracket pattern extraction - Implemented role deduplication logic - Added fallback to default 'Narrator' role when no patterns found - Created /api/tts/roles endpoint using RoleDetector - Added comprehensive test suite for role detection - All tests passing (7/7) Requirements: 2.1, 2.2, 2.3, 2.4, 2.5, 2.6
127 lines
4.3 KiB
Python
127 lines
4.3 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
|
|
|
|
# Import services
|
|
from services.role_detector import RoleDetector
|
|
|
|
# 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)}
|
|
}
|
|
)
|
|
|
|
# Initialize services
|
|
role_detector = RoleDetector()
|
|
|
|
# API Endpoints
|
|
@app.get("/api/health")
|
|
async def health_check():
|
|
"""Health check endpoint."""
|
|
logger.info("Health check requested")
|
|
return {"status": "ok"}
|
|
|
|
@app.post("/api/tts/roles", response_model=RolesResponse)
|
|
async def detect_roles(request: RoleDetectionRequest):
|
|
"""
|
|
Analyzes text and returns detected speaker roles.
|
|
|
|
Supports two pattern types:
|
|
- Colon pattern: "RoleName:" at line start
|
|
- Bracket pattern: "[RoleName]" preceding dialogue
|
|
|
|
Returns default "Narrator" role if no patterns found.
|
|
"""
|
|
logger.info(f"Role detection requested for text of length {len(request.text)}")
|
|
|
|
try:
|
|
roles = role_detector.detect_roles(request.text)
|
|
logger.info(f"Detected {len(roles)} roles: {[r['label'] for r in roles]}")
|
|
return RolesResponse(roles=roles)
|
|
except Exception as e:
|
|
logger.error(f"Error during role detection: {str(e)}", exc_info=True)
|
|
raise
|