refactor: extract chunk utils to domain/chunk_utils.py

- Extract safe_int, group_chunks_by_chapter, record_override_usage, chunk_text_for_tts
- Add tests/test_chunk_utils.py (15 tests)
- Update test_chunk_helpers.py and test_chunk_text_for_tts_prefers_raw.py imports
- conversion_runner.py: 1574 → 1518 lines
- All tests pass

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

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]],

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:

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"

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"})