From 2228f37c06446687d33d7f76bbfe8730df1eaa56 Mon Sep 17 00:00:00 2001 From: Artem Akymenko Date: Thu, 16 Jul 2026 08:53:15 +0000 Subject: [PATCH] =?UTF-8?q?refactor(domain):=20add=20prepare=5Ftext=5Ffor?= =?UTF-8?q?=5Ftts=20=E2=80=94=20unified=20normalization=20pipeline?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- abogen/domain/normalization.py | 70 +++++++++- abogen/webui/conversion_runner.py | 18 +-- tests/test_domain_normalization_pipeline.py | 147 ++++++++++++++++++++ 3 files changed, 221 insertions(+), 14 deletions(-) create mode 100644 tests/test_domain_normalization_pipeline.py diff --git a/abogen/domain/normalization.py b/abogen/domain/normalization.py index 4bd34fd..053368a 100644 --- a/abogen/domain/normalization.py +++ b/abogen/domain/normalization.py @@ -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) diff --git a/abogen/webui/conversion_runner.py b/abogen/webui/conversion_runner.py index 5dcf912..ad49dfb 100644 --- a/abogen/webui/conversion_runner.py +++ b/abogen/webui/conversion_runner.py @@ -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") diff --git a/tests/test_domain_normalization_pipeline.py b/tests/test_domain_normalization_pipeline.py new file mode 100644 index 0000000..86a0b3d --- /dev/null +++ b/tests/test_domain_normalization_pipeline.py @@ -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)