From c94347b33b85fac1794989bba9faaa78b7e9a85a Mon Sep 17 00:00:00 2001 From: Artem Akymenko Date: Wed, 15 Jul 2026 10:20:58 +0000 Subject: [PATCH] refactor: extract voice resolution to domain/voice_resolution.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 | 166 ++++++++++++++++++ abogen/webui/conversion_runner.py | 161 +----------------- tests/test_voice_resolution.py | 274 ++++++++++++++++++++++++++++++ 3 files changed, 448 insertions(+), 153 deletions(-) create mode 100644 abogen/domain/voice_resolution.py create mode 100644 tests/test_voice_resolution.py diff --git a/abogen/domain/voice_resolution.py b/abogen/domain/voice_resolution.py new file mode 100644 index 0000000..7612275 --- /dev/null +++ b/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) diff --git a/abogen/webui/conversion_runner.py b/abogen/webui/conversion_runner.py index be39f6c..6de393e 100644 --- a/abogen/webui/conversion_runner.py +++ b/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 []: diff --git a/tests/test_voice_resolution.py b/tests/test_voice_resolution.py new file mode 100644 index 0000000..6c6dc14 --- /dev/null +++ b/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)