mirror of
https://github.com/denizsafak/abogen.git
synced 2026-07-18 05:40:26 +02:00
feat: Add pending ID handling and pronunciation/manual overrides to audio preview generation
This commit is contained in:
@@ -215,6 +215,7 @@ def api_voice_profiles_preview() -> ResponseReturnValue:
|
||||
@api_bp.post("/speaker-preview")
|
||||
def api_speaker_preview() -> ResponseReturnValue:
|
||||
payload = request.get_json(force=True, silent=True) or {}
|
||||
pending_id = str(payload.get("pending_id") or "").strip()
|
||||
text = payload.get("text", "Hello world")
|
||||
voice = payload.get("voice", "af_heart")
|
||||
language = payload.get("language", "a")
|
||||
@@ -243,6 +244,19 @@ def api_speaker_preview() -> ResponseReturnValue:
|
||||
|
||||
if not resolved_provider:
|
||||
resolved_provider = "supertonic" if str(base_spec or "").strip() in {"M1","M2","M3","M4","M5","F1","F2","F3","F4","F5"} else "kokoro"
|
||||
|
||||
pronunciation_overrides = None
|
||||
manual_overrides = None
|
||||
speakers = None
|
||||
if pending_id:
|
||||
try:
|
||||
pending = get_service().get_pending_job(pending_id)
|
||||
except Exception:
|
||||
pending = None
|
||||
if pending is not None:
|
||||
manual_overrides = getattr(pending, "manual_overrides", None)
|
||||
pronunciation_overrides = getattr(pending, "pronunciation_overrides", None)
|
||||
speakers = getattr(pending, "speakers", None)
|
||||
|
||||
try:
|
||||
return synthesize_preview(
|
||||
@@ -254,6 +268,9 @@ def api_speaker_preview() -> ResponseReturnValue:
|
||||
,
|
||||
tts_provider=resolved_provider,
|
||||
supertonic_total_steps=supertonic_total_steps or int(settings.get("supertonic_total_steps") or 5),
|
||||
pronunciation_overrides=pronunciation_overrides,
|
||||
manual_overrides=manual_overrides,
|
||||
speakers=speakers,
|
||||
)
|
||||
except Exception as e:
|
||||
return jsonify({"error": str(e)}), 500
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import io
|
||||
import threading
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple
|
||||
import numpy as np
|
||||
import soundfile as sf
|
||||
from flask import current_app, send_file
|
||||
@@ -61,21 +61,47 @@ def generate_preview_audio(
|
||||
tts_provider: str = "kokoro",
|
||||
supertonic_total_steps: int = 5,
|
||||
max_seconds: float = 8.0,
|
||||
pronunciation_overrides: Optional[Iterable[Mapping[str, Any]]] = None,
|
||||
manual_overrides: Optional[Iterable[Mapping[str, Any]]] = None,
|
||||
speakers: Optional[Mapping[str, Any]] = None,
|
||||
) -> bytes:
|
||||
if not text.strip():
|
||||
raise ValueError("Preview text is required")
|
||||
|
||||
provider = (tts_provider or "kokoro").strip().lower()
|
||||
|
||||
normalized_text = text
|
||||
# Apply pronunciation/manual overrides first so tokens like `Unfu*k` still match
|
||||
# before any downstream normalization potentially strips punctuation.
|
||||
source_text = text
|
||||
if pronunciation_overrides or manual_overrides or speakers:
|
||||
try:
|
||||
from abogen.web import conversion_runner as runner
|
||||
|
||||
class _PreviewJob:
|
||||
def __init__(self):
|
||||
self.language = language
|
||||
self.voice = voice_spec
|
||||
self.speakers = speakers
|
||||
self.manual_overrides = list(manual_overrides or [])
|
||||
self.pronunciation_overrides = list(pronunciation_overrides or [])
|
||||
|
||||
job = _PreviewJob()
|
||||
merged = runner._merge_pronunciation_overrides(job)
|
||||
rules = runner._compile_pronunciation_rules(merged)
|
||||
source_text = runner._apply_pronunciation_rules(source_text, rules)
|
||||
except Exception:
|
||||
current_app.logger.exception("Preview override application failed; using raw text")
|
||||
source_text = text
|
||||
|
||||
normalized_text = source_text
|
||||
if provider != "supertonic":
|
||||
try:
|
||||
from abogen.kokoro_text_normalization import normalize_for_pipeline
|
||||
|
||||
normalized_text = normalize_for_pipeline(text)
|
||||
normalized_text = normalize_for_pipeline(source_text)
|
||||
except Exception:
|
||||
current_app.logger.exception("Preview normalization failed; using raw text")
|
||||
normalized_text = text
|
||||
normalized_text = source_text
|
||||
|
||||
if provider == "supertonic":
|
||||
from abogen.tts_supertonic import SupertonicPipeline
|
||||
@@ -152,6 +178,9 @@ def synthesize_preview(
|
||||
tts_provider: str = "kokoro",
|
||||
supertonic_total_steps: int = 5,
|
||||
max_seconds: float = 8.0,
|
||||
pronunciation_overrides: Optional[Iterable[Mapping[str, Any]]] = None,
|
||||
manual_overrides: Optional[Iterable[Mapping[str, Any]]] = None,
|
||||
speakers: Optional[Mapping[str, Any]] = None,
|
||||
) -> ResponseReturnValue:
|
||||
try:
|
||||
audio_bytes = generate_preview_audio(
|
||||
@@ -163,6 +192,9 @@ def synthesize_preview(
|
||||
tts_provider=tts_provider,
|
||||
supertonic_total_steps=supertonic_total_steps,
|
||||
max_seconds=max_seconds,
|
||||
pronunciation_overrides=pronunciation_overrides,
|
||||
manual_overrides=manual_overrides,
|
||||
speakers=speakers,
|
||||
)
|
||||
except Exception as e:
|
||||
raise e
|
||||
|
||||
@@ -77,6 +77,15 @@ const playPreview = async (button) => {
|
||||
max_seconds: 8,
|
||||
};
|
||||
|
||||
const pendingId =
|
||||
button.dataset.pendingId ||
|
||||
button.closest("[data-pending-id]")?.dataset.pendingId ||
|
||||
document.querySelector('[data-role="prepare-form"]')?.dataset.pendingId ||
|
||||
"";
|
||||
if (pendingId) {
|
||||
payload.pending_id = pendingId;
|
||||
}
|
||||
|
||||
stopCurrentPlayback();
|
||||
activeButton = button;
|
||||
setLoadingState(button, true);
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
from abogen.web.routes.utils import preview
|
||||
|
||||
|
||||
def test_preview_applies_manual_override_before_normalization(monkeypatch):
|
||||
# Don't run real TTS/normalization; just exercise the override stage by
|
||||
# forcing provider=kokoro and then stubbing normalize_for_pipeline.
|
||||
|
||||
monkeypatch.setattr(preview, "get_preview_pipeline", lambda language, device: None)
|
||||
|
||||
# Stub normalize_for_pipeline to be identity; we only care that overrides run.
|
||||
class _Norm:
|
||||
@staticmethod
|
||||
def normalize_for_pipeline(text):
|
||||
return text
|
||||
|
||||
monkeypatch.setitem(__import__("sys").modules, "abogen.kokoro_text_normalization", _Norm)
|
||||
|
||||
# And stub the kokoro pipeline path so generate_preview_audio won't proceed.
|
||||
# We'll instead validate by calling the override logic through generate_preview_audio
|
||||
# with provider=supertonic and stub SupertonicPipeline to capture input.
|
||||
captured = {}
|
||||
|
||||
class DummyPipeline:
|
||||
def __init__(self, **kwargs):
|
||||
pass
|
||||
|
||||
def __call__(self, text, **kwargs):
|
||||
captured["text"] = text
|
||||
return iter(())
|
||||
|
||||
monkeypatch.setitem(__import__("sys").modules, "abogen.tts_supertonic", type("M", (), {"SupertonicPipeline": DummyPipeline}))
|
||||
|
||||
try:
|
||||
preview.generate_preview_audio(
|
||||
text="He said Unfu*k loudly.",
|
||||
voice_spec="M1",
|
||||
language="en",
|
||||
speed=1.0,
|
||||
use_gpu=False,
|
||||
tts_provider="supertonic",
|
||||
manual_overrides=[{"token": "Unfu*k", "pronunciation": "Unfuck"}],
|
||||
)
|
||||
except Exception:
|
||||
# generate_preview_audio will raise because no audio chunks; that's fine.
|
||||
pass
|
||||
|
||||
assert "text" in captured
|
||||
assert "Unfuck" in captured["text"]
|
||||
assert "Unfu*k" not in captured["text"]
|
||||
Reference in New Issue
Block a user