diff --git a/abogen/web/routes/api.py b/abogen/web/routes/api.py index 84973b0..57b1579 100644 --- a/abogen/web/routes/api.py +++ b/abogen/web/routes/api.py @@ -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 diff --git a/abogen/web/routes/utils/preview.py b/abogen/web/routes/utils/preview.py index b8eed4e..95bf6ba 100644 --- a/abogen/web/routes/utils/preview.py +++ b/abogen/web/routes/utils/preview.py @@ -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 diff --git a/abogen/web/static/speakers.js b/abogen/web/static/speakers.js index fed4117..23affcb 100644 --- a/abogen/web/static/speakers.js +++ b/abogen/web/static/speakers.js @@ -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); diff --git a/tests/test_preview_applies_manual_overrides.py b/tests/test_preview_applies_manual_overrides.py new file mode 100644 index 0000000..5044c51 --- /dev/null +++ b/tests/test_preview_applies_manual_overrides.py @@ -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"]