pdf-quiz-generator/backend/app/routers/tts.py

from fastapi import APIRouter, Depends, HTTPException, UploadFile, File
from fastapi.responses import Response
from pydantic import BaseModel
from sqlalchemy.orm import Session

from app.database import get_db
from app.models.user import User
from app.models.ai_model_config import AIModelConfig
from app.services import ai_service, tts_voices
from app.utils.auth import get_current_user, check_rate_limit

router = APIRouter()


class TTSRequest(BaseModel):
    text: str
    voice: str | None = None  # model_id override


MAX_AUDIO_UPLOAD_BYTES = 25 * 1024 * 1024


def _task_model(db: Session, task: str, fallback: str) -> tuple[str, str | None]:
    config = db.query(AIModelConfig).filter(
        AIModelConfig.task == task,
        AIModelConfig.is_active == True,
        AIModelConfig.is_default == True,
    ).first()
    if config:
        return config.model_id, config.api_key or None
    return fallback, None


def _default_tts_model(db: Session) -> tuple[str, str | None]:
    # Whatever the administrator marked default. The "local-%" filter that
    # used to sit here meant an Orpheus default was ignored in favour of the
    # hard-coded Kokoro voice below, silently.
    config = db.query(AIModelConfig).filter(
        AIModelConfig.task == "tts",
        AIModelConfig.is_active == True,
        AIModelConfig.is_default == True,
    ).first()
    if config:
        return config.model_id, config.api_key or None
    return "local-kokoro-tts:am_adam", None


@router.get("/voices")
def get_voices(
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Every active voice, each saying which model serves it.

    All of them, not only the locally served ones: a learner is offered every
    voice an administrator added, whichever model it belongs to.
    """
    query = db.query(AIModelConfig).filter(
        AIModelConfig.task == "tts",
        AIModelConfig.is_active == True,
    )
    db_models = query.order_by(AIModelConfig.is_default.desc(), AIModelConfig.name).all()
    return [{"id": m.model_id, "name": m.name, "is_default": m.is_default,
             "model": tts_voices.split(m.model_id)[0]} for m in db_models]


@router.post("/speak")
def text_to_speech(
    request: TTSRequest,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Convert text to speech using configured or user-selected TTS model."""
    # Rate limit: 60 TTS requests per user per hour (admins/unthrottled users exempt)
    check_rate_limit(
        key=f"tts_speak:{current_user.id}",
        max_calls=240,
        window_seconds=3600,
        detail="You've reached the audio limit. The limit resets automatically — try again shortly. Contact an admin if you need this raised.",
        user=current_user,
    )
    if not request.text.strip():
        raise HTTPException(status_code=400, detail="Text cannot be empty")

    text = request.text[:2000]

    # The player no longer asks, so the learner's own setting is what decides —
    # falling through to whatever an administrator marked default when they have
    # never chosen one. A caller may still name a voice, which is how a preview
    # in Settings plays the one being considered rather than the one in force.
    wanted = request.voice or current_user.tts_voice
    # Any active voice, not only a locally served one. The prefix check that
    # used to guard this meant a site adding a hosted voice would offer it in
    # Settings, save the learner's choice, and then quietly read every question
    # in the default voice instead.
    config = db.query(AIModelConfig).filter(
        AIModelConfig.task == "tts",
        AIModelConfig.is_active == True,
        AIModelConfig.model_id == wanted,
    ).first() if wanted else None
    if config:
        model_id, api_key = config.model_id, config.api_key or None
    else:
        model_id, api_key = _default_tts_model(db)

    audio = ai_service.generate_tts_audio(text, model_id=model_id, api_key=api_key)
    if audio is None:
        raise HTTPException(status_code=500, detail="TTS generation failed. Check model configuration.")

    return Response(content=audio, media_type=tts_voices.media_type(model_id))


@router.post("/transcribe")
def speech_to_text(
    file: UploadFile = File(...),
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Transcribe microphone audio using the configured STT model."""
    check_rate_limit(
        key=f"stt_transcribe:{current_user.id}",
        max_calls=120,
        window_seconds=3600,
        detail="You've reached the speech transcription limit. Try again shortly.",
        user=current_user,
    )
    audio = file.file.read()
    if not audio:
        raise HTTPException(status_code=400, detail="Audio file is empty")
    if len(audio) > MAX_AUDIO_UPLOAD_BYTES:
        raise HTTPException(status_code=413, detail="Audio file is too large")

    model_id, api_key = _task_model(db, "stt", "local-parakeet-v3")
    text = ai_service.transcribe_audio(
        audio,
        filename=file.filename or "audio.webm",
        content_type=file.content_type or "audio/webm",
        model_id=model_id,
        api_key=api_key,
    )
    if text is None:
        raise HTTPException(status_code=502, detail="Speech transcription failed. Check STT model configuration.")
    return {"text": text, "model": model_id}