Back-end/services/tts_service.py

# File: tts_service.py
from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse
from pydantic import BaseModel
from gtts import gTTS
import hashlib
import os
import sys
import logging
import uvicorn
from pathlib import Path
from fastapi.middleware.cors import CORSMiddleware

# Configuration du logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    stream=sys.stdout
)
logger = logging.getLogger("TTS_Service")

app = FastAPI()

# Chemins absolus
BASE_DIR = Path(__file__).parent.parent.resolve()
CACHE_DIR = BASE_DIR / "tts_cache"
CACHE_DIR.mkdir(exist_ok=True)

# Configuration CORS
origins = ["*"]  # Autoriser toutes les origines pour les tests

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Langues supportées
SUPPORTED_LANGUAGES = {"fr", "en", "ar", "es", "de", "it"}

class TTSRequest(BaseModel):
    text: str
    lang: str

@app.post("/generate-tts")
async def generate_tts(request: TTSRequest):
    text = request.text.strip()
    lang = request.lang.strip().lower()
    
    # Validation des entrées
    if not text:
        raise HTTPException(status_code=400, detail="Le texte ne peut pas être vide")
    if lang not in SUPPORTED_LANGUAGES:
        raise HTTPException(
            status_code=400, 
            detail=f"Langue non supportée. Choisissez parmi: {', '.join(SUPPORTED_LANGUAGES)}"
        )

    logger.info(f"Génération TTS: {len(text)} caractères en {lang}")

    # Génération du hash
    normalized_text = text.replace(" ", "_").lower()
    text_hash = hashlib.md5(f"{normalized_text}_{lang}".encode('utf-8')).hexdigest()
    filename = f"{text_hash}.mp3"
    filepath = CACHE_DIR / filename

    # Utilisation du cache si disponible
    if filepath.exists():
        logger.info(f"Fichier existant: {filepath}")
        return {
            "audioPath": str(filepath.relative_to(BASE_DIR)),
            "url": f"/audio/{filepath.relative_to(BASE_DIR)}"
        }

    # Génération du fichier audio
    try:
        tts = gTTS(text=text, lang=lang)
        tts.save(str(filepath))
        logger.info(f"Fichier généré: {filepath}")
        
        return {
            "audioPath": str(filepath.relative_to(BASE_DIR)),
            "url": f"/audio/{filepath.relative_to(BASE_DIR)}"
        }
    except Exception as e:
        logger.error(f"Erreur TTS: {str(e)}", exc_info=True)
        raise HTTPException(
            status_code=500, 
            detail=f"Échec de génération audio: {str(e)}"
        )
    
@app.get("/audio/{file_path:path}")
async def get_audio(file_path: str):
    # Sécurité: empêcher les accès en dehors du cache
    if ".." in file_path or file_path.startswith("/"):
        raise HTTPException(status_code=403, detail="Accès non autorisé")
    
    full_path = BASE_DIR / file_path
    
    # Vérifier que le chemin est dans le cache
    if not str(full_path).startswith(str(CACHE_DIR)):
        raise HTTPException(status_code=403, detail="Accès non autorisé")
    
    if not full_path.exists():
        raise HTTPException(status_code=404, detail="Fichier audio non trouvé")
    
    return FileResponse(full_path, media_type="audio/mpeg")

@app.get("/health")
async def health_check():
    return {"status": "ok", "service": "tts"}

def run_service():
    uvicorn.run(app, host="0.0.0.0", port=5000)

if __name__ == "__main__":
    run_service()