diff --git a/abogen/domain/voice_resolution.py b/abogen/domain/voice_resolution.py index 7612275..83bfd1f 100644 --- a/abogen/domain/voice_resolution.py +++ b/abogen/domain/voice_resolution.py @@ -8,7 +8,7 @@ from __future__ import annotations from typing import Any, Dict, Optional, Set -from abogen.tts_plugin.utils import get_voices +from abogen.tts_plugin.utils import get_voices, get_default_voice from abogen.voice_formulas import extract_voice_ids from abogen.voice_cache import ensure_voice_assets @@ -164,3 +164,27 @@ def chunk_voice_spec(job: Any, chunk: Dict[str, Any], fallback: str) -> str: if fallback: return fallback return job_voice_fallback(job) + + +def resolve_fallback_voice_spec( + base_spec: str, + job_voice: str, + voice_cache_keys: list[str], + provider: str = "kokoro", +) -> str: + """Resolve the voice spec for intro/outro with a priority fallback chain. + + Priority: base_spec → job_voice → first voice_cache key → default voice. + ``"__custom_mix"`` is treated as empty (it is not a usable voice spec). + """ + spec = base_spec or job_voice + if spec == "__custom_mix": + spec = job_voice or "" + if not spec: + for key in voice_cache_keys: + if key and key != "__custom_mix": + spec = key.split(":", 1)[-1] + break + if not spec: + spec = get_default_voice(provider) + return spec diff --git a/abogen/webui/conversion_runner.py b/abogen/webui/conversion_runner.py index 5bb9049..9cc008f 100644 --- a/abogen/webui/conversion_runner.py +++ b/abogen/webui/conversion_runner.py @@ -84,6 +84,7 @@ from abogen.domain.voice_resolution import ( initialize_voice_cache as _initialize_voice_cache, chapter_voice_spec as _chapter_voice_spec, chunk_voice_spec as _chunk_voice_spec, + resolve_fallback_voice_spec as _resolve_fallback_voice_spec, ) from abogen.domain.chapter_overrides import apply_chapter_overrides as _apply_chapter_overrides from abogen.domain.metadata_merge import merge_metadata as _merge_metadata @@ -535,15 +536,9 @@ def run_conversion_job(job: Job) -> None: preview = book_intro_text if len(book_intro_text) <= 120 else f"{book_intro_text[:117]}…" job.add_log(f"Title intro enabled: {preview}", level="debug") - intro_voice_spec = base_voice_spec or job.voice - if intro_voice_spec == "__custom_mix": - intro_voice_spec = base_voice_spec or "" - if not intro_voice_spec: - fallback_key = next(iter(voice_cache.keys()), "") - if fallback_key and fallback_key != "__custom_mix": - intro_voice_spec = fallback_key.split(":", 1)[-1] - if not intro_voice_spec: - intro_voice_spec = get_default_voice("kokoro") + intro_voice_spec = _resolve_fallback_voice_spec( + base_voice_spec, job.voice, list(voice_cache.keys()) + ) if intro_voice_spec: intro_provider, _, intro_voice_choice, intro_speed, intro_steps = resolve_voice_choice( @@ -960,16 +955,9 @@ def run_conversion_job(job: Job) -> None: if getattr(job, "read_closing_outro", True): outro_text = _build_outro_text(job.metadata_tags, job.original_filename) - outro_voice_spec = base_voice_spec or job.voice - if outro_voice_spec == "__custom_mix": - outro_voice_spec = base_voice_spec or "" - if not outro_voice_spec: - fallback_key = next(iter(voice_cache.keys()), "") - if fallback_key and fallback_key != "__custom_mix": - # `voice_cache` keys are internal and include provider prefixes. - outro_voice_spec = fallback_key.split(":", 1)[-1] - if not outro_voice_spec: - outro_voice_spec = get_default_voice("kokoro") + outro_voice_spec = _resolve_fallback_voice_spec( + base_voice_spec, job.voice, list(voice_cache.keys()) + ) if outro_text and outro_voice_spec: outro_start_time = current_time diff --git a/tests/test_voice_fallback.py b/tests/test_voice_fallback.py new file mode 100644 index 0000000..9746576 --- /dev/null +++ b/tests/test_voice_fallback.py @@ -0,0 +1,139 @@ +import sys +import types + +if "soundfile" not in sys.modules: + soundfile_stub = types.ModuleType("soundfile") + + class _SoundFileStub: # pragma: no cover - placeholder to satisfy imports + def __init__(self, *args: object, **kwargs: object) -> None: + raise RuntimeError("soundfile is not installed in the test environment") + + soundfile_stub.SoundFile = _SoundFileStub # type: ignore[attr-defined] + sys.modules["soundfile"] = soundfile_stub + +if "static_ffmpeg" not in sys.modules: + sys.modules["static_ffmpeg"] = types.ModuleType("static_ffmpeg") + +if "ebooklib" not in sys.modules: + ebooklib_stub = types.ModuleType("ebooklib") + ebooklib_epub_stub = types.ModuleType("ebooklib.epub") + ebooklib_stub.epub = ebooklib_epub_stub # type: ignore[attr-defined] + sys.modules["ebooklib"] = ebooklib_stub + sys.modules["ebooklib.epub"] = ebooklib_epub_stub + +if "fitz" not in sys.modules: + sys.modules["fitz"] = types.ModuleType("fitz") + +if "markdown" not in sys.modules: + markdown_stub = types.ModuleType("markdown") + + class _MarkdownStub: + def __init__(self, *args: object, **kwargs: object) -> None: + self.toc_tokens = [] + + def convert(self, text: str) -> str: + return text + + markdown_stub.Markdown = _MarkdownStub # type: ignore[attr-defined] + sys.modules["markdown"] = markdown_stub + +if "bs4" not in sys.modules: + bs4_stub = types.ModuleType("bs4") + + class _BeautifulSoupStub: + def __init__(self, *args: object, **kwargs: object) -> None: + self._text = "" + + def find(self, *args: object, **kwargs: object) -> None: + return None + + def get_text(self) -> str: + return self._text + + def decompose(self) -> None: # pragma: no cover - compatibility shim + return None + + class _NavigableStringStub(str): + pass + + bs4_stub.BeautifulSoup = _BeautifulSoupStub # type: ignore[attr-defined] + bs4_stub.NavigableString = _NavigableStringStub # type: ignore[attr-defined] + sys.modules["bs4"] = bs4_stub + + +from unittest.mock import patch, MagicMock + + +class TestResolveFallbackVoiceSpec: + """Tests for the voice fallback resolution helper.""" + + def test_uses_base_voice_spec(self) -> None: + from abogen.domain.voice_resolution import resolve_fallback_voice_spec + + result = resolve_fallback_voice_spec( + base_spec="af_heart", + job_voice="af_bella", + voice_cache_keys=[], + ) + assert result == "af_heart" + + def test_falls_back_to_job_voice(self) -> None: + from abogen.domain.voice_resolution import resolve_fallback_voice_spec + + result = resolve_fallback_voice_spec( + base_spec="", + job_voice="af_bella", + voice_cache_keys=[], + ) + assert result == "af_bella" + + def test_skips_custom_mix_uses_job_voice(self) -> None: + from abogen.domain.voice_resolution import resolve_fallback_voice_spec + + result = resolve_fallback_voice_spec( + base_spec="__custom_mix", + job_voice="af_bella", + voice_cache_keys=[], + ) + assert result == "af_bella" + + def test_falls_back_to_voice_cache(self) -> None: + from abogen.domain.voice_resolution import resolve_fallback_voice_spec + + result = resolve_fallback_voice_spec( + base_spec="", + job_voice="", + voice_cache_keys=["kokoro:af_heart"], + ) + assert result == "af_heart" + + def test_skips_custom_mix_in_cache(self) -> None: + from abogen.domain.voice_resolution import resolve_fallback_voice_spec + + result = resolve_fallback_voice_spec( + base_spec="", + job_voice="", + voice_cache_keys=["__custom_mix", "kokoro:af_heart"], + ) + assert result == "af_heart" + + def test_falls_back_to_default_voice(self) -> None: + from abogen.domain.voice_resolution import resolve_fallback_voice_spec + + with patch("abogen.domain.voice_resolution.get_default_voice", return_value="af_heart"): + result = resolve_fallback_voice_spec( + base_spec="", + job_voice="", + voice_cache_keys=[], + ) + assert result == "af_heart" + + def test_empty_base_and_job_with_cache(self) -> None: + from abogen.domain.voice_resolution import resolve_fallback_voice_spec + + result = resolve_fallback_voice_spec( + base_spec="", + job_voice="", + voice_cache_keys=["kokoro:af_bella", "kokoro:af_heart"], + ) + assert result == "af_bella"