mARTin-B78
c7a1e35539
Security fixes: - Block /proc /sys /dev /run /boot in /api/browse-dirs (path traversal) - Verify yt-dlp output stays inside TEMP_DIR before registration - Remove Access-Control-Allow-Origin: * from /api/proxy-audio - TTL-based temp file registry (default 2h) to prevent disk fill Performance: - Cache settings + routing rules in memory (mtime-checked); eliminates per-request disk reads on every TTS call UI: - Add container name (optional) field to Docker stack TTS/STT engine cards (Qwen3 Voice Clone, Voice Design, Custom Voice, Streaming, NVIDIA Magpie, Parakeet) — enables Stop/Start/Restart buttons on all engine cards, matching the existing Other Local TTS/STT cards Refactor — backend: - server.py: 5560 lines → 43-line entry point - core/ package: constants, registry, validation, docker_client, config, routing, audio, voice, presets, tts_helpers - routes/ package: admin, settings, library, stt, sources, docker, tts, conversation (FastAPI APIRouter modules) - Dockerfile + docker-compose.yml updated to include core/ and routes/ Refactor — frontend: - static/app.js: 8744 lines → 16 modules in static/js/ utils, voice-inspector, voice-sources, integrations, routing, settings, voice-clone, voice-library, tts-preview, benchmark, stt, init, engines, ai-backends, generation, conversation - static/loader.js updated to load modules sequentially Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
591 lines
24 KiB
Python
591 lines
24 KiB
Python
"""TTS request helpers: config, audio request, streaming, voice design, NVIDIA, benchmark."""
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import struct
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
from pydub import AudioSegment
|
|
|
|
from core.config import (
|
|
_clean_preview_backend,
|
|
_preview_backend_base_url,
|
|
_tts_extra_params,
|
|
_apply_tts_extra_params,
|
|
_post_tts_with_fallback,
|
|
)
|
|
from core.constants import (
|
|
_VOICES_DIR_DEFAULT,
|
|
_VOICE_DESIGN_DEFAULT,
|
|
_VOICE_DESIGN_MODEL,
|
|
_MAX_TTS_OUTPUT_SECONDS,
|
|
)
|
|
from core.audio import _to_wav_24k, _duration as _dur
|
|
from core.voice import _find_voice_audio, _read_reference_text
|
|
|
|
|
|
# ── NVIDIA language codes ─────────────────────────────────────────────────────
|
|
|
|
_NVIDIA_LANGUAGE_CODES = {
|
|
"EN": "en-US", "DE": "de-DE", "ES": "es-ES", "FR": "fr-FR",
|
|
"IT": "it-IT", "PT": "pt-PT", "NL": "nl-NL", "PL": "pl-PL",
|
|
}
|
|
|
|
|
|
def _nvidia_clone_language_code(voice: str, language: str = "") -> str:
|
|
raw = str(language or "").strip()
|
|
if raw and raw.lower() not in {"auto", "*"}:
|
|
return raw
|
|
prefix = str(voice or "").split("_", 1)[0].upper()
|
|
return _NVIDIA_LANGUAGE_CODES.get(prefix, "en-US")
|
|
|
|
|
|
def _form_value(value) -> str:
|
|
if isinstance(value, bool):
|
|
return "true" if value else "false"
|
|
return str(value)
|
|
|
|
|
|
# ── WAV helpers ───────────────────────────────────────────────────────────────
|
|
|
|
def _wav_data_offset(data: bytes) -> int | None:
|
|
if len(data) < 12:
|
|
return None
|
|
if data[:4] != b"RIFF" or data[8:12] != b"WAVE":
|
|
return 0
|
|
pos = 12
|
|
while pos + 8 <= len(data):
|
|
chunk_sz = struct.unpack_from("<I", data, pos + 4)[0]
|
|
if data[pos:pos + 4] == b"data":
|
|
return pos + 8
|
|
pos += 8 + chunk_sz + (chunk_sz % 2)
|
|
return None
|
|
|
|
|
|
def _audio_duration_from_bytes(audio: bytes, media_type: str) -> float | None:
|
|
try:
|
|
source_format = "wav" if audio[:4] == b"RIFF" or "wav" in media_type.lower() else None
|
|
return len(AudioSegment.from_file(io.BytesIO(audio), format=source_format)) / 1000.0
|
|
except Exception:
|
|
offset = _wav_data_offset(audio)
|
|
if offset is not None and offset > 0 and len(audio) > offset:
|
|
return (len(audio) - offset) / (24000 * 2)
|
|
return None
|
|
|
|
|
|
# ── TTS request config & audio ────────────────────────────────────────────────
|
|
|
|
def _tts_request_config(
|
|
text: str,
|
|
voice: str,
|
|
settings: dict,
|
|
response_format: str = "wav",
|
|
instruct: str = "",
|
|
url_override: str = "",
|
|
api_key_override: str | None = None,
|
|
backend_override: str = "",
|
|
extra_backend: str = "",
|
|
) -> tuple[str, dict, dict]:
|
|
from core.constants import _TTS_DEFAULT
|
|
from core.validation import _validate_http_url
|
|
|
|
tts_url = _validate_http_url(url_override or settings.get("tts_url", _TTS_DEFAULT), allow_private=True).rstrip("/")
|
|
backend = backend_override or settings.get("tts_backend", "openai")
|
|
tts_key = (api_key_override if api_key_override is not None else settings.get("tts_api_key", "")).strip()
|
|
tts_hdrs = {"Authorization": f"Bearer {tts_key}"} if tts_key else {}
|
|
|
|
if backend == "localai":
|
|
endpoint, payload = f"{tts_url}/tts", {"input": text, "model": voice, "response_format": response_format}
|
|
elif backend == "pocket":
|
|
endpoint, payload = f"{tts_url}/v1/audio/speech", {"input": text, "voice": voice, "response_format": response_format}
|
|
else:
|
|
endpoint, payload = f"{tts_url}/v1/audio/speech", {"model": "tts-1", "input": text, "voice": voice, "response_format": response_format}
|
|
if instruct.strip():
|
|
payload["instruct"] = instruct.strip()
|
|
_apply_tts_extra_params(payload, settings, extra_backend or backend_override or "voice_clone")
|
|
return endpoint, payload, tts_hdrs
|
|
|
|
|
|
def _tts_request_audio(
|
|
text: str,
|
|
voice: str,
|
|
settings: dict,
|
|
instruct: str = "",
|
|
url_override: str = "",
|
|
api_key_override: str | None = None,
|
|
backend_override: str = "",
|
|
extra_backend: str = "",
|
|
) -> tuple[bytes, str]:
|
|
endpoint, payload, tts_hdrs = _tts_request_config(
|
|
text, voice, settings, "wav", instruct, url_override, api_key_override, backend_override, extra_backend
|
|
)
|
|
resp = _post_tts_with_fallback(endpoint, payload, tts_hdrs, timeout=120)
|
|
resp.raise_for_status()
|
|
audio = resp.content
|
|
if not audio:
|
|
raise RuntimeError("backend returned empty audio")
|
|
media_type = resp.headers.get("content-type", "audio/wav").split(";", 1)[0] or "audio/wav"
|
|
return audio, media_type
|
|
|
|
|
|
# ── TTS streaming helpers ─────────────────────────────────────────────────────
|
|
|
|
def _tts_stream_request_config(text: str, voice: str, settings: dict, instruct: str = "") -> tuple[str, dict, dict]:
|
|
from core.constants import _TTS_STREAM_DEFAULT
|
|
from core.validation import _validate_http_url
|
|
|
|
stream_url = settings.get("tts_stream_url") or settings.get("tts_url") or _TTS_STREAM_DEFAULT
|
|
tts_url = _validate_http_url(stream_url, allow_private=True).rstrip("/")
|
|
tts_key = settings.get("tts_api_key", "").strip()
|
|
tts_hdrs = {"Authorization": f"Bearer {tts_key}"} if tts_key else {}
|
|
payload = {"model": "tts-1", "input": text, "voice": voice, "response_format": "wav"}
|
|
if instruct.strip():
|
|
payload["instruct"] = instruct.strip()
|
|
_apply_tts_extra_params(payload, settings, "streaming")
|
|
return f"{tts_url}/v1/audio/speech", payload, tts_hdrs
|
|
|
|
|
|
def _open_tts_stream_response(text: str, voice: str, settings: dict, instruct: str = "") -> requests.Response:
|
|
endpoint, payload, tts_hdrs = _tts_stream_request_config(text, voice, settings, instruct)
|
|
resp = _post_tts_with_fallback(endpoint, payload, tts_hdrs, stream=True, timeout=(10, 900))
|
|
try:
|
|
resp.raise_for_status()
|
|
except Exception as exc:
|
|
detail = ""
|
|
try:
|
|
detail = resp.text[:500]
|
|
except Exception:
|
|
pass
|
|
resp.close()
|
|
raise RuntimeError(f"streaming backend error: {exc}{(': ' + detail) if detail else ''}") from exc
|
|
return resp
|
|
|
|
|
|
def _read_tts_stream_response(resp: requests.Response) -> tuple[bytes, str]:
|
|
try:
|
|
media_type = resp.headers.get("content-type", "audio/wav").split(";", 1)[0] or "audio/wav"
|
|
return b"".join(chunk for chunk in resp.iter_content(chunk_size=64 * 1024) if chunk), media_type
|
|
finally:
|
|
resp.close()
|
|
|
|
|
|
def _iter_tts_stream_response(resp: requests.Response):
|
|
try:
|
|
for chunk in resp.iter_content(chunk_size=64 * 1024):
|
|
if chunk:
|
|
yield chunk
|
|
finally:
|
|
resp.close()
|
|
|
|
|
|
# ── Voice design request helpers ──────────────────────────────────────────────
|
|
|
|
def _apply_voice_design_gender(instruct: str, gender: str) -> str:
|
|
gender = (gender or "").strip().upper()
|
|
if gender == "F":
|
|
prefix = (
|
|
"MANDATORY SPEAKER IDENTITY: Female speaker / woman.\n"
|
|
"gender: Female.\n"
|
|
"Use a clearly feminine vocal timbre, light-to-medium resonance, and soprano or mezzo-soprano pitch range.\n"
|
|
"Avoid male baritone, bass, chest-heavy, or masculine vocal qualities."
|
|
)
|
|
elif gender == "M":
|
|
prefix = (
|
|
"MANDATORY SPEAKER IDENTITY: Male speaker / man.\n"
|
|
"gender: Male.\n"
|
|
"Use a clearly masculine vocal timbre, medium-to-deep resonance, and tenor, baritone, or bass pitch range.\n"
|
|
"Avoid feminine soprano or mezzo-soprano vocal qualities."
|
|
)
|
|
else:
|
|
return instruct
|
|
return f"{prefix}\n\n{instruct.strip()}"
|
|
|
|
|
|
def _voice_design_request_audio(
|
|
instruct: str,
|
|
text: str,
|
|
language: str,
|
|
settings: dict,
|
|
gender: str = "",
|
|
) -> tuple[bytes, str]:
|
|
from core.validation import _validate_http_url
|
|
vd_url = _validate_http_url(settings.get("voice_design_url") or _VOICE_DESIGN_DEFAULT, allow_private=True).rstrip("/")
|
|
vd_key = (settings.get("voice_design_api_key") or settings.get("tts_api_key", "")).strip()
|
|
vd_hdrs = {"Authorization": f"Bearer {vd_key}"} if vd_key else {}
|
|
payload = {
|
|
"model": _VOICE_DESIGN_MODEL,
|
|
"input": text,
|
|
"instruct": _apply_voice_design_gender(instruct, gender),
|
|
"language": language,
|
|
"response_format": "wav",
|
|
}
|
|
_apply_tts_extra_params(payload, settings, "voice_design")
|
|
resp = _post_tts_with_fallback(f"{vd_url}/v1/audio/speech", payload, vd_hdrs, timeout=180)
|
|
resp.raise_for_status()
|
|
audio = resp.content
|
|
if not audio:
|
|
raise RuntimeError("backend returned empty audio")
|
|
media_type = resp.headers.get("content-type", "audio/wav").split(";", 1)[0] or "audio/wav"
|
|
return audio, media_type
|
|
|
|
|
|
def _voice_design_voice_request_audio(
|
|
voice: str,
|
|
text: str,
|
|
settings: dict,
|
|
instruct: str = "",
|
|
language: str = "Auto",
|
|
) -> tuple[bytes, str]:
|
|
from core.validation import _validate_http_url
|
|
vd_url = _validate_http_url(settings.get("voice_design_url") or _VOICE_DESIGN_DEFAULT, allow_private=True).rstrip("/")
|
|
vd_key = (settings.get("voice_design_api_key") or settings.get("tts_api_key", "")).strip()
|
|
vd_hdrs = {"Authorization": f"Bearer {vd_key}"} if vd_key else {}
|
|
payload = {
|
|
"model": _VOICE_DESIGN_MODEL,
|
|
"input": text,
|
|
"voice": voice,
|
|
"response_format": "wav",
|
|
}
|
|
if instruct.strip():
|
|
payload["instruct"] = instruct.strip()
|
|
if language and language != "Auto":
|
|
payload["language"] = language
|
|
_apply_tts_extra_params(payload, settings, "voice_design")
|
|
resp = _post_tts_with_fallback(f"{vd_url}/v1/audio/speech", payload, vd_hdrs, timeout=180)
|
|
resp.raise_for_status()
|
|
audio = resp.content
|
|
if not audio:
|
|
raise RuntimeError("backend returned empty audio")
|
|
media_type = resp.headers.get("content-type", "audio/wav").split(";", 1)[0] or "audio/wav"
|
|
return audio, media_type
|
|
|
|
|
|
# ── NVIDIA clone request ──────────────────────────────────────────────────────
|
|
|
|
def _nvidia_clone_request_audio(
|
|
text: str,
|
|
voice: str,
|
|
settings: dict,
|
|
mode: str = "zeroshot",
|
|
reference_transcript: str = "",
|
|
language: str = "",
|
|
) -> tuple[bytes, str]:
|
|
from core.validation import _validate_http_url
|
|
|
|
mode = "flow" if str(mode).lower().endswith("flow") else "zeroshot"
|
|
backend = "nvidia_flow" if mode == "flow" else "nvidia_zeroshot"
|
|
base_url = _validate_http_url(_preview_backend_base_url(settings, backend), allow_private=True).rstrip("/")
|
|
key = (settings.get("voice_design_api_key") or settings.get("tts_api_key", "")).strip()
|
|
headers = {"Authorization": f"Bearer {key}"} if key else {}
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
audio_path = _find_voice_audio(voice, scan_dir)
|
|
if audio_path is None:
|
|
raise RuntimeError(f"reference voice not found in library: {voice}")
|
|
prompt_wav = _to_wav_24k(audio_path)
|
|
has_ref, saved_transcript = _read_reference_text(audio_path)
|
|
prompt_transcript = str(reference_transcript or saved_transcript or "").strip()
|
|
if mode == "flow" and not prompt_transcript:
|
|
raise RuntimeError("NVIDIA Magpie Flow requires the selected voice to have an exact saved reference transcript")
|
|
data = {
|
|
"language": _nvidia_clone_language_code(voice, language),
|
|
"text": text,
|
|
}
|
|
if mode == "flow":
|
|
data["audio_prompt_transcript"] = prompt_transcript
|
|
params = _tts_extra_params(settings, backend)
|
|
for k, v in params.items():
|
|
if k in {"audio_prompt", "audio_prompt_transcript", "text", "language"}:
|
|
continue
|
|
data[k] = _form_value(v)
|
|
endpoint = f"{base_url}/v1/audio/synthesize"
|
|
with prompt_wav.open("rb") as f:
|
|
files = {"audio_prompt": ("prompt.wav", f, "audio/wav")}
|
|
resp = requests.post(endpoint, data=data, files=files, headers=headers, timeout=180)
|
|
if resp.status_code in {400, 404, 415, 422} and params:
|
|
try:
|
|
resp.close()
|
|
except Exception:
|
|
pass
|
|
fallback = {k: v for k, v in data.items() if k in {"language", "text", "audio_prompt_transcript"}}
|
|
with prompt_wav.open("rb") as f:
|
|
files = {"audio_prompt": ("prompt.wav", f, "audio/wav")}
|
|
resp = requests.post(endpoint, data=fallback, files=files, headers=headers, timeout=180)
|
|
resp.raise_for_status()
|
|
audio = resp.content
|
|
if not audio:
|
|
raise RuntimeError("NVIDIA clone backend returned empty audio")
|
|
media_type = resp.headers.get("content-type", "audio/wav").split(";", 1)[0] or "audio/wav"
|
|
return audio, media_type
|
|
|
|
|
|
# ── VibeVoice request ─────────────────────────────────────────────────────────
|
|
|
|
def _vibevoice_request_audio(text: str, settings: dict) -> tuple[bytes, str]:
|
|
url = _preview_backend_base_url(settings, "vibevoice").rstrip("/")
|
|
resp = requests.post(f"{url}/tts", json={"text": text}, timeout=60)
|
|
resp.raise_for_status()
|
|
audio = resp.content
|
|
if not audio:
|
|
raise RuntimeError("VibeVoice backend returned empty audio")
|
|
media_type = resp.headers.get("content-type", "audio/wav").split(";", 1)[0] or "audio/wav"
|
|
return audio, media_type
|
|
|
|
|
|
# ── Preview request dispatcher ────────────────────────────────────────────────
|
|
|
|
def _preview_request_audio(
|
|
text: str,
|
|
voice: str,
|
|
settings: dict,
|
|
instruct: str = "",
|
|
backend: str = "voice_clone",
|
|
) -> tuple[bytes, str]:
|
|
backend = _clean_preview_backend(backend)
|
|
if backend == "streaming":
|
|
resp = _open_tts_stream_response(text, voice, settings, instruct)
|
|
audio, media_type = _read_tts_stream_response(resp)
|
|
if not audio:
|
|
raise RuntimeError("backend returned empty audio")
|
|
return audio, media_type
|
|
if backend == "customvoice":
|
|
return _tts_request_audio(
|
|
text, voice, settings, instruct,
|
|
_preview_backend_base_url(settings, "customvoice"),
|
|
settings.get("tts_api_key", ""),
|
|
"openai",
|
|
"customvoice",
|
|
)
|
|
if backend == "voice_design":
|
|
return _voice_design_voice_request_audio(voice, text, settings, instruct)
|
|
if backend == "nvidia_magpie":
|
|
return _tts_request_audio(
|
|
text, voice, settings, instruct,
|
|
_preview_backend_base_url(settings, "nvidia_magpie"),
|
|
settings.get("tts_api_key", ""),
|
|
"nvidia_magpie",
|
|
"nvidia_magpie",
|
|
)
|
|
if backend == "nvidia_zeroshot":
|
|
return _nvidia_clone_request_audio(text, voice, settings, "zeroshot")
|
|
if backend == "nvidia_flow":
|
|
return _nvidia_clone_request_audio(text, voice, settings, "flow")
|
|
if backend == "kokoro":
|
|
return _tts_request_audio(
|
|
text, voice, settings, instruct,
|
|
url_override=_preview_backend_base_url(settings, "kokoro"),
|
|
api_key_override=settings.get("tts_api_key", ""),
|
|
backend_override="openai",
|
|
extra_backend="kokoro",
|
|
)
|
|
if backend == "vibevoice":
|
|
return _vibevoice_request_audio(text, settings)
|
|
if backend == "xtts":
|
|
return _tts_request_audio(
|
|
text, voice, settings, instruct,
|
|
url_override=_preview_backend_base_url(settings, "xtts"),
|
|
api_key_override=settings.get("tts_api_key", ""),
|
|
backend_override="openai",
|
|
extra_backend="xtts",
|
|
)
|
|
return _tts_request_audio(text, voice, settings, instruct)
|
|
|
|
|
|
# ── Response format helpers ───────────────────────────────────────────────────
|
|
|
|
def _requested_response_format(data: dict) -> str:
|
|
requested = str(data.get("response_format") or data.get("format") or "wav").strip().lower()
|
|
if requested in {"mp3", "mpeg"}:
|
|
return "mp3"
|
|
if requested in {"wav", "pcm"}:
|
|
return "wav"
|
|
return "wav"
|
|
|
|
|
|
def _audio_ext_media(response_format: str) -> tuple[str, str]:
|
|
if response_format == "mp3":
|
|
return "mp3", "audio/mpeg"
|
|
return "wav", "audio/wav"
|
|
|
|
|
|
def _prepare_proxy_audio(audio: bytes, media_type: str, response_format: str) -> tuple[bytes, str, str, float | None, bool]:
|
|
"""Rewrite backend audio so clients receive playable headers and requested format."""
|
|
try:
|
|
source_format = "wav" if audio[:4] == b"RIFF" or "wav" in media_type.lower() else None
|
|
segment = AudioSegment.from_file(io.BytesIO(audio), format=source_format)
|
|
except Exception:
|
|
ext, wanted_media = _audio_ext_media(response_format)
|
|
return audio, media_type or wanted_media, ext, None, False
|
|
|
|
duration = len(segment) / 1000.0
|
|
clipped = False
|
|
if _MAX_TTS_OUTPUT_SECONDS > 0 and duration > _MAX_TTS_OUTPUT_SECONDS:
|
|
segment = segment[:int(_MAX_TTS_OUTPUT_SECONDS * 1000)]
|
|
duration = len(segment) / 1000.0
|
|
clipped = True
|
|
ext, wanted_media = _audio_ext_media(response_format)
|
|
out = io.BytesIO()
|
|
export_format = "mp3" if response_format == "mp3" else "wav"
|
|
segment.export(out, format=export_format)
|
|
return out.getvalue(), wanted_media, ext, duration, clipped
|
|
|
|
|
|
# ── Route sound helpers ───────────────────────────────────────────────────────
|
|
|
|
def _route_sound_path(settings: dict, value: str) -> Path | None:
|
|
from core.validation import _safe_child_path
|
|
from core.voice import _AUDIO_EXTS
|
|
|
|
value = str(value or "").strip()
|
|
if not value:
|
|
return None
|
|
scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
|
|
candidate = Path(value)
|
|
if not candidate.is_absolute():
|
|
candidate = scan_dir / candidate
|
|
path = _safe_child_path(scan_dir, candidate)
|
|
if not path.exists() or not path.is_file() or path.suffix.lower() not in _AUDIO_EXTS:
|
|
raise RuntimeError(f"Route sound not found or unsupported: {value}")
|
|
return path
|
|
|
|
|
|
def _sound_segment(path: Path) -> AudioSegment:
|
|
return AudioSegment.from_file(str(path)).set_channels(1).set_sample_width(2).set_frame_rate(24000)
|
|
|
|
|
|
def _apply_route_sounds(audio: bytes, media_type: str, route: dict | None, settings: dict) -> tuple[bytes, str, list[str]]:
|
|
if not route:
|
|
return audio, media_type, []
|
|
before = _route_sound_path(settings, str(route.get("before_sound", "")))
|
|
after = _route_sound_path(settings, str(route.get("after_sound", "")))
|
|
if not before and not after:
|
|
return audio, media_type, []
|
|
|
|
source_format = "wav" if audio[:4] == b"RIFF" or "wav" in media_type.lower() else None
|
|
speech = AudioSegment.from_file(io.BytesIO(audio), format=source_format)
|
|
speech = speech.set_channels(1).set_sample_width(2).set_frame_rate(24000)
|
|
combined = AudioSegment.empty()
|
|
applied = []
|
|
if before:
|
|
combined += _sound_segment(before)
|
|
applied.append(f"before:{before.name}")
|
|
combined += speech
|
|
if after:
|
|
combined += _sound_segment(after)
|
|
applied.append(f"after:{after.name}")
|
|
|
|
out = io.BytesIO()
|
|
combined.export(out, format="wav")
|
|
return out.getvalue(), "audio/wav", applied
|
|
|
|
|
|
# ── Voice design dialogue helpers ─────────────────────────────────────────────
|
|
|
|
def _parse_voice_design_dialogue(instruct: str, script: str) -> tuple[dict[str, str], list[tuple[str, str]]] | None:
|
|
import re
|
|
speakers: dict[str, str] = {}
|
|
for raw in instruct.splitlines():
|
|
line = raw.strip()
|
|
if not line:
|
|
continue
|
|
match = re.match(r'^"?([^":]+)"?\s*:\s*"?(.+?)"?$', line)
|
|
if match:
|
|
speakers[match.group(1).strip()] = match.group(2).strip()
|
|
|
|
turns: list[tuple[str, str]] = []
|
|
for raw in script.splitlines():
|
|
line = raw.strip()
|
|
if not line:
|
|
continue
|
|
match = re.match(r"^([^:]{1,40}):\s*(.+)$", line)
|
|
if match:
|
|
speaker = match.group(1).strip()
|
|
text = match.group(2).strip()
|
|
if speaker in speakers and text:
|
|
turns.append((speaker, text))
|
|
|
|
if len(speakers) < 2 or len(turns) < 2:
|
|
return None
|
|
if len({speaker for speaker, _text in turns}) < 2:
|
|
return None
|
|
return speakers, turns
|
|
|
|
|
|
def _infer_voice_design_gender(description: str) -> str:
|
|
import re
|
|
text = f" {description.lower()} "
|
|
if re.search(r"\b(female|woman|girl|feminine|soprano|mezzo-soprano|mezzo)\b", text):
|
|
return "F"
|
|
if re.search(r"\b(male|man|boy|masculine|tenor|baritone|bass)\b", text):
|
|
return "M"
|
|
return ""
|
|
|
|
|
|
def _voice_design_dialogue_request_audio(
|
|
speakers: dict[str, str],
|
|
turns: list[tuple[str, str]],
|
|
language: str,
|
|
settings: dict,
|
|
) -> tuple[bytes, str]:
|
|
from core.audio import _audio_segment_from_wav, _wav_bytes_from_segment
|
|
combined = AudioSegment.silent(duration=120, frame_rate=24000).set_channels(1).set_sample_width(2)
|
|
pause = AudioSegment.silent(duration=180, frame_rate=24000).set_channels(1).set_sample_width(2)
|
|
|
|
for speaker, text in turns:
|
|
description = speakers[speaker]
|
|
turn_instruct = f'Speaker "{speaker}".\n{description}'
|
|
audio, _media_type = _voice_design_request_audio(
|
|
turn_instruct,
|
|
text,
|
|
language,
|
|
settings,
|
|
gender=_infer_voice_design_gender(description),
|
|
)
|
|
combined += _audio_segment_from_wav(audio) + pause
|
|
|
|
return _wav_bytes_from_segment(combined), "audio/wav"
|
|
|
|
|
|
# ── Benchmark request ─────────────────────────────────────────────────────────
|
|
|
|
def _tts_benchmark_request(text: str, voice: str, settings: dict, label: str) -> dict:
|
|
endpoint, payload, tts_hdrs = _tts_request_config(text, voice, settings, "wav")
|
|
start = time.perf_counter()
|
|
first_audio_at = None
|
|
raw = bytearray()
|
|
media_type = "audio/wav"
|
|
|
|
with _post_tts_with_fallback(endpoint, payload, tts_hdrs, stream=True, timeout=180) as resp:
|
|
resp.raise_for_status()
|
|
media_type = resp.headers.get("content-type", "audio/wav").split(";", 1)[0] or "audio/wav"
|
|
for chunk in resp.iter_content(chunk_size=512):
|
|
if not chunk:
|
|
continue
|
|
raw.extend(chunk)
|
|
if first_audio_at is None:
|
|
if payload.get("response_format") == "wav" or "wav" in media_type.lower():
|
|
offset = _wav_data_offset(bytes(raw))
|
|
if offset is not None and len(raw) > offset:
|
|
first_audio_at = time.perf_counter()
|
|
else:
|
|
first_audio_at = time.perf_counter()
|
|
|
|
total = time.perf_counter() - start
|
|
if not raw:
|
|
raise RuntimeError("backend returned empty audio")
|
|
audio_sec = _audio_duration_from_bytes(bytes(raw), media_type)
|
|
rtf = total / audio_sec if audio_sec and audio_sec > 0 else None
|
|
speed = audio_sec / total if audio_sec and total > 0 else None
|
|
return {
|
|
"ok": True,
|
|
"label": label,
|
|
"text": text,
|
|
"ttfa_ms": round(((first_audio_at or time.perf_counter()) - start) * 1000, 1),
|
|
"total_sec": round(total, 3),
|
|
"audio_sec": round(audio_sec, 3) if audio_sec is not None else None,
|
|
"rtf": round(rtf, 3) if rtf is not None else None,
|
|
"speed": round(speed, 3) if speed is not None else None,
|
|
"bytes": len(raw),
|
|
}
|