app.py

#!/usr/bin/env python3

"""
Text-to-Speech API using Edge-TTS with FastAPI
Optimized for Hugging Face Spaces deployment
"""

import edge_tts
import asyncio
import os
import tempfile
import uuid
import re
from fastapi import FastAPI, HTTPException, Form, UploadFile
from fastapi.responses import FileResponse, JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field, validator
import logging
from typing import Optional
import aiofiles

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# FastAPI app initialization
app = FastAPI(
    title="Text-to-Speech API",
    description="Convert text to speech using Microsoft Edge TTS with customizable voice, pitch, and rate",
    version="1.0.0",
    docs_url="/",  # Swagger UI at root for easy access
    redoc_url="/redoc"
)

# Add CORS middleware for public API access
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Allow all origins for public API
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Configuration
TEMP_DIR = tempfile.gettempdir()
MAX_TEXT_LENGTH = 5000

# Pydantic models for request validation
class TTSRequest(BaseModel):
    text: str = Field(..., min_length=1, max_length=MAX_TEXT_LENGTH, description="Text to convert to speech")
    voice: str = Field(default="en-US-AriaNeural", description="Voice identifier (e.g., 'en-GB-SoniaNeural')")
    pitch: str = Field(default="+0Hz", description="Pitch adjustment (e.g., '+10Hz', '-15Hz')")
    rate: str = Field(default="+0%", description="Rate adjustment (e.g., '+20%', '-10%')")
    
    @validator('pitch')
    def validate_pitch(cls, v):
        if not re.match(r'^[+-]?\d+Hz$', v):
            raise ValueError("Pitch must be in format like '+10Hz' or '-15Hz'")
        pitch_value = int(v.replace('Hz', '').replace('+', ''))
        if not -50 <= pitch_value <= 50:
            raise ValueError("Pitch value must be between -50 and 50")
        return v
    
    @validator('rate')
    def validate_rate(cls, v):
        if not re.match(r'^[+-]?\d+%$', v):
            raise ValueError("Rate must be in format like '+15%' or '-20%'")
        rate_value = int(v.replace('%', '').replace('+', ''))
        if not -50 <= rate_value <= 50:
            raise ValueError("Rate value must be between -50 and 50")
        return v

class VoiceInfo(BaseModel):
    name: str
    short_name: str
    gender: str
    locale: str
    language: str
    display_name: str

class HealthResponse(BaseModel):
    status: str
    service: str
    version: str

class VoicesResponse(BaseModel):
    voices: list[VoiceInfo]
    count: int

# Utility functions
async def generate_speech_async(text: str, voice: str, pitch: str, rate: str, output_file: str) -> bool:
    """Generate speech asynchronously"""
    try:
        # Use edge_tts.Communicate with direct parameters (no SSML needed)
        communicate = edge_tts.Communicate(text=text, voice=voice, rate=rate, pitch=pitch)
        await communicate.save(output_file)
        return True
    except Exception as e:
        logger.error(f"Error generating speech: {str(e)}")
        return False

def cleanup_file(file_path: str):
    """Clean up temporary file"""
    try:
        if os.path.exists(file_path):
            os.remove(file_path)
            logger.info(f"Cleaned up temporary file: {file_path}")
    except Exception as e:
        logger.warning(f"Failed to clean up temp file {file_path}: {str(e)}")

# API Endpoints
@app.get("/health", response_model=HealthResponse, tags=["Health"])
async def health_check():
    """Health check endpoint"""
    return HealthResponse(
        status="healthy",
        service="TTS API",
        version="1.0.0"
    )

@app.get("/voices", response_model=VoicesResponse, tags=["Voices"])
async def get_voices():
    """Get list of available voices"""
    try:
        voices = await edge_tts.list_voices()
        
        voice_list = [
            VoiceInfo(
                name=voice["Name"],
                short_name=voice["ShortName"],
                gender=voice["Gender"],
                locale=voice["Locale"],
                language=voice.get("Language", ""),
                display_name=voice.get("DisplayName", "")
            )
            for voice in voices
        ]
        
        return VoicesResponse(voices=voice_list, count=len(voice_list))
    except Exception as e:
        logger.error(f"Error fetching voices: {str(e)}")
        raise HTTPException(status_code=500, detail="Failed to fetch voices")

