server.py

"""
Noesis — micro-service TTS gratuit et modulaire (Pattern Strategy).

Expose un endpoint /tts interchangeable entre plusieurs moteurs :
  - edge   : voix neuronales Microsoft Edge (gratuit, sans clé, très naturel en FR)  [par défaut]
  - piper  : moteur ONNX ultra-léger, hors-ligne (optionnel — voir install plus bas)
  - kokoro : modèle 82M, plus riche (optionnel — voir install plus bas)

Lancement :
    pip install -r requirements.txt
    uvicorn server:app --host 0.0.0.0 --port 8008

L'app appelle :  http://127.0.0.1:8008/tts?engine=edge&voice=fr-FR-DeniseNeural&text=...
Renvoie un MP3 (audio/mpeg).
"""
import io, asyncio, os
from fastapi import FastAPI, Response, Query, HTTPException
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI(title="Noesis TTS")
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])

# ----------------------------------------------------------------------------
# Interface commune (Strategy)
#   Chaque moteur implémente synth(text, voice, rate) -> bytes (mp3/wav) + son media_type
# ----------------------------------------------------------------------------

# --- 1) EDGE-TTS (recommandé : gratuit, sans clé, voix FR très naturelles) ---
import edge_tts

async def _edge_synth(text: str, voice: str, rate: str) -> bytes:
    communicate = edge_tts.Communicate(text, voice, rate=rate)
    buf = io.BytesIO()
    async for chunk in communicate.stream():
        if chunk["type"] == "audio":
            buf.write(chunk["data"])
    return buf.getvalue()

@app.get("/voices")
async def voices():
    vs = await edge_tts.list_voices()
    fr = [v for v in vs if v["Locale"].startswith("fr")]
    return [{"name": v["ShortName"], "gender": v["Gender"], "locale": v["Locale"]} for v in fr]

# --- 2) PIPER (optionnel, hors-ligne). Installe : pip install piper-tts ------
#     Télécharge une voix FR (.onnx + .onnx.json), ex. fr_FR-siwis-medium,
#     depuis https://huggingface.co/rhasspy/piper-voices  et mets le chemin ci-dessous.
PIPER_MODEL = os.environ.get("PIPER_MODEL", "")  # ex: "voices/fr_FR-siwis-medium.onnx"
_piper = None
def _piper_synth(text: str) -> bytes:
    global _piper
    from piper.voice import PiperVoice
    import wave
    if _piper is None:
        if not PIPER_MODEL:
            raise HTTPException(500, "PIPER_MODEL non défini")
        _piper = PiperVoice.load(PIPER_MODEL)
    buf = io.BytesIO()
    with wave.open(buf, "wb") as wf:
        _piper.synthesize(text, wf)
    return buf.getvalue()

# --- 3) KOKORO (optionnel). Installe : pip install kokoro-onnx soundfile ------
#     Télécharge kokoro-v1.0.onnx + voices-v1.0.bin (repo hexgrad/Kokoro-82M).
_kokoro = None
def _kokoro_synth(text: str, voice: str) -> bytes:
    global _kokoro
    from kokoro_onnx import Kokoro
    import soundfile as sf
    if _kokoro is None:
        _kokoro = Kokoro("kokoro-v1.0.onnx", "voices-v1.0.bin")
    samples, sr = _kokoro.create(text, voice=voice or "ff_siwis", lang="fr-fr")
    buf = io.BytesIO()
    sf.write(buf, samples, sr, format="WAV")
    return buf.getvalue()

# ----------------------------------------------------------------------------
@app.get("/tts")
async def tts(
    text: str = Query(..., min_length=1),
    engine: str = "edge",
    voice: str = "fr-FR-DeniseNeural",
    rate: str = "+0%",
):
    if engine == "edge":
        data = await _edge_synth(text, voice, rate)
        return Response(content=data, media_type="audio/mpeg")
    elif engine == "piper":
        data = await asyncio.to_thread(_piper_synth, text)
        return Response(content=data, media_type="audio/wav")
    elif engine == "kokoro":
        data = await asyncio.to_thread(_kokoro_synth, text, voice)
        return Response(content=data, media_type="audio/wav")
    raise HTTPException(400, f"moteur inconnu: {engine}")

@app.get("/health")
def health():
    return {"ok": True}