feat: Implement unsupported character handling in Supertonic pipeline and add related tests

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(
+            chunk_to_speak = chunk
-                text=chunk,
+            removed: set[str] = set()
-                voice_style=style,
+            last_exc: Exception | None = None
-                total_steps=steps,
+
-                speed=speed_value,
+            # SuperTonic can raise ValueError for unsupported characters; strip and retry.
-                max_chunk_length=self.max_chunk_length,
+            for attempt in range(3):
-                silence_duration=0.0,
+                try:
-                verbose=False,
+                    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)

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()

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("/<job_id>/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("/<job_id>/logs/stream")
 def stream_logs(job_id: str) -> ResponseReturnValue:
     job = get_service().get_job(job_id)

abogen/web/templates/job_detail.html

@@ -149,9 +149,7 @@
 </section>
 {% endif %}
 
-<section class="card" id="logs" hx-get="{{ url_for('jobs.job_logs', job_id=job.id) }}" hx-trigger="load, every 2s" hx-target="#logs" hx-swap="innerHTML">
+{% include "partials/logs_section.html" %}
-  {% include "partials/logs.html" %}
-</section>
 {% endblock %}
 
 {% block scripts %}

abogen/web/templates/partials/logs_section.html

@@ -0,0 +1,7 @@
+<section class="card" id="logs"
+         hx-get="{{ url_for('jobs.job_logs_partial', job_id=job.id) }}"
+         hx-trigger="load, every 2s"
+         hx-target="#logs"
+         hx-swap="outerHTML">
+  {% include "partials/logs.html" %}
+</section>

abogen/web/templates/partials/logs_section_missing.html

@@ -0,0 +1,6 @@
+<section class="card" id="logs">
+  <div class="card__title-row">
+    <div class="card__title">Live log</div>
+  </div>
+  <p>Job not found (it may have completed, been removed, or the server restarted). Refresh the page to load an active job.</p>
+</section>

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

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 == []