@app.post("/synthesize", tags=["TTS"])
async def synthesize_speech(request: TTSRequest):
    """
    Convert text to speech and return audio file
    
    - **text**: Text to convert to speech (required)
    - **voice**: Voice identifier (default: en-US-AriaNeural)
    - **pitch**: Pitch adjustment like '+10Hz' or '-15Hz' (default: +0Hz)
    - **rate**: Rate adjustment like '+20%' or '-10%' (default: +0%)
    """
    output_file = None
    try:
        # Generate unique filename
        file_id = str(uuid.uuid4())
        output_file = os.path.join(TEMP_DIR, f"tts_{file_id}.mp3")
        
        # Generate speech
        success = await generate_speech_async(
            request.text, request.voice, request.pitch, request.rate, output_file
        )
        
        if not success:
            raise HTTPException(status_code=500, detail="Failed to generate speech")
        
        if not os.path.exists(output_file):
            raise HTTPException(status_code=500, detail="Audio file was not generated")
        
        # Return the audio file directly
        return FileResponse(
            output_file,
            media_type="audio/mpeg",
            filename=f"speech_{file_id}.mp3",
            background=None  # Don't cleanup immediately, let the response complete first
        )
        
    except HTTPException:
        if output_file:
            cleanup_file(output_file)
        raise
    except Exception as e:
        if output_file:
            cleanup_file(output_file)
        logger.error(f"Error in synthesize_speech: {str(e)}")
        raise HTTPException(status_code=500, detail="Internal server error")

@app.post("/synthesize-form", tags=["TTS"])
async def synthesize_speech_form(
    text: str = Form(..., description="Text to convert to speech"),
    voice: str = Form(default="en-US-AriaNeural", description="Voice identifier"),
    pitch: str = Form(default="+0Hz", description="Pitch adjustment (e.g., '+10Hz')"),
    rate: str = Form(default="+0%", description="Rate adjustment (e.g., '+20%')")
):
    """
    Convert text to speech using form data (alternative endpoint)
    Useful for HTML forms or when JSON is not preferred
    """
    # Create request object and validate
    try:
        request = TTSRequest(text=text, voice=voice, pitch=pitch, rate=rate)
        return await synthesize_speech(request)
    except ValueError as e:
        raise HTTPException(status_code=422, detail=str(e))

@app.get("/", include_in_schema=False)
async def root():
    """Root endpoint redirects to API documentation"""
    return JSONResponse({
        "message": "Welcome to Text-to-Speech API",
        "documentation": "/docs",
        "health": "/health",
        "voices": "/voices",
        "synthesize": "/synthesize"
    })

# Exception handlers
@app.exception_handler(422)
async def validation_exception_handler(request, exc):
    return JSONResponse(
        status_code=422,
        content={"detail": "Validation error", "errors": exc.detail}
    )

@app.exception_handler(500)
async def internal_exception_handler(request, exc):
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error"}
    )

# Startup event
@app.on_event("startup")
async def startup_event():
    logger.info("TTS API is starting up...")
    # Test edge-tts functionality
    try:
        voices = await edge_tts.list_voices()
        logger.info(f"Successfully loaded {len(voices)} voices")
    except Exception as e:
        logger.error(f"Failed to load voices: {e}")

@app.on_event("shutdown")
async def shutdown_event():
    logger.info("TTS API is shutting down...")

if __name__ == "__main__":
    import uvicorn
    print("Starting TTS API Server with FastAPI...")
    print("API Documentation will be available at: http://localhost:7860/")
    print("Health check: http://localhost:7860/health")
    print("Available voices: http://localhost:7860/voices")
    print("\nExample usage (saves audio file locally):")
    print("curl -X POST 'http://localhost:7860/synthesize' \\")
    print("  -H 'Content-Type: application/json' \\")
    print("  -d '{\"text\":\"Hello from Hugging Face!\",\"voice\":\"en-GB-SoniaNeural\",\"pitch\":\"-10Hz\",\"rate\":\"+15%\"}' \\")
    print("  --output speech.mp3")
    print("\nFor your deployed space:")
    print("curl -X POST 'https://nitinbot001-tts-api.hf.space/synthesize' \\")
    print("  -H 'Content-Type: application/json' \\")
    print("  -d '{\"text\":\"hello my name is nitin\",\"voice\":\"en-US-AriaNeural\",\"pitch\":\"+0Hz\",\"rate\":\"+0%\"}' \\")
    print("  --output speech.mp3")
    
    uvicorn.run(app, host="0.0.0.0", port=7860)