diff --git a/abogen/tts_supertonic.py b/abogen/tts_supertonic.py index 709965b..53b0f0e 100644 --- a/abogen/tts_supertonic.py +++ b/abogen/tts_supertonic.py @@ -1,6 +1,8 @@ from __future__ import annotations +import ast from dataclasses import dataclass +import logging import math import re from typing import Any, Iterable, Iterator, Optional @@ -8,6 +10,9 @@ from typing import Any, Iterable, Iterator, Optional import numpy as np +logger = logging.getLogger(__name__) + + DEFAULT_SUPERTONIC_VOICES = ("M1", "M2", "M3", "M4", "M5", "F1", "F2", "F3", "F4", "F5") @@ -76,6 +81,48 @@ def _split_text(text: str, *, split_pattern: Optional[str], max_chunk_length: in return result +_UNSUPPORTED_CHARS_RE = re.compile(r"unsupported character\(s\):\s*(\[[^\]]*\])", re.IGNORECASE) + + +def _parse_unsupported_characters(error: BaseException) -> list[str]: + """Best-effort extraction of unsupported characters from SuperTonic errors.""" + + message = " ".join(str(part) for part in getattr(error, "args", ()) if part is not None) or str(error) + match = _UNSUPPORTED_CHARS_RE.search(message) + if not match: + return [] + + raw = match.group(1) + try: + value = ast.literal_eval(raw) + except Exception: + return [] + + if isinstance(value, (list, tuple)): + out: list[str] = [] + for item in value: + if item is None: + continue + s = str(item) + if s: + out.append(s) + return out + + if isinstance(value, str) and value: + return [value] + + return [] + + +def _remove_unsupported_characters(text: str, unsupported: Iterable[str]) -> str: + result = text + for item in unsupported: + if not item: + continue + result = result.replace(item, "") + return result + + class SupertonicPipeline: """Minimal adapter that mimics Kokoro's pipeline iteration interface.""" @@ -118,15 +165,57 @@ class SupertonicPipeline: style = self._tts.get_voice_style(voice_name=voice_name) chunks = _split_text(text, split_pattern=split_pattern, max_chunk_length=self.max_chunk_length) for chunk in chunks: - wav, duration = self._tts.synthesize( - text=chunk, - voice_style=style, - total_steps=steps, - speed=speed_value, - max_chunk_length=self.max_chunk_length, - silence_duration=0.0, - verbose=False, - ) + chunk_to_speak = chunk + removed: set[str] = set() + last_exc: Exception | None = None + + # SuperTonic can raise ValueError for unsupported characters; strip and retry. + for attempt in range(3): + try: + wav, duration = self._tts.synthesize( + text=chunk_to_speak, + voice_style=style, + total_steps=steps, + speed=speed_value, + max_chunk_length=self.max_chunk_length, + silence_duration=0.0, + verbose=False, + ) + break + except ValueError as exc: + last_exc = exc + unsupported = _parse_unsupported_characters(exc) + if not unsupported: + raise + + removed.update(unsupported) + sanitized = _remove_unsupported_characters(chunk_to_speak, unsupported).strip() + + # If we didn't change anything, don't loop forever. + if sanitized == chunk_to_speak.strip(): + raise + + chunk_to_speak = sanitized + if not chunk_to_speak: + logger.warning( + "SuperTonic: dropped a chunk after removing unsupported characters: %s", + sorted(removed), + ) + break + + if attempt == 0: + logger.warning( + "SuperTonic: removed unsupported characters %s and retried.", + sorted(removed), + ) + else: + # Exhausted retries. + assert last_exc is not None + raise last_exc + + if not chunk_to_speak: + continue + audio = _ensure_float32_mono(wav) # If duration is present, infer the source sample rate and resample if needed. @@ -143,4 +232,4 @@ class SupertonicPipeline: if src_rate != self.sample_rate: audio = _resample_linear(audio, src_rate, self.sample_rate) - yield SupertonicSegment(graphemes=chunk, audio=audio) + yield SupertonicSegment(graphemes=chunk_to_speak, audio=audio) diff --git a/abogen/web/conversion_runner.py b/abogen/web/conversion_runner.py index 2d72e4e..1827808 100644 --- a/abogen/web/conversion_runner.py +++ b/abogen/web/conversion_runner.py @@ -30,6 +30,7 @@ from abogen.normalization_settings import ( apply_overrides as apply_normalization_overrides, ) from abogen.entity_analysis import normalize_token as normalize_entity_token +from abogen.entity_analysis import normalize_manual_override_token from abogen.text_extractor import ExtractedChapter, extract_from_path from abogen.utils import ( calculate_text_length, @@ -907,6 +908,97 @@ def _normalize_for_pipeline( return normalize_for_pipeline(text, config=apostrophe_config, settings=runtime_settings) +def _merge_pronunciation_overrides(job: Any) -> List[Dict[str, Any]]: + """Return pronunciation override entries, ensuring manual overrides are included. + + Pending jobs keep both `manual_overrides` and `pronunciation_overrides`, but the + latter can be stale if the UI didn't resync before enqueue. During conversion, + we must merge manual overrides so they always apply (before TTS). + + Precedence: manual overrides win over existing entries for the same normalized key. + """ + + collected: Dict[str, Dict[str, Any]] = {} + + existing = getattr(job, "pronunciation_overrides", None) + if isinstance(existing, list): + for entry in existing: + if not isinstance(entry, Mapping): + continue + token_value = str(entry.get("token") or "").strip() + pronunciation_value = str(entry.get("pronunciation") or "").strip() + if not token_value or not pronunciation_value: + continue + normalized = str(entry.get("normalized") or "").strip() or normalize_entity_token(token_value) + if not normalized: + continue + collected[normalized] = { + "token": token_value, + "normalized": normalized, + "pronunciation": pronunciation_value, + "voice": str(entry.get("voice") or "").strip() or None, + "notes": str(entry.get("notes") or "").strip() or None, + "context": str(entry.get("context") or "").strip() or None, + "source": str(entry.get("source") or "pronunciation"), + "language": getattr(job, "language", None), + } + + # Speaker pronunciation entries (optional), mirrored from the pending-job collector. + speakers = getattr(job, "speakers", None) + if isinstance(speakers, dict): + for payload in speakers.values(): + if not isinstance(payload, Mapping): + continue + token_value = str(payload.get("token") or "").strip() + pronunciation_value = str(payload.get("pronunciation") or "").strip() + if not token_value or not pronunciation_value: + continue + normalized = normalize_entity_token(token_value) + if not normalized: + continue + collected[normalized] = { + "token": token_value, + "normalized": normalized, + "pronunciation": pronunciation_value, + "voice": str( + payload.get("resolved_voice") + or payload.get("voice") + or getattr(job, "voice", "") + ).strip() + or None, + "notes": None, + "context": None, + "source": "speaker", + "language": getattr(job, "language", None), + } + + # Manual overrides should take precedence. + manual = getattr(job, "manual_overrides", None) + if isinstance(manual, list): + for entry in manual: + if not isinstance(entry, Mapping): + continue + token_value = str(entry.get("token") or "").strip() + pronunciation_value = str(entry.get("pronunciation") or "").strip() + if not token_value or not pronunciation_value: + continue + normalized = str(entry.get("normalized") or "").strip() or normalize_manual_override_token(token_value) + if not normalized: + continue + collected[normalized] = { + "token": token_value, + "normalized": normalized, + "pronunciation": pronunciation_value, + "voice": str(entry.get("voice") or "").strip() or None, + "notes": str(entry.get("notes") or "").strip() or None, + "context": str(entry.get("context") or "").strip() or None, + "source": str(entry.get("source") or "manual"), + "language": getattr(job, "language", None), + } + + return list(collected.values()) + + def _compile_pronunciation_rules( overrides: Optional[Iterable[Mapping[str, Any]]], ) -> List[Dict[str, Any]]: @@ -1535,7 +1627,8 @@ def run_conversion_job(job: Job) -> None: extraction = extract_from_path(job.stored_path) file_type = _infer_file_type(job.stored_path) - pronunciation_rules = _compile_pronunciation_rules(job.pronunciation_overrides) + pronunciation_overrides = _merge_pronunciation_overrides(job) + pronunciation_rules = _compile_pronunciation_rules(pronunciation_overrides) heteronym_sentence_rules = _compile_heteronym_sentence_rules( getattr(job, "heteronym_overrides", None) ) @@ -1550,7 +1643,7 @@ def run_conversion_job(job: Job) -> None: f"Applying {count} pronunciation override{'s' if count != 1 else ''} during conversion.", level="debug", ) - for override_entry in job.pronunciation_overrides or []: + for override_entry in pronunciation_overrides or []: if not isinstance(override_entry, Mapping): continue raw_token = str(override_entry.get("token") or "").strip() diff --git a/abogen/web/routes/jobs.py b/abogen/web/routes/jobs.py index 6d91c1d..b546f5e 100644 --- a/abogen/web/routes/jobs.py +++ b/abogen/web/routes/jobs.py @@ -252,6 +252,15 @@ def job_logs(job_id: str) -> str: abort(404) return render_template("job_logs_static.html", job=job) + +@jobs_bp.get("//logs/partial") +def job_logs_partial(job_id: str) -> ResponseReturnValue: + job = get_service().get_job(job_id) + if not job: + # Return a non-polling section so HTMX stops retrying. + return render_template("partials/logs_section_missing.html"), 200 + return render_template("partials/logs_section.html", job=job) + @jobs_bp.get("//logs/stream") def stream_logs(job_id: str) -> ResponseReturnValue: job = get_service().get_job(job_id) diff --git a/abogen/web/templates/job_detail.html b/abogen/web/templates/job_detail.html index a7bfefc..6c45261 100644 --- a/abogen/web/templates/job_detail.html +++ b/abogen/web/templates/job_detail.html @@ -149,9 +149,7 @@ {% endif %} -
- {% include "partials/logs.html" %} -
+{% include "partials/logs_section.html" %} {% endblock %} {% block scripts %} diff --git a/abogen/web/templates/partials/logs_section.html b/abogen/web/templates/partials/logs_section.html new file mode 100644 index 0000000..fe68e90 --- /dev/null +++ b/abogen/web/templates/partials/logs_section.html @@ -0,0 +1,7 @@ +
+ {% include "partials/logs.html" %} +
diff --git a/abogen/web/templates/partials/logs_section_missing.html b/abogen/web/templates/partials/logs_section_missing.html new file mode 100644 index 0000000..612014c --- /dev/null +++ b/abogen/web/templates/partials/logs_section_missing.html @@ -0,0 +1,6 @@ +
+
+
Live log
+
+

Job not found (it may have completed, been removed, or the server restarted). Refresh the page to load an active job.

+
diff --git a/tests/test_manual_overrides_applied_first.py b/tests/test_manual_overrides_applied_first.py new file mode 100644 index 0000000..efae6f4 --- /dev/null +++ b/tests/test_manual_overrides_applied_first.py @@ -0,0 +1,51 @@ +from abogen.web import conversion_runner + + +class DummyJob: + def __init__(self): + self.language = "en" + self.voice = "M1" + self.speakers = None + self.manual_overrides = [] + self.pronunciation_overrides = [] + + +def _apply(text: str, job: DummyJob) -> str: + merged = conversion_runner._merge_pronunciation_overrides(job) + rules = conversion_runner._compile_pronunciation_rules(merged) + return conversion_runner._apply_pronunciation_rules(text, rules) + + +def test_manual_override_is_applied_even_if_pronunciation_overrides_stale(): + job = DummyJob() + job.manual_overrides = [ + { + "token": "Unfu*k", + "pronunciation": "Unfuck", + } + ] + + out = _apply("He said Unfu*k loudly.", job) + assert "Unfuck" in out + assert "Unfu*k" not in out + + +def test_manual_override_takes_precedence_over_existing_pronunciation_override(): + job = DummyJob() + job.pronunciation_overrides = [ + { + "token": "Unfu*k", + "normalized": "unfu*k", + "pronunciation": "WRONG", + } + ] + job.manual_overrides = [ + { + "token": "Unfu*k", + "pronunciation": "RIGHT", + } + ] + + out = _apply("Unfu*k.", job) + assert "RIGHT" in out + assert "WRONG" not in out diff --git a/tests/test_tts_supertonic_unsupported_chars.py b/tests/test_tts_supertonic_unsupported_chars.py new file mode 100644 index 0000000..c08ca2c --- /dev/null +++ b/tests/test_tts_supertonic_unsupported_chars.py @@ -0,0 +1,53 @@ +import numpy as np + +from abogen.tts_supertonic import SupertonicPipeline + + +class _DummyTTS: + def get_voice_style(self, voice_name: str): + return {"voice": voice_name} + + def synthesize( + self, + *, + text: str, + voice_style, + total_steps: int, + speed: float, + max_chunk_length: int, + silence_duration: float, + verbose: bool, + ): + if "•" in text: + raise ValueError("Found 1 unsupported character(s): ['•']") + # Return 50ms of audio at 24kHz. + sr = 24000 + audio = np.zeros(int(0.05 * sr), dtype="float32") + return audio, 0.05 + + +def test_supertonic_pipeline_strips_unsupported_characters_and_retries(): + # Avoid importing/initializing real supertonic by manually constructing the pipeline. + pipeline = SupertonicPipeline.__new__(SupertonicPipeline) + pipeline.sample_rate = 24000 + pipeline.total_steps = 5 + pipeline.max_chunk_length = 1000 + pipeline._tts = _DummyTTS() + + segs = list(pipeline("Hello • world", voice="M1", speed=1.0)) + assert len(segs) == 1 + assert segs[0].graphemes == "Hello world" or segs[0].graphemes == "Hello world" + assert isinstance(segs[0].audio, np.ndarray) + assert segs[0].audio.dtype == np.float32 + assert segs[0].audio.size > 0 + + +def test_supertonic_pipeline_drops_chunk_if_only_unsupported_characters(): + pipeline = SupertonicPipeline.__new__(SupertonicPipeline) + pipeline.sample_rate = 24000 + pipeline.total_steps = 5 + pipeline.max_chunk_length = 1000 + pipeline._tts = _DummyTTS() + + segs = list(pipeline("•", voice="M1", speed=1.0)) + assert segs == []