tts-voice-creator-clone-and-design-2/routes/settings.py

"""Settings, TTS routes, routing log, server logs, and voice design presets endpoints."""
from __future__ import annotations

from fastapi import APIRouter, HTTPException, Request, Response

from core.config import _load_settings, _save_settings, _normalize_settings, _SETTINGS_KEYS, _ensure_external_api_key
from core.routing import _load_tts_routes, _save_tts_routes
from core.constants import _log_buffer, _LOG_BUFFER_MAX, _routing_log, _ROUTING_LOG_MAX
from core.presets import _load_design_presets, _save_design_presets

router = APIRouter()


_SECRET_SETTING_KEYS = {
    "groq_api_key", "whisper_api_key", "tts_api_key", "voice_design_api_key",
    "elevenlabs_api_key", "llm_api_key", "engine_api_keys", "external_api_key",
}
_SECRET_MASK = "__configured__"


@router.get("/api/settings")
async def get_settings(request: Request, response: Response):
    s = _load_settings()
    # Lazily generated on first read, not at server startup — a fresh
    # install shows a real usable key in Settings immediately without a
    # migration step, whether or not the gate is actually turned on yet.
    s["external_api_key"] = _ensure_external_api_key()
    # Never return provider credentials to browser JavaScript.  The mask lets
    # the Settings UI indicate that a value exists without making it usable in
    # client-side requests or browser storage.
    for key in _SECRET_SETTING_KEYS:
        value = s.get(key)
        if key == "engine_api_keys" and isinstance(value, dict):
            s[key] = {name: _SECRET_MASK for name, secret in value.items() if secret}
        elif value:
            s[key] = _SECRET_MASK
        else:
            s.pop(key, None)
    response.headers["Cache-Control"] = "no-store, no-cache, max-age=0"
    response.headers["Pragma"] = "no-cache"
    return s


@router.post("/api/settings/regenerate-api-key")
async def regenerate_api_key(response: Response):
    import secrets
    s = _load_settings()
    s["external_api_key"] = secrets.token_urlsafe(32)
    _save_settings(s)
    response.headers["Cache-Control"] = "no-store, no-cache, max-age=0"
    response.headers["Pragma"] = "no-cache"
    return {"external_api_key": s["external_api_key"]}


@router.post("/api/settings")
async def post_settings(request: Request):
    data = await request.json()
    s = _load_settings()
    clear = {str(k) for k in (data.get("clear_secrets") or [])} if isinstance(data.get("clear_secrets"), list) else set()
    for key, value in data.items():
        if key not in _SETTINGS_KEYS:
            continue
        if key in _SECRET_SETTING_KEYS:
            if value == _SECRET_MASK:
                continue
            if key == "engine_api_keys" and isinstance(value, dict):
                old = s.get(key) if isinstance(s.get(key), dict) else {}
                merged = dict(old)
                for name, secret in value.items():
                    if secret and secret != _SECRET_MASK:
                        merged[name] = secret
                for name in clear:
                    merged.pop(name, None)
                s[key] = merged
                continue
            if value in (None, "") and key not in clear:
                continue
        s[key] = value
    s = _normalize_settings(s)
    _save_settings(s)
    return {"ok": True}


@router.get("/api/tts-routes")
async def get_tts_routes():
    return {"routes": _load_tts_routes()}


@router.post("/api/tts-routes")
async def post_tts_routes(request: Request):
    data = await request.json()
    routes = data.get("routes", data) if isinstance(data, dict) else data
    if not isinstance(routes, list):
        raise HTTPException(400, "routes must be a list")
    _save_tts_routes(routes)
    return {"ok": True, "routes": _load_tts_routes()}


@router.get("/api/tts-routing-log")
async def get_tts_routing_log(limit: int = 60):
    limit = max(1, min(int(limit or 60), _ROUTING_LOG_MAX))
    return {"items": _routing_log[:limit], "max": _ROUTING_LOG_MAX}


@router.delete("/api/tts-routing-log")
async def clear_tts_routing_log():
    _routing_log.clear()
    return {"ok": True, "items": []}


@router.get("/api/logs")
async def get_server_logs(limit: int = 200):
    limit = max(1, min(int(limit or 200), _LOG_BUFFER_MAX))
    return {"items": _log_buffer[:limit], "max": _LOG_BUFFER_MAX}


@router.delete("/api/logs")
async def clear_server_logs():
    _log_buffer.clear()
    return {"ok": True}


@router.get("/api/voice-design-presets")
async def get_voice_design_presets():
    return _load_design_presets()


@router.post("/api/voice-design-presets")
async def post_voice_design_presets(request: Request):
    data = await request.json()
    if not isinstance(data, dict):
        raise HTTPException(400, "Expected a preset object")
    _save_design_presets(data)
    return {"ok": True, "presets": _load_design_presets()}