diff --git a/abogen/domain/chunk_utils.py b/abogen/domain/chunk_utils.py new file mode 100644 index 0000000..815fe55 --- /dev/null +++ b/abogen/domain/chunk_utils.py @@ -0,0 +1,75 @@ +"""Chunk processing utilities. + +Functions for grouping chunks, recording override usage, and selecting +text for TTS synthesis. +""" + +from __future__ import annotations + +from collections import defaultdict +from typing import Any, Dict, Iterable, Mapping, Optional + +from abogen.pronunciation_store import increment_usage + + +def safe_int(value: Any, default: int = 0) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default + + +def group_chunks_by_chapter(chunks: Iterable[Dict[str, Any]]) -> Dict[int, List[Dict[str, Any]]]: + grouped: Dict[int, List[Dict[str, Any]]] = defaultdict(list) + for entry in chunks or []: + if not isinstance(entry, dict): + continue + try: + chapter_index = int(entry.get("chapter_index", 0)) + except (TypeError, ValueError): + chapter_index = 0 + grouped[chapter_index].append(dict(entry)) + + for chapter_index, items in grouped.items(): + items.sort(key=lambda payload: safe_int(payload.get("chunk_index"))) + + return grouped + + +def record_override_usage( + job: Any, + usage_counter: Mapping[str, int], + token_map: Mapping[str, str], +) -> None: + if not usage_counter: + return + + language = getattr(job, "language", "") or "a" + for normalized, amount in usage_counter.items(): + if amount <= 0: + continue + token_value = token_map.get(normalized, normalized) + try: + increment_usage(language=language, token=token_value, amount=int(amount)) + except Exception: # pragma: no cover - defensive logging + job.add_log(f"Failed to record usage for override {token_value}", level="warning") + + +def chunk_text_for_tts(entry: Mapping[str, Any]) -> str: + """Choose the best source text for synthesis. + + We must prefer the raw chunk text (``text`` / ``original_text``) so + manual/pronunciation overrides can match against the original tokens + (e.g. censored words like ``Unfu*k``). ``normalized_text`` may have + already been run through ``normalize_for_pipeline``, which can remove + punctuation and prevent overrides from triggering. + """ + + if not isinstance(entry, Mapping): + return "" + return str( + entry.get("text") + or entry.get("original_text") + or entry.get("normalized_text") + or "" + ).strip() diff --git a/abogen/webui/conversion_runner.py b/abogen/webui/conversion_runner.py index e8a24b2..deb525b 100644 --- a/abogen/webui/conversion_runner.py +++ b/abogen/webui/conversion_runner.py @@ -89,6 +89,12 @@ from abogen.domain.voice_resolution import ( ) from abogen.domain.chapter_overrides import apply_chapter_overrides as _apply_chapter_overrides from abogen.domain.metadata_merge import merge_metadata as _merge_metadata +from abogen.domain.chunk_utils import ( + safe_int as _safe_int, + group_chunks_by_chapter as _group_chunks_by_chapter, + record_override_usage as _record_override_usage, + chunk_text_for_tts as _chunk_text_for_tts, +) from .service import Job, JobStatus @@ -211,68 +217,6 @@ def _normalize_for_pipeline( return normalize_for_pipeline(text, config=apostrophe_config, settings=runtime_settings) -def _group_chunks_by_chapter(chunks: Iterable[Dict[str, Any]]) -> Dict[int, List[Dict[str, Any]]]: - grouped: Dict[int, List[Dict[str, Any]]] = defaultdict(list) - for entry in chunks or []: - if not isinstance(entry, dict): - continue - try: - chapter_index = int(entry.get("chapter_index", 0)) - except (TypeError, ValueError): - chapter_index = 0 - grouped[chapter_index].append(dict(entry)) - - for chapter_index, items in grouped.items(): - items.sort(key=lambda payload: _safe_int(payload.get("chunk_index"))) - - return grouped - - -def _record_override_usage( - job: Job, - usage_counter: Mapping[str, int], - token_map: Mapping[str, str], -) -> None: - if not usage_counter: - return - - language = getattr(job, "language", "") or "a" - for normalized, amount in usage_counter.items(): - if amount <= 0: - continue - token_value = token_map.get(normalized, normalized) - try: - increment_usage(language=language, token=token_value, amount=int(amount)) - except Exception: # pragma: no cover - defensive logging - job.add_log(f"Failed to record usage for override {token_value}", level="warning") - - -def _safe_int(value: Any, default: int = 0) -> int: - try: - return int(value) - except (TypeError, ValueError): - return default - - -def _chunk_text_for_tts(entry: Mapping[str, Any]) -> str: - """Choose the best source text for synthesis. - - We must prefer the raw chunk text (`text` / `original_text`) so manual/pronunciation - overrides can match against the original tokens (e.g. censored words like `Unfu*k`). - `normalized_text` may have already been run through `normalize_for_pipeline`, which - can remove punctuation and prevent overrides from triggering. - """ - - if not isinstance(entry, Mapping): - return "" - return str( - entry.get("text") - or entry.get("original_text") - or entry.get("normalized_text") - or "" - ).strip() - - def _apply_m4b_chapters_with_mutagen( audio_path: Path, chapters: List[Dict[str, Any]], diff --git a/tests/test_chunk_helpers.py b/tests/test_chunk_helpers.py index ee9da79..49eb3e5 100644 --- a/tests/test_chunk_helpers.py +++ b/tests/test_chunk_helpers.py @@ -3,7 +3,8 @@ from __future__ import annotations from types import SimpleNamespace from abogen.chunking import chunk_text -from abogen.webui.conversion_runner import _chunk_voice_spec, _group_chunks_by_chapter +from abogen.domain.voice_resolution import chunk_voice_spec +from abogen.domain.chunk_utils import group_chunks_by_chapter def test_group_chunks_by_chapter_orders_and_groups() -> None: @@ -13,7 +14,7 @@ def test_group_chunks_by_chapter_orders_and_groups() -> None: {"chapter_index": 1, "chunk_index": 0, "text": "next"}, ] - grouped = _group_chunks_by_chapter(chunks) + grouped = group_chunks_by_chapter(chunks) assert [entry["text"] for entry in grouped[0]] == ["body", "tail"] assert grouped[1][0]["text"] == "next" @@ -23,7 +24,7 @@ def test_chunk_voice_spec_prefers_chunk_overrides() -> None: job = SimpleNamespace(voice="base_voice", speakers={}) chunk = {"voice": "override_voice", "speaker_id": "narrator"} - assert _chunk_voice_spec(job, chunk, "fallback") == "override_voice" + assert chunk_voice_spec(job, chunk, "fallback") == "override_voice" def test_chunk_voice_spec_falls_back_to_speaker_voice() -> None: @@ -32,14 +33,14 @@ def test_chunk_voice_spec_falls_back_to_speaker_voice() -> None: ) chunk = {"speaker_id": "narrator"} - assert _chunk_voice_spec(job, chunk, "fallback") == "speaker_voice" + assert chunk_voice_spec(job, chunk, "fallback") == "speaker_voice" def test_chunk_voice_spec_uses_fallback_when_no_overrides() -> None: job = SimpleNamespace(voice="base_voice", speakers={}) chunk = {"speaker_id": "unknown"} - assert _chunk_voice_spec(job, chunk, "fallback") == "fallback" + assert chunk_voice_spec(job, chunk, "fallback") == "fallback" def test_chunk_text_merges_title_abbreviations() -> None: diff --git a/tests/test_chunk_text_for_tts_prefers_raw.py b/tests/test_chunk_text_for_tts_prefers_raw.py index faf3c67..31b7615 100644 --- a/tests/test_chunk_text_for_tts_prefers_raw.py +++ b/tests/test_chunk_text_for_tts_prefers_raw.py @@ -1,4 +1,4 @@ -from abogen.webui.conversion_runner import _chunk_text_for_tts +from abogen.domain.chunk_utils import chunk_text_for_tts def test_chunk_text_for_tts_prefers_text_over_normalized_text(): @@ -9,7 +9,7 @@ def test_chunk_text_for_tts_prefers_text_over_normalized_text(): "text": "Unfu*k", } - assert _chunk_text_for_tts(entry) == "Unfu*k" + assert chunk_text_for_tts(entry) == "Unfu*k" def test_chunk_text_for_tts_falls_back_to_original_text_then_normalized_text(): @@ -17,9 +17,9 @@ def test_chunk_text_for_tts_falls_back_to_original_text_then_normalized_text(): "original_text": "Hello * world", "normalized_text": "Hello world", } - assert _chunk_text_for_tts(entry) == "Hello * world" + assert chunk_text_for_tts(entry) == "Hello * world" entry2 = { "normalized_text": "Only normalized", } - assert _chunk_text_for_tts(entry2) == "Only normalized" + assert chunk_text_for_tts(entry2) == "Only normalized" diff --git a/tests/test_chunk_utils.py b/tests/test_chunk_utils.py new file mode 100644 index 0000000..d08fc90 --- /dev/null +++ b/tests/test_chunk_utils.py @@ -0,0 +1,123 @@ +"""Tests for chunk processing utilities. + +Tests import from domain.chunk_utils (new location). +""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import patch + +import pytest + + +# --------------------------------------------------------------------------- +# safe_int +# --------------------------------------------------------------------------- + + +class TestSafeInt: + """safe_int safely converts to int with a default.""" + + def test_int_value(self): + from abogen.domain.chunk_utils import safe_int + + assert safe_int(42) == 42 + + def test_string_number(self): + from abogen.domain.chunk_utils import safe_int + + assert safe_int("7") == 7 + + def test_float_truncated(self): + from abogen.domain.chunk_utils import safe_int + + assert safe_int(3.9) == 3 + + def test_none_returns_default(self): + from abogen.domain.chunk_utils import safe_int + + assert safe_int(None) == 0 + + def test_garbage_returns_default(self): + from abogen.domain.chunk_utils import safe_int + + assert safe_int("abc") == 0 + + def test_custom_default(self): + from abogen.domain.chunk_utils import safe_int + + assert safe_int(None, default=-1) == -1 + + +# --------------------------------------------------------------------------- +# chunk_text_for_tts (supplement existing tests) +# --------------------------------------------------------------------------- + + +class TestChunkTextForTts: + """chunk_text_for_tts selects the best text source.""" + + def test_non_mapping_returns_empty(self): + from abogen.domain.chunk_utils import chunk_text_for_tts + + assert chunk_text_for_tts("not a dict") == "" + + def test_none_returns_empty(self): + from abogen.domain.chunk_utils import chunk_text_for_tts + + assert chunk_text_for_tts(None) == "" + + def test_empty_dict_returns_empty(self): + from abogen.domain.chunk_utils import chunk_text_for_tts + + assert chunk_text_for_tts({}) == "" + + def test_whitespace_only_returns_empty(self): + from abogen.domain.chunk_utils import chunk_text_for_tts + + assert chunk_text_for_tts({"text": " "}) == "" + + +# --------------------------------------------------------------------------- +# record_override_usage +# --------------------------------------------------------------------------- + + +class TestRecordOverrideUsage: + """record_override_usage records pronunciation override usage.""" + + def test_noop_when_empty(self): + from abogen.domain.chunk_utils import record_override_usage + + job = SimpleNamespace(language="en") + record_override_usage(job, {}, {}) + + def test_noop_when_all_zero(self): + from abogen.domain.chunk_utils import record_override_usage + + job = SimpleNamespace(language="en") + record_override_usage(job, {"hello": 0}, {"hello": "hi"}) + + def test_records_usage(self): + from abogen.domain.chunk_utils import record_override_usage + + job = SimpleNamespace(language="en", add_log=lambda *a, **kw: None) + with patch("abogen.domain.chunk_utils.increment_usage") as mock_inc: + record_override_usage(job, {"hello": 2}, {"hello": "hi"}) + mock_inc.assert_called_once_with(language="en", token="hi", amount=2) + + def test_fallback_token_from_normalized(self): + from abogen.domain.chunk_utils import record_override_usage + + job = SimpleNamespace(language="ja", add_log=lambda *a, **kw: None) + with patch("abogen.domain.chunk_utils.increment_usage") as mock_inc: + record_override_usage(job, {"test": 1}, {}) + mock_inc.assert_called_once_with(language="ja", token="test", amount=1) + + def test_handles_exception_gracefully(self): + from abogen.domain.chunk_utils import record_override_usage + + job = SimpleNamespace(language="en", add_log=lambda *a, **kw: None) + with patch("abogen.domain.chunk_utils.increment_usage", side_effect=RuntimeError("db error")): + record_override_usage(job, {"hello": 1}, {"hello": "hi"})