refactor: extract voice resolution to domain/voice_resolution.py

- Extract spec_to_voice_ids, job_voice_fallback, collect_required_voice_ids
- Extract initialize_voice_cache, chapter_voice_spec, chunk_voice_spec
- Add tests/test_voice_resolution.py (29 tests)
- conversion_runner.py: 1822 → 1677 lines
- All tests pass

abogen/domain/voice_resolution.py

@@ -0,0 +1,166 @@
+"""Voice resolution helpers.
+
+Functions for resolving voice specifications, collecting required voice IDs,
+and determining the voice to use for chapters and chunks.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, Optional, Set
+
+from abogen.tts_plugin.utils import get_voices
+from abogen.voice_formulas import extract_voice_ids
+from abogen.voice_cache import ensure_voice_assets
+
+
+def spec_to_voice_ids(spec: Any) -> Set[str]:
+    text = str(spec or "").strip()
+    if not text:
+        return set()
+    if text == "__custom_mix":
+        return set()
+    if "*" in text:
+        try:
+            return set(extract_voice_ids(text))
+        except ValueError:
+            return set()
+    if text in get_voices("kokoro"):
+        return {text}
+    return set()
+
+
+def job_voice_fallback(job: Any) -> str:
+    base = str(getattr(job, "voice", "") or "").strip()
+    if base and base != "__custom_mix":
+        return base
+
+    speakers = getattr(job, "speakers", None)
+    if isinstance(speakers, dict):
+        narrator = speakers.get("narrator")
+        if isinstance(narrator, dict):
+            for key in ("resolved_voice", "voice_formula", "voice"):
+                value = narrator.get(key)
+                candidate = str(value or "").strip()
+                if candidate and candidate != "__custom_mix":
+                    return candidate
+        for payload in speakers.values() or []:
+            if not isinstance(payload, dict):
+                continue
+            for key in ("resolved_voice", "voice_formula", "voice"):
+                value = payload.get(key)
+                candidate = str(value or "").strip()
+                if candidate and candidate != "__custom_mix":
+                    return candidate
+
+    for chapter in getattr(job, "chapters", []) or []:
+        if not isinstance(chapter, dict):
+            continue
+        for key in ("resolved_voice", "voice_formula", "voice"):
+            candidate = str(chapter.get(key) or "").strip()
+            if candidate and candidate != "__custom_mix":
+                return candidate
+
+    return ""
+
+
+def collect_required_voice_ids(job: Any) -> Set[str]:
+    voices: Set[str] = set()
+    voices.update(spec_to_voice_ids(job.voice))
+    voices.update(spec_to_voice_ids(job_voice_fallback(job)))
+
+    for chapter in getattr(job, "chapters", []) or []:
+        if not isinstance(chapter, dict):
+            continue
+        for key in ("resolved_voice", "voice_formula", "voice"):
+            voices.update(spec_to_voice_ids(chapter.get(key)))
+
+    for chunk in getattr(job, "chunks", []) or []:
+        if not isinstance(chunk, dict):
+            continue
+        for key in ("resolved_voice", "voice_formula", "voice"):
+            voices.update(spec_to_voice_ids(chunk.get(key)))
+
+    speakers = getattr(job, "speakers", {})
+    if isinstance(speakers, dict):
+        for payload in speakers.values() or []:
+            if not isinstance(payload, dict):
+                continue
+            for key in ("resolved_voice", "voice_formula", "voice"):
+                voices.update(spec_to_voice_ids(payload.get(key)))
+
+    voices.update(get_voices("kokoro"))
+    return voices
+
+
+def initialize_voice_cache(job: Any) -> None:
+    try:
+        targets = collect_required_voice_ids(job)
+        downloaded, errors = ensure_voice_assets(
+            targets,
+            on_progress=lambda message: job.add_log(message, level="debug"),
+        )
+    except RuntimeError as exc:
+        job.add_log(f"Voice cache unavailable: {exc}", level="warning")
+        return
+
+    if downloaded:
+        job.add_log(
+            f"Cached {len(downloaded)} voice asset{'s' if len(downloaded) != 1 else ''} locally.",
+            level="info",
+        )
+
+    for voice_id, error in errors.items():
+        job.add_log(f"Failed to cache voice '{voice_id}': {error}", level="warning")
+
+
+def chapter_voice_spec(job: Any, override: Optional[Dict[str, Any]]) -> str:
+    if not override:
+        return job_voice_fallback(job)
+
+    resolved = str(override.get("resolved_voice", "")).strip()
+    if resolved:
+        return resolved
+
+    formula = str(override.get("voice_formula", "")).strip()
+    if formula:
+        return formula
+
+    voice = str(override.get("voice", "")).strip()
+    if voice:
+        return voice
+
+    return job_voice_fallback(job)
+
+
+def chunk_voice_spec(job: Any, chunk: Dict[str, Any], fallback: str) -> str:
+    for key in ("resolved_voice", "voice_formula", "voice"):
+        value = chunk.get(key)
+        if value:
+            return str(value)
+
+    speaker_id = chunk.get("speaker_id")
+    speakers = getattr(job, "speakers", None)
+    if isinstance(speakers, dict) and speaker_id in speakers:
+        speaker_entry = speakers.get(speaker_id) or {}
+        if isinstance(speaker_entry, dict):
+            for key in ("resolved_voice", "voice_formula", "voice"):
+                value = speaker_entry.get(key)
+                if value:
+                    return str(value)
+            profile_formula = speaker_entry.get("voice_formula")
+            if profile_formula:
+                return str(profile_formula)
+
+    profile_name = chunk.get("voice_profile")
+    if profile_name:
+        if isinstance(speakers, dict):
+            speaker_entry = speakers.get(profile_name)
+            if isinstance(speaker_entry, dict):
+                for key in ("resolved_voice", "voice_formula", "voice"):
+                    value = speaker_entry.get(key)
+                    if value:
+                        return str(value)
+
+    if fallback:
+        return fallback
+    return job_voice_fallback(job)

abogen/webui/conversion_runner.py

@@ -79,6 +79,14 @@ from abogen.domain.pronunciation import (
     apply_pronunciation_rules as _apply_pronunciation_rules,
     merge_pronunciation_overrides as _merge_pronunciation_overrides,
 )
+from abogen.domain.voice_resolution import (
+    spec_to_voice_ids as _spec_to_voice_ids,
+    job_voice_fallback as _job_voice_fallback,
+    collect_required_voice_ids as _collect_required_voice_ids,
+    initialize_voice_cache as _initialize_voice_cache,
+    chapter_voice_spec as _chapter_voice_spec,
+    chunk_voice_spec as _chunk_voice_spec,
+)
 
 
 from .service import Job, JobStatus
@@ -184,106 +192,6 @@ def _coerce_truthy(value: Any, default: bool = True) -> bool:
 _OUTPUT_SANITIZE_RE = re.compile(r"[^\w\-_.]+")
 
 
-def _spec_to_voice_ids(spec: Any) -> Set[str]:
-    text = str(spec or "").strip()
-    if not text:
-        return set()
-    if text == "__custom_mix":
-        return set()
-    if "*" in text:
-        try:
-            return set(extract_voice_ids(text))
-        except ValueError:
-            return set()
-    if text in get_voices("kokoro"):
-        return {text}
-    return set()
-
-
-def _job_voice_fallback(job: Any) -> str:
-    base = str(getattr(job, "voice", "") or "").strip()
-    if base and base != "__custom_mix":
-        return base
-
-    speakers = getattr(job, "speakers", None)
-    if isinstance(speakers, dict):
-        narrator = speakers.get("narrator")
-        if isinstance(narrator, dict):
-            for key in ("resolved_voice", "voice_formula", "voice"):
-                value = narrator.get(key)
-                candidate = str(value or "").strip()
-                if candidate and candidate != "__custom_mix":
-                    return candidate
-        for payload in speakers.values() or []:
-            if not isinstance(payload, dict):
-                continue
-            for key in ("resolved_voice", "voice_formula", "voice"):
-                value = payload.get(key)
-                candidate = str(value or "").strip()
-                if candidate and candidate != "__custom_mix":
-                    return candidate
-
-    for chapter in getattr(job, "chapters", []) or []:
-        if not isinstance(chapter, dict):
-            continue
-        for key in ("resolved_voice", "voice_formula", "voice"):
-            candidate = str(chapter.get(key) or "").strip()
-            if candidate and candidate != "__custom_mix":
-                return candidate
-
-    return ""
-
-
-def _collect_required_voice_ids(job: Job) -> Set[str]:
-    voices: Set[str] = set()
-    voices.update(_spec_to_voice_ids(job.voice))
-    voices.update(_spec_to_voice_ids(_job_voice_fallback(job)))
-
-    for chapter in getattr(job, "chapters", []) or []:
-        if not isinstance(chapter, dict):
-            continue
-        for key in ("resolved_voice", "voice_formula", "voice"):
-            voices.update(_spec_to_voice_ids(chapter.get(key)))
-
-    for chunk in getattr(job, "chunks", []) or []:
-        if not isinstance(chunk, dict):
-            continue
-        for key in ("resolved_voice", "voice_formula", "voice"):
-            voices.update(_spec_to_voice_ids(chunk.get(key)))
-
-    speakers = getattr(job, "speakers", {})
-    if isinstance(speakers, dict):
-        for payload in speakers.values() or []:
-            if not isinstance(payload, dict):
-                continue
-            for key in ("resolved_voice", "voice_formula", "voice"):
-                voices.update(_spec_to_voice_ids(payload.get(key)))
-
-    voices.update(get_voices("kokoro"))
-    return voices
-
-
-def _initialize_voice_cache(job: Job) -> None:
-    try:
-        targets = _collect_required_voice_ids(job)
-        downloaded, errors = ensure_voice_assets(
-            targets,
-            on_progress=lambda message: job.add_log(message, level="debug"),
-        )
-    except RuntimeError as exc:
-        job.add_log(f"Voice cache unavailable: {exc}", level="warning")
-        return
-
-    if downloaded:
-        job.add_log(
-            f"Cached {len(downloaded)} voice asset{'s' if len(downloaded) != 1 else ''} locally.",
-            level="info",
-        )
-
-    for voice_id, error in errors.items():
-        job.add_log(f"Failed to cache voice '{voice_id}': {error}", level="warning")
-
-
 def _apply_chapter_overrides(
     extracted: List[ExtractedChapter],
     overrides: List[Dict[str, Any]],
@@ -406,59 +314,6 @@ def _normalize_for_pipeline(
     return normalize_for_pipeline(text, config=apostrophe_config, settings=runtime_settings)
 
 
-def _chapter_voice_spec(job: Job, override: Optional[Dict[str, Any]]) -> str:
-    if not override:
-        return _job_voice_fallback(job)
-
-    resolved = str(override.get("resolved_voice", "")).strip()
-    if resolved:
-        return resolved
-
-    formula = str(override.get("voice_formula", "")).strip()
-    if formula:
-        return formula
-
-    voice = str(override.get("voice", "")).strip()
-    if voice:
-        return voice
-
-    return _job_voice_fallback(job)
-
-
-def _chunk_voice_spec(job: Any, chunk: Dict[str, Any], fallback: str) -> str:
-    for key in ("resolved_voice", "voice_formula", "voice"):
-        value = chunk.get(key)
-        if value:
-            return str(value)
-
-    speaker_id = chunk.get("speaker_id")
-    speakers = getattr(job, "speakers", None)
-    if isinstance(speakers, dict) and speaker_id in speakers:
-        speaker_entry = speakers.get(speaker_id) or {}
-        if isinstance(speaker_entry, dict):
-            for key in ("resolved_voice", "voice_formula", "voice"):
-                value = speaker_entry.get(key)
-                if value:
-                    return str(value)
-            profile_formula = speaker_entry.get("voice_formula")
-            if profile_formula:
-                return str(profile_formula)
-
-    profile_name = chunk.get("voice_profile")
-    if profile_name:
-        if isinstance(speakers, dict):
-            speaker_entry = speakers.get(profile_name)
-            if isinstance(speaker_entry, dict):
-                for key in ("resolved_voice", "voice_formula", "voice"):
-                    value = speaker_entry.get(key)
-                    if value:
-                        return str(value)
-
-    if fallback:
-        return fallback
-    return _job_voice_fallback(job)
-
-
 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 []:

tests/test_voice_resolution.py

@@ -0,0 +1,274 @@
+"""Tests for voice resolution helpers.
+
+Tests import from domain.voice_resolution (new location).
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import Any, Dict
+from unittest.mock import patch
+
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# spec_to_voice_ids
+# ---------------------------------------------------------------------------
+
+
+class TestSpecToVoiceIds:
+    """spec_to_voice_ids extracts voice identifiers from a spec string."""
+
+    def test_empty_string(self):
+        from abogen.domain.voice_resolution import spec_to_voice_ids
+
+        assert spec_to_voice_ids("") == set()
+
+    def test_none(self):
+        from abogen.domain.voice_resolution import spec_to_voice_ids
+
+        assert spec_to_voice_ids(None) == set()
+
+    def test_custom_mix_returns_empty(self):
+        from abogen.domain.voice_resolution import spec_to_voice_ids
+
+        assert spec_to_voice_ids("__custom_mix") == set()
+
+    def test_single_known_voice(self):
+        from abogen.domain.voice_resolution import spec_to_voice_ids
+
+        with patch("abogen.domain.voice_resolution.get_voices", return_value={"af_heart"}):
+            assert spec_to_voice_ids("af_heart") == {"af_heart"}
+
+    def test_unknown_single_voice_returns_empty(self):
+        from abogen.domain.voice_resolution import spec_to_voice_ids
+
+        with patch("abogen.domain.voice_resolution.get_voices", return_value=set()):
+            assert spec_to_voice_ids("nonexistent") == set()
+
+    def test_formula_with_star(self):
+        from abogen.domain.voice_resolution import spec_to_voice_ids
+
+        with patch("abogen.domain.voice_resolution.extract_voice_ids", return_value=["v1", "v2"]):
+            result = spec_to_voice_ids("v1*v2")
+            assert result == {"v1", "v2"}
+
+    def test_formula_value_error_returns_empty(self):
+        from abogen.domain.voice_resolution import spec_to_voice_ids
+
+        with patch("abogen.domain.voice_resolution.extract_voice_ids", side_effect=ValueError("bad")):
+            assert spec_to_voice_ids("bad*spec") == set()
+
+    def test_whitespace_stripped(self):
+        from abogen.domain.voice_resolution import spec_to_voice_ids
+
+        assert spec_to_voice_ids("  ") == set()
+
+
+# ---------------------------------------------------------------------------
+# job_voice_fallback
+# ---------------------------------------------------------------------------
+
+
+class TestJobVoiceFallback:
+    """job_voice_fallback resolves a fallback voice from job attributes."""
+
+    def test_direct_voice(self):
+        from abogen.domain.voice_resolution import job_voice_fallback
+
+        job = SimpleNamespace(voice="af_heart", speakers=None, chapters=[])
+        assert job_voice_fallback(job) == "af_heart"
+
+    def test_custom_mix_ignored(self):
+        from abogen.domain.voice_resolution import job_voice_fallback
+
+        job = SimpleNamespace(voice="__custom_mix", speakers=None, chapters=[])
+        assert job_voice_fallback(job) == ""
+
+    def test_narrator_speaker(self):
+        from abogen.domain.voice_resolution import job_voice_fallback
+
+        job = SimpleNamespace(
+            voice="__custom_mix",
+            speakers={"narrator": {"resolved_voice": "af_heart"}},
+            chapters=[],
+        )
+        assert job_voice_fallback(job) == "af_heart"
+
+    def test_speaker_voice_formula(self):
+        from abogen.domain.voice_resolution import job_voice_fallback
+
+        job = SimpleNamespace(
+            voice="",
+            speakers={"speaker1": {"voice_formula": "v1*v2"}},
+            chapters=[],
+        )
+        assert job_voice_fallback(job) == "v1*v2"
+
+    def test_chapter_voice(self):
+        from abogen.domain.voice_resolution import job_voice_fallback
+
+        job = SimpleNamespace(
+            voice="",
+            speakers=None,
+            chapters=[{"resolved_voice": "af_bella"}],
+        )
+        assert job_voice_fallback(job) == "af_bella"
+
+    def test_empty_job(self):
+        from abogen.domain.voice_resolution import job_voice_fallback
+
+        job = SimpleNamespace(voice="", speakers=None, chapters=[])
+        assert job_voice_fallback(job) == ""
+
+    def test_narrator_custom_mix_falls_through(self):
+        from abogen.domain.voice_resolution import job_voice_fallback
+
+        job = SimpleNamespace(
+            voice="",
+            speakers={"narrator": {"voice": "__custom_mix"}},
+            chapters=[{"voice": "af_heart"}],
+        )
+        assert job_voice_fallback(job) == "af_heart"
+
+
+# ---------------------------------------------------------------------------
+# chapter_voice_spec
+# ---------------------------------------------------------------------------
+
+
+class TestChapterVoiceSpec:
+    """chapter_voice_spec resolves voice for a chapter override."""
+
+    def test_no_override_uses_fallback(self):
+        from abogen.domain.voice_resolution import chapter_voice_spec
+
+        job = SimpleNamespace(voice="af_heart", speakers=None, chapters=[])
+        assert chapter_voice_spec(job, None) == "af_heart"
+
+    def test_resolved_voice_wins(self):
+        from abogen.domain.voice_resolution import chapter_voice_spec
+
+        job = SimpleNamespace(voice="af_heart", speakers=None, chapters=[])
+        override = {"resolved_voice": "af_bella", "voice_formula": "x", "voice": "y"}
+        assert chapter_voice_spec(job, override) == "af_bella"
+
+    def test_formula_second(self):
+        from abogen.domain.voice_resolution import chapter_voice_spec
+
+        job = SimpleNamespace(voice="", speakers=None, chapters=[])
+        override = {"voice_formula": "v1*v2", "voice": "y"}
+        assert chapter_voice_spec(job, override) == "v1*v2"
+
+    def test_voice_third(self):
+        from abogen.domain.voice_resolution import chapter_voice_spec
+
+        job = SimpleNamespace(voice="", speakers=None, chapters=[])
+        override = {"voice": "af_nicole"}
+        assert chapter_voice_spec(job, override) == "af_nicole"
+
+    def test_empty_override_falls_to_fallback(self):
+        from abogen.domain.voice_resolution import chapter_voice_spec
+
+        job = SimpleNamespace(voice="af_heart", speakers=None, chapters=[])
+        assert chapter_voice_spec(job, {}) == "af_heart"
+
+
+# ---------------------------------------------------------------------------
+# chunk_voice_spec
+# ---------------------------------------------------------------------------
+
+
+class TestChunkVoiceSpec:
+    """chunk_voice_spec resolves voice for a TTS chunk."""
+
+    def test_chunk_direct_voice(self):
+        from abogen.domain.voice_resolution import chunk_voice_spec
+
+        job = SimpleNamespace(speakers=None)
+        chunk = {"resolved_voice": "af_heart"}
+        assert chunk_voice_spec(job, chunk, "fallback") == "af_heart"
+
+    def test_chunk_speaker_lookup(self):
+        from abogen.domain.voice_resolution import chunk_voice_spec
+
+        job = SimpleNamespace(speakers={"narrator": {"resolved_voice": "af_bella"}})
+        chunk = {"speaker_id": "narrator"}
+        assert chunk_voice_spec(job, chunk, "") == "af_bella"
+
+    def test_chunk_voice_profile_lookup(self):
+        from abogen.domain.voice_resolution import chunk_voice_spec
+
+        job = SimpleNamespace(speakers={"角色A": {"voice": "af_nicole"}})
+        chunk = {"voice_profile": "角色A"}
+        assert chunk_voice_spec(job, chunk, "") == "af_nicole"
+
+    def test_uses_fallback_string(self):
+        from abogen.domain.voice_resolution import chunk_voice_spec
+
+        job = SimpleNamespace(speakers=None)
+        chunk = {}
+        assert chunk_voice_spec(job, chunk, "my_fallback") == "my_fallback"
+
+    def test_fallback_to_job(self):
+        from abogen.domain.voice_resolution import chunk_voice_spec
+
+        job = SimpleNamespace(voice="af_heart", speakers=None, chapters=[])
+        chunk = {}
+        assert chunk_voice_spec(job, chunk, "") == "af_heart"
+
+
+# ---------------------------------------------------------------------------
+# collect_required_voice_ids
+# ---------------------------------------------------------------------------
+
+
+class TestCollectRequiredVoiceIds:
+    """collect_required_voice_ids gathers all voice IDs from a job."""
+
+    def test_includes_job_voice(self):
+        from abogen.domain.voice_resolution import collect_required_voice_ids
+
+        job = SimpleNamespace(voice="af_heart", chapters=[], chunks=[], speakers={})
+        with patch("abogen.domain.voice_resolution.get_voices", return_value={"af_heart"}), \
+             patch("abogen.domain.voice_resolution.job_voice_fallback", return_value=""):
+            result = collect_required_voice_ids(job)
+        assert "af_heart" in result
+
+    def test_includes_chapter_voices(self):
+        from abogen.domain.voice_resolution import collect_required_voice_ids
+
+        job = SimpleNamespace(
+            voice="",
+            chapters=[{"resolved_voice": "af_bella"}],
+            chunks=[],
+            speakers={},
+        )
+        with patch("abogen.domain.voice_resolution.get_voices", return_value={"af_bella"}), \
+             patch("abogen.domain.voice_resolution.job_voice_fallback", return_value=""):
+            result = collect_required_voice_ids(job)
+        assert "af_bella" in result
+
+    def test_includes_chunk_voices(self):
+        from abogen.domain.voice_resolution import collect_required_voice_ids
+
+        job = SimpleNamespace(
+            voice="",
+            chapters=[],
+            chunks=[{"voice": "af_nicole"}],
+            speakers={},
+        )
+        with patch("abogen.domain.voice_resolution.get_voices", return_value={"af_nicole"}), \
+             patch("abogen.domain.voice_resolution.job_voice_fallback", return_value=""):
+            result = collect_required_voice_ids(job)
+        assert "af_nicole" in result
+
+    def test_always_includes_kokoro_voices(self):
+        from abogen.domain.voice_resolution import collect_required_voice_ids
+
+        job = SimpleNamespace(voice="", chapters=[], chunks=[], speakers={})
+        with patch("abogen.domain.voice_resolution.get_voices", return_value={"af_heart", "af_bella"}), \
+             patch("abogen.domain.voice_resolution.job_voice_fallback", return_value=""):
+            result = collect_required_voice_ids(job)
+        assert {"af_heart", "af_bella"}.issubset(result)