import asyncio
import html as html_module
import logging
import re
import struct
import subprocess
import tempfile
from pathlib import Path
from typing import Dict, List, Optional
from urllib.parse import quote
import httpx
from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
from fastapi.responses import Response
from pydantic import BaseModel
from app.core.auth import get_current_active_user
from app.models.user import User
from app.config import get_settings
logger = logging.getLogger(__name__)
router = APIRouter()
voice_settings = get_settings().voice
TTS_BASE = voice_settings.tts_endpoint
STT_BASE = voice_settings.stt_endpoint
HTTP_TIMEOUT = 120
# Coqui uses GET /synthesize/{text}; keep chunks small for URL limits.
MAX_SYNTH_CHUNK = 200
_SECTION_HEADERS = re.compile(
r"\b(Thought|What to do|Next step)\s*[:.]?\s*", flags=re.IGNORECASE
)
class TTSRequest(BaseModel):
text: str
def _rough_spoken_text(raw: str) -> str:
"""Light cleanup so model markdown is tolerable for TTS (no extra deps)."""
t = raw.strip()
t = re.sub(r"```[\s\S]*?```", " ", t)
t = re.sub(r"`([^`]+)`", r"\1", t)
t = re.sub(r"\[([^\]]+)\]\([^)]+\)", r"\1", t)
t = re.sub(r"!\[([^\]]*)\]\([^)]+\)", r"\1", t)
t = re.sub(r"^#{1,6}\s+", "", t, flags=re.MULTILINE)
t = re.sub(r"[*_]{1,3}", "", t)
t = html_module.unescape(t)
t = _SECTION_HEADERS.sub(" ", t)
t = re.sub(r"\s+", " ", t).strip()
return t
def _text_chunks(text: str, max_len: int = MAX_SYNTH_CHUNK) -> List[str]:
text = text.strip()
if not text:
return []
out: List[str] = []
while text:
if len(text) <= max_len:
out.append(text)
break
cut = text.rfind(" ", 0, max_len)
if cut <= 0:
cut = max_len
piece = text[:cut].strip()
if piece:
out.append(piece)
text = text[cut:].strip()
return out
async def _synthesize_one(client: httpx.AsyncClient, chunk: str) -> Optional[bytes]:
url = f"{TTS_BASE}/synthesize/{quote(chunk, safe='')}"
try:
r = await client.get(url)
r.raise_for_status()
return r.content
except Exception as exc:
logger.warning("TTS chunk failed (%s chars): %s", len(chunk), exc)
return None
def _concat_wav(segments: List[bytes]) -> bytes:
if len(segments) == 1:
return segments[0]
pcm_parts: List[bytes] = []
first_header = b""
for i, seg in enumerate(segments):
# PCM WAV files begin with a 44-byte header; if the segment is shorter,
# then it cannot be valid WAV audio
if len(seg) < 44:
continue
if i == 0:
first_header = seg[:44]
data_offset = 44
idx = seg.find(b"data")
if idx != -1 and idx + 8 <= len(seg):
data_offset = idx + 8
pcm_parts.append(seg[data_offset:])
if not pcm_parts or not first_header:
return segments[0] if segments else b""
all_pcm = b"".join(pcm_parts)
pcm_len = len(all_pcm)
header = bytearray(first_header)
struct.pack_into(" bytes:
with tempfile.TemporaryDirectory() as tmp:
ext = "webm" if "webm" in (src_mime or "") else "ogg"
src = Path(tmp) / f"in.{ext}"
dst = Path(tmp) / "out.wav"
src.write_bytes(audio_bytes)
result = subprocess.run(
[
"ffmpeg",
"-y",
"-i",
str(src),
"-ar",
"16000",
"-ac",
"1",
"-f",
"wav",
str(dst),
],
capture_output=True,
timeout=30,
)
if result.returncode != 0:
logger.warning(
"ffmpeg stderr: %s",
result.stderr.decode(errors="replace")[-500:],
)
raise RuntimeError("ffmpeg conversion failed")
return dst.read_bytes()
@router.get("/voice/status")
async def voice_status(
current_user: User = Depends(get_current_active_user),
) -> Dict[str, bool]:
tts_ready = False
stt_ready = False
async with httpx.AsyncClient(timeout=5) as client:
if TTS_BASE not in ("", None):
try:
resp = await client.get(f"{TTS_BASE}/status")
tts_ready = resp.status_code == 200
except Exception:
tts_ready = False
if STT_BASE not in ("", None):
try:
resp = await client.get(f"{STT_BASE}/status")
stt_ready = resp.status_code == 200
except Exception:
stt_ready = False
return {"tts_ready": tts_ready, "stt_ready": stt_ready}
@router.post("/voice/transcribe")
async def transcribe_audio(
audio: UploadFile = File(...),
current_user: User = Depends(get_current_active_user),
) -> Dict[str, str]:
"""
Transcribe an uploaded audio file and return the text.
@param audio: The uploaded audio file (from a browser recording)
@param current_user: The authenticated user making the request
@return: A dictionary with the transcribed text under key "text"
"""
contents = await audio.read()
if not contents:
return {"text": ""}
mime = audio.content_type or "audio/webm"
logger.info("STT: received %s bytes (%s)", len(contents), mime)
need_convert = "wav" not in mime.lower()
if need_convert:
try:
loop = asyncio.get_running_loop()
wav_bytes = await loop.run_in_executor(
None, _convert_to_wav, contents, mime
)
logger.info("STT: converted to WAV (%s bytes)", len(wav_bytes))
except Exception as e:
logger.error("STT conversion error: %s", e)
raise HTTPException(status_code=500, detail="Audio conversion failed")
else:
wav_bytes = contents
try:
async with httpx.AsyncClient(timeout=HTTP_TIMEOUT) as client:
resp = await client.post(
f"{STT_BASE}/stt",
content=wav_bytes,
headers={"Content-Type": "audio/wav"},
)
resp.raise_for_status()
text = resp.text.strip().strip('"')
logger.info("STT result: %r", text[:100])
return {"text": text}
except httpx.TimeoutException:
raise HTTPException(status_code=504, detail="STT service timed out")
except Exception as e:
logger.error("STT proxy error: %s", e)
raise HTTPException(status_code=502, detail="STT service unavailable")
@router.post("/voice/tts")
async def text_to_speech(
req: TTSRequest,
current_user: User = Depends(get_current_active_user),
) -> Response:
"""
Generate spoken audio for the input text
@param req: A TTSRequest containing the text to synthesize
@param current_user: The authenticated user making the request
@return: A Response containing the synthesized audio in WAV format
"""
raw = req.text.strip()
if not raw:
raise HTTPException(status_code=400, detail="Text is required")
if len(raw) > 5000:
raw = raw[:5000]
spoken = _rough_spoken_text(raw)
chunks = _text_chunks(spoken)
logger.info(
"TTS: %s chunk(s) from %s input chars → %s spoken chars",
len(chunks),
len(req.text),
len(spoken),
)
try:
async with httpx.AsyncClient(timeout=HTTP_TIMEOUT) as client:
results = await asyncio.gather(
*[_synthesize_one(client, c) for c in chunks]
)
wav_segments = [r for r in results if r and len(r) > 44]
if not wav_segments:
raise HTTPException(
status_code=502, detail="TTS synthesis failed for all chunks"
)
combined = _concat_wav(wav_segments)
return Response(content=combined, media_type="audio/wav")
except HTTPException:
raise
except httpx.TimeoutException:
raise HTTPException(status_code=504, detail="TTS service timed out")
except Exception as e:
logger.error("TTS proxy error: %s", e)
raise HTTPException(status_code=502, detail="TTS service unavailable")