133 lines
5.5 KiB
Python
133 lines
5.5 KiB
Python
"""Voice archetype gallery: catalog, cached preview rendering, save-as-designed-voice."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import io
|
|
import logging
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, HTTPException, Query
|
|
from fastapi.responses import FileResponse
|
|
|
|
from core import voice_archetypes as va
|
|
from core.audio import _export_normalized_wav
|
|
from core.config import _load_settings
|
|
from core.constants import CONFIG_DIR, _VOICE_DESIGN_MODEL, _VOICES_DIR_DEFAULT
|
|
from core.provenance import validate_provenance
|
|
from core.voice import _active_voices_dir, _find_voice_audio, _load_meta, _save_meta
|
|
from core.voice_index import upsert_voice_in_index
|
|
|
|
logger = logging.getLogger("uvicorn.error")
|
|
router = APIRouter()
|
|
|
|
RENDER_TIMEOUT_S = 240
|
|
MAX_PREVIEW_BYTES = 8 * 1024 * 1024
|
|
_LOCKS: dict[str, asyncio.Lock] = {}
|
|
|
|
|
|
def cache_dir() -> Path:
|
|
return CONFIG_DIR / "archetype_previews" # config/ is git-ignored; catalog is small so the cache is bounded
|
|
|
|
|
|
def _get(aid: str) -> dict:
|
|
a = va.BY_ID.get(aid)
|
|
if a is None:
|
|
raise HTTPException(404, "Unknown archetype")
|
|
return a
|
|
|
|
|
|
def _sample_text(a: dict) -> str:
|
|
return a["sample_script_de"] if a["language"] == "German" else a["sample_script_en"]
|
|
|
|
|
|
def _is_blank(audio: bytes) -> bool:
|
|
from pydub import AudioSegment
|
|
seg = AudioSegment.from_file(io.BytesIO(audio))
|
|
return len(seg) < 300 or seg.max_possible_amplitude == 0 or seg.rms < 20
|
|
|
|
|
|
def _render(a: dict, settings: dict) -> bytes:
|
|
"""Blocking synthesis through the existing voice-design path, seed pinned."""
|
|
from core.tts_helpers import _voice_design_request_audio
|
|
pinned = {**settings, "tts_stability_enabled": True,
|
|
"tts_extra_params_by_backend": {"voice_design": {"seed": va.SEED}}}
|
|
gender = {"male": "M", "female": "F"}.get(a["facets"]["gender"], "")
|
|
audio, _ = _voice_design_request_audio(a["instruct"], _sample_text(a), a["language"], pinned, gender)
|
|
return audio
|
|
|
|
|
|
async def _cached_preview(a: dict) -> Path:
|
|
key = va.preview_key(a, _VOICE_DESIGN_MODEL)
|
|
path = cache_dir() / f"{key[:32]}.wav"
|
|
if path.exists():
|
|
return path
|
|
lock = _LOCKS.setdefault(key, asyncio.Lock())
|
|
async with lock:
|
|
if path.exists():
|
|
return path
|
|
from routes.tts import _bounded_tts_call
|
|
try:
|
|
audio = await asyncio.wait_for(_bounded_tts_call(_render, a, _load_settings()), RENDER_TIMEOUT_S)
|
|
except HTTPException:
|
|
raise
|
|
except asyncio.TimeoutError:
|
|
raise HTTPException(504, "Preview rendering timed out")
|
|
except Exception as exc:
|
|
logger.warning("archetype preview failed (%s): %s", a["id"], exc)
|
|
raise HTTPException(502, "Preview rendering failed")
|
|
if not audio or len(audio) > MAX_PREVIEW_BYTES or _is_blank(audio):
|
|
raise HTTPException(502, "Engine returned blank or oversized audio")
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
fd, tmp = tempfile.mkstemp(dir=path.parent, suffix=".tmp")
|
|
with open(fd, "wb") as fh:
|
|
fh.write(audio)
|
|
Path(tmp).replace(path)
|
|
return path
|
|
|
|
|
|
@router.get("/api/archetypes")
|
|
async def archetypes_list(gender: str = "", age: str = "", pitch: str = "", accent: str = "",
|
|
language: str = "", category: str = "", q: str = "",
|
|
offset: int = Query(0, ge=0), limit: int = Query(24, ge=1, le=va.MAX_LIMIT)):
|
|
res = va.list_archetypes(gender, age, pitch, accent, language, category, q, offset, limit)
|
|
res["categories"] = list(va.CATEGORIES)
|
|
res["cached"] = [a["id"] for a in res["items"] if (cache_dir() / f"{va.preview_key(a, _VOICE_DESIGN_MODEL)[:32]}.wav").exists()]
|
|
return res
|
|
|
|
|
|
@router.post("/api/archetypes/{aid}/preview")
|
|
async def archetype_preview(aid: str):
|
|
return FileResponse(await _cached_preview(_get(aid)), media_type="audio/wav")
|
|
|
|
|
|
@router.post("/api/archetypes/{aid}/use")
|
|
async def archetype_use(aid: str):
|
|
a = _get(aid)
|
|
voice_id = f"archetype_{aid}"
|
|
settings = _load_settings()
|
|
existing = _find_voice_audio(voice_id, Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT)))
|
|
if existing is not None: # idempotent per archetype id
|
|
return {"voice_id": voice_id, "created": False, "instruct": a["instruct"], "language": a["language"]}
|
|
src = await _cached_preview(a)
|
|
out_dir = _active_voices_dir(settings)
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
wav = out_dir / f"{voice_id}.wav"
|
|
text = _sample_text(a)
|
|
await asyncio.to_thread(_export_normalized_wav, src, wav)
|
|
wav.with_suffix(".reference.txt").write_text(text, encoding="utf-8")
|
|
meta = _load_meta(wav)
|
|
meta.update({
|
|
"enabled": True, "origin": "designed", "name": a["name_de"], "gender": a["facets"]["gender"][:1].upper(),
|
|
"lang": a["language"], "transcript": text, "voice_design_prompt": a["instruct"],
|
|
"note": "Voice Archetype: " + a["instruct"][:240],
|
|
# no 'synthetic' basis exists in core.provenance.BASES; 'unknown' is the closest honest value
|
|
# (no real person, no consent/licence needed). ':' is avoided in source because
|
|
# validate_provenance parses 'x:' as a URL scheme.
|
|
"provenance": validate_provenance(
|
|
{"basis": "unknown", "source": f"archetype/{aid}", "note": "synthetic designed voice"}),
|
|
})
|
|
_save_meta(wav, meta)
|
|
await asyncio.to_thread(upsert_voice_in_index, settings, wav)
|
|
return {"voice_id": voice_id, "created": True, "instruct": a["instruct"], "language": a["language"]}
|