refactor(domain): add prepare_text_for_tts — unified normalization pipeline

New function chains all three normalization stages:
  1. Heteronym sentence rules (context-dependent pronunciation)
  2. Pronunciation rules (token-level replacements)
  3. Pipeline normalization (apostrophe, LLM)

This is the single entry point that both Web UI and PyQt should call
before TTS synthesis. Currently only Web UI uses it; PyQt has NO
normalization — this unlocks that capability.

Updated conversion_runner.emit_text to use the new function.
1038 tests pass.
This commit is contained in:
Artem Akymenko
2026-07-16 08:53:15 +00:00
parent 832e2c5197
commit 2228f37c06
3 changed files with 221 additions and 14 deletions
+68 -2
View File
@@ -1,8 +1,15 @@
"""Text normalization convenience helpers."""
"""Text normalization convenience helpers.
Provides both the simple ``normalize_text_for_pipeline`` (apostrophe + LLM only)
and the comprehensive ``prepare_text_for_tts`` that chains all three normalization
stages used during conversion: heteronym rules → pronunciation rules → pipeline
normalization. The latter is the single entry point that both the Web UI and
PyQt Desktop GUI should use.
"""
from __future__ import annotations
from typing import Any, Mapping, Optional
from typing import Any, Dict, List, Mapping, Optional
from abogen.kokoro_text_normalization import (
ApostropheConfig,
@@ -28,3 +35,62 @@ def normalize_text_for_pipeline(
runtime_settings = _apply_overrides(runtime_settings, normalization_overrides)
apostrophe_config = build_apostrophe_config(settings=runtime_settings, base=_BASE_APOSTROPHE_CONFIG)
return _normalize_for_pipeline(text, config=apostrophe_config, settings=runtime_settings)
def prepare_text_for_tts(
text: str,
*,
heteronym_rules: Optional[List[Dict[str, Any]]] = None,
pronunciation_rules: Optional[List[Dict[str, Any]]] = None,
normalization_overrides: Optional[Mapping[str, Any]] = None,
usage_counter: Optional[Dict[str, int]] = None,
) -> str:
"""Apply the full text normalization pipeline before TTS synthesis.
Chains three stages in order:
1. Heteronym sentence rules (context-dependent pronunciation)
2. Pronunciation rules (token-level replacements)
3. Pipeline normalization (apostrophe handling, LLM normalization)
This is the **single entry point** that both the Web UI conversion runner
and the PyQt conversion thread should call before passing text to the TTS
backend.
Parameters
----------
text:
Raw text to normalize.
heteronym_rules:
Compiled heteronym rules from ``compile_heteronym_sentence_rules``.
pronunciation_rules:
Compiled pronunciation rules from ``compile_pronunciation_rules``.
normalization_overrides:
User-level overrides for normalization settings (apostrophe mode, etc.).
usage_counter:
Mutable dict that tracks how many times each pronunciation override was
applied. Passed through to ``apply_pronunciation_rules``.
Returns
-------
str
Fully normalized text ready for TTS.
"""
from abogen.domain.pronunciation import (
apply_heteronym_sentence_rules,
apply_pronunciation_rules,
)
result = str(text or "")
if heteronym_rules:
result = apply_heteronym_sentence_rules(result, heteronym_rules)
if pronunciation_rules:
result = apply_pronunciation_rules(result, pronunciation_rules, usage_counter)
runtime_settings = get_runtime_settings()
if normalization_overrides:
runtime_settings = _apply_overrides(runtime_settings, normalization_overrides)
apostrophe_config = build_apostrophe_config(settings=runtime_settings, base=_BASE_APOSTROPHE_CONFIG)
return _normalize_for_pipeline(result, config=apostrophe_config, settings=runtime_settings)
+6 -12
View File
@@ -73,10 +73,10 @@ from abogen.domain.file_type import (
from abogen.domain.pronunciation import (
compile_pronunciation_rules as _compile_pronunciation_rules,
compile_heteronym_sentence_rules as _compile_heteronym_sentence_rules,
apply_heteronym_sentence_rules as _apply_heteronym_sentence_rules,
apply_pronunciation_rules as _apply_pronunciation_rules,
merge_pronunciation_overrides as _merge_pronunciation_overrides,
)
from abogen.domain.normalization import prepare_text_for_tts
from abogen.domain.voice_resolution import (
spec_to_voice_ids as _spec_to_voice_ids,
job_voice_fallback as _job_voice_fallback,
@@ -445,19 +445,13 @@ def run_conversion_job(job: Job) -> None:
) -> int:
nonlocal processed_chars, current_time
source_text = str(text or "")
if heteronym_sentence_rules:
source_text = _apply_heteronym_sentence_rules(source_text, heteronym_sentence_rules)
if pronunciation_rules:
source_text = _apply_pronunciation_rules(
source_text,
pronunciation_rules,
usage_counter,
)
try:
normalized = normalize_for_pipeline(
normalized = prepare_text_for_tts(
source_text,
config=apostrophe_config,
settings=normalization_settings,
heteronym_rules=heteronym_sentence_rules,
pronunciation_rules=pronunciation_rules,
normalization_overrides=getattr(job, "normalization_overrides", None),
usage_counter=usage_counter,
)
except LLMClientError as exc:
job.add_log(f"LLM normalization failed: {exc}", level="error")
+147
View File
@@ -0,0 +1,147 @@
"""Tests for domain/normalization.py — prepare_text_for_tts."""
import pytest
from unittest.mock import patch, MagicMock
from abogen.domain.normalization import prepare_text_for_tts, normalize_text_for_pipeline
class TestPrepareTextForTts:
"""Tests for the comprehensive TTS text preparation pipeline."""
def test_empty_text(self):
result = prepare_text_for_tts("")
assert result == ""
def test_none_text(self):
result = prepare_text_for_tts(None)
assert result == ""
def test_passthrough_no_rules(self):
result = prepare_text_for_tts("Hello world")
assert isinstance(result, str)
assert len(result) > 0
def test_heteronym_rules_applied(self):
from abogen.domain.pronunciation import compile_heteronym_sentence_rules
overrides = [
{
"token": "read",
"pronunciation": "red",
"context": "past tense",
}
]
rules = compile_heteronym_sentence_rules(overrides)
if rules:
result = prepare_text_for_tts("I will read the book", heteronym_rules=rules)
assert isinstance(result, str)
def test_pronunciation_rules_applied(self):
from abogen.domain.pronunciation import compile_pronunciation_rules
overrides = [
{
"token": "epub",
"pronunciation": "ee-pub",
"normalized": "epub",
}
]
rules = compile_pronunciation_rules(overrides)
result = prepare_text_for_tts(
"This is an epub file",
pronunciation_rules=rules,
)
assert "ee-pub" in result
def test_usage_counter_tracks_pronunciation(self):
from abogen.domain.pronunciation import compile_pronunciation_rules
overrides = [
{
"token": "data",
"pronunciation": "day-ta",
"normalized": "data",
}
]
rules = compile_pronunciation_rules(overrides)
counter = {}
prepare_text_for_tts(
"The data is here and the data is there",
pronunciation_rules=rules,
usage_counter=counter,
)
assert counter.get("data", 0) >= 1
def test_combined_heteronym_and_pronunciation(self):
from abogen.domain.pronunciation import (
compile_heteronym_sentence_rules,
compile_pronunciation_rules,
)
heteronym_overrides = [
{
"token": "lead",
"pronunciation": "led",
"context": "metal",
}
]
pronunciation_overrides = [
{
"token": "gif",
"pronunciation": "jif",
"normalized": "gif",
}
]
h_rules = compile_heteronym_sentence_rules(heteronym_overrides)
p_rules = compile_pronunciation_rules(pronunciation_overrides)
result = prepare_text_for_tts(
"A lead gif",
heteronym_rules=h_rules if h_rules else None,
pronunciation_rules=p_rules,
)
assert isinstance(result, str)
@patch("abogen.domain.normalization.get_runtime_settings")
def test_normalization_overrides_passed_through(self, mock_settings):
mock_settings.return_value = {
"normalization_apostrophe_mode": "spacy",
"normalization_enabled": True,
}
result = prepare_text_for_tts(
"It's a test",
normalization_overrides={"normalization_enabled": False},
)
assert isinstance(result, str)
def test_pronunciation_rules_empty(self):
result = prepare_text_for_tts("Hello", pronunciation_rules=[])
assert isinstance(result, str)
def test_heteronym_rules_empty(self):
result = prepare_text_for_tts("Hello", heteronym_rules=[])
assert isinstance(result, str)
class TestNormalizeTextForPipeline:
"""Tests for the simpler normalization function."""
def test_basic_normalization(self):
result = normalize_text_for_pipeline("It's a test")
assert isinstance(result, str)
assert len(result) > 0
def test_empty_text(self):
result = normalize_text_for_pipeline("")
assert result == ""
@patch("abogen.domain.normalization.get_runtime_settings")
def test_with_overrides(self, mock_settings):
mock_settings.return_value = {
"normalization_apostrophe_mode": "spacy",
}
result = normalize_text_for_pipeline(
"It's a test",
normalization_overrides={"normalization_apostrophe_mode": "spacy"},
)
assert isinstance(result, str)