feat: add voiceclone bridge (tts_bridge.py)

tts_bridge.py

@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+import hmac
+import os
+from contextlib import asynccontextmanager
+from pathlib import Path
+
+import httpx
+from fastapi import Depends, FastAPI, Header, HTTPException
+from fastapi.responses import Response
+from pydantic import BaseModel, Field
+
+BUTLER_URL = os.environ.get("BUTLER_URL", "http://10.4.1.116:8888").rstrip("/")
+BUTLER_TOKEN_FILE = Path(os.environ.get("BUTLER_TOKEN_FILE", "/run/secrets/butler_token"))
+CLIENT_TOKEN_FILE = Path(os.environ.get("CLIENT_TOKEN_FILE", "/run/secrets/client_token"))
+DEFAULT_VOICE = os.environ.get("DEFAULT_VOICE", "deep_thought.mp3")
+
+
+def _read_secret(path: Path, label: str) -> str:
+    try:
+        value = path.read_text().strip()
+    except OSError as exc:
+        raise RuntimeError(f"{label} secret is unavailable") from exc
+    if not value:
+        raise RuntimeError(f"{label} secret is empty")
+    return value
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    app.state.butler_token = _read_secret(BUTLER_TOKEN_FILE, "Butler")
+    app.state.client_token = _read_secret(CLIENT_TOKEN_FILE, "Client")
+    app.state.http = httpx.AsyncClient(timeout=httpx.Timeout(190.0, connect=10.0))
+    yield
+    await app.state.http.aclose()
+
+
+app = FastAPI(title="Pfannkuchen TTS Bridge", version="1.0.0", lifespan=lifespan, docs_url=None, redoc_url=None, openapi_url=None)
+
+
+class GenerateRequest(BaseModel):
+    text: str = Field(min_length=1, max_length=2000)
+    voice: str = Field(default=DEFAULT_VOICE, pattern=r"^[A-Za-z0-9_.-]+$")
+    language: str = Field(default="de", pattern=r"^[A-Za-z]{2,8}(?:-[A-Za-z0-9]{2,8})?$")
+
+
+def verify_client(authorization: str | None = Header(default=None)) -> None:
+    if not authorization or not authorization.startswith("Bearer "):
+        raise HTTPException(status_code=401, detail="Missing bearer token")
+    supplied = authorization.removeprefix("Bearer ").strip()
+    if not hmac.compare_digest(supplied, app.state.client_token):
+        raise HTTPException(status_code=401, detail="Invalid bearer token")
+
+
+@app.get("/health")
+async def health():
+    try:
+        response = await app.state.http.get(f"{BUTLER_URL}/tts/health", headers={"Authorization": f"Bearer {app.state.butler_token}"})
+        response.raise_for_status()
+        chatterbox_ok = response.json().get("chatterbox") == "ok"
+    except (httpx.HTTPError, ValueError):
+        chatterbox_ok = False
+    return {"status": "ok" if chatterbox_ok else "degraded", "chatterbox": chatterbox_ok}
+
+
+@app.post("/v1/tts", response_class=Response, dependencies=[Depends(verify_client)])
+async def generate(req: GenerateRequest):
+    try:
+        upstream = await app.state.http.post(f"{BUTLER_URL}/tts/generate", headers={"Authorization": f"Bearer {app.state.butler_token}"}, json=req.model_dump())
+    except httpx.RequestError:
+        raise HTTPException(status_code=502, detail="Butler is unavailable")
+    if upstream.status_code != 200:
+        raise HTTPException(status_code=502, detail="Speech generation failed")
+    if not upstream.content.startswith(b"RIFF"):
+        raise HTTPException(status_code=502, detail="Invalid audio received")
+    return Response(content=upstream.content, media_type="audio/wav", headers={"Content-Disposition": 'inline; filename="voiceclone.wav"', "Cache-Control": "no-store", "X-Content-Type-Options": "nosniff"})