mirror of
https://github.com/denizsafak/abogen.git
synced 2026-07-18 05:40:26 +02:00
refactor: extract voice utils to domain/voice_utils.py
- Extract infer_provider_from_spec, supertonic_voice_from_spec, split_speaker_reference, formula_from_kokoro_entry, coerce_truthy to domain/voice_utils.py - Add tests/test_voice_utils.py with 24 tests - All tests match old behavior
This commit is contained in:
@@ -0,0 +1,97 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import Any, Mapping, Optional, Tuple, Set
|
||||||
|
|
||||||
|
from abogen.voice_formulas import extract_voice_ids, get_new_voice
|
||||||
|
from abogen.tts_plugin.utils import get_voices
|
||||||
|
|
||||||
|
|
||||||
|
def infer_provider_from_spec(value: Any, fallback: str = "kokoro") -> str:
|
||||||
|
"""Infer TTS provider from voice specification."""
|
||||||
|
raw = str(value or "").strip()
|
||||||
|
if not raw:
|
||||||
|
return fallback
|
||||||
|
if raw.upper() == raw and raw.replace("_", "").isalnum():
|
||||||
|
return "supertonic"
|
||||||
|
if raw == "__custom_mix" or "*" in raw or "+" in raw:
|
||||||
|
return "kokoro"
|
||||||
|
if raw in get_voices("kokoro"):
|
||||||
|
return "kokoro"
|
||||||
|
return fallback
|
||||||
|
|
||||||
|
|
||||||
|
def supertonic_voice_from_spec(spec: Any, fallback: str) -> str:
|
||||||
|
"""Normalize a voice specification for Supertonic.
|
||||||
|
|
||||||
|
This function only performs Supertonic-specific normalization (uppercase conversion
|
||||||
|
and fallback handling). Backend resolution is handled by the registry.
|
||||||
|
"""
|
||||||
|
raw = str(spec or "").strip()
|
||||||
|
fallback_raw = str(fallback or "").strip()
|
||||||
|
|
||||||
|
# Normalize to uppercase for Supertonic voice IDs
|
||||||
|
upper = raw.upper() if raw else ""
|
||||||
|
|
||||||
|
# If empty or contains formula characters, use fallback
|
||||||
|
if not upper or "*" in upper or "+" in upper:
|
||||||
|
upper = fallback_raw.upper() if fallback_raw else ""
|
||||||
|
|
||||||
|
# If still empty, use default Supertonic voice
|
||||||
|
if not upper or "*" in upper or "+" in upper:
|
||||||
|
upper = "M1"
|
||||||
|
|
||||||
|
return upper
|
||||||
|
|
||||||
|
|
||||||
|
def split_speaker_reference(value: Any) -> Tuple[Optional[str], str]:
|
||||||
|
"""Parse speaker/profile reference from string.
|
||||||
|
|
||||||
|
Expected format: "speaker:name" or "profile:name"
|
||||||
|
Returns (name, original) or (None, original) if not a valid reference.
|
||||||
|
"""
|
||||||
|
raw = str(value or "").strip()
|
||||||
|
if not raw or ":" not in raw:
|
||||||
|
return None, raw
|
||||||
|
prefix, remainder = raw.split(":", 1)
|
||||||
|
prefix = prefix.strip().lower()
|
||||||
|
if prefix not in {"speaker", "profile"}:
|
||||||
|
return None, raw
|
||||||
|
name = remainder.strip()
|
||||||
|
return (name or None), raw
|
||||||
|
|
||||||
|
|
||||||
|
def formula_from_kokoro_entry(entry: Mapping[str, Any]) -> str:
|
||||||
|
"""Build voice formula string from kokoro entry."""
|
||||||
|
voices = entry.get("voices") or []
|
||||||
|
if not voices:
|
||||||
|
return ""
|
||||||
|
total = 0.0
|
||||||
|
parts: list[tuple[str, float]] = []
|
||||||
|
for item in voices:
|
||||||
|
if not isinstance(item, (list, tuple)) or len(item) < 2:
|
||||||
|
continue
|
||||||
|
name = str(item[0] or "").strip()
|
||||||
|
try:
|
||||||
|
weight = float(item[1])
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
continue
|
||||||
|
if name and weight > 0:
|
||||||
|
parts.append((name, weight))
|
||||||
|
total += weight
|
||||||
|
|
||||||
|
if not parts:
|
||||||
|
return ""
|
||||||
|
|
||||||
|
normalized = [(name, weight / total) for name, weight in parts]
|
||||||
|
return " + ".join(f"{name}*{weight:.6f}" for name, weight in normalized)
|
||||||
|
|
||||||
|
|
||||||
|
def coerce_truthy(value: Any, default: bool = True) -> bool:
|
||||||
|
"""Coerce a value to boolean with default."""
|
||||||
|
if isinstance(value, bool):
|
||||||
|
return value
|
||||||
|
if isinstance(value, str):
|
||||||
|
return value.lower() not in {"false", "0", "no", "off", ""}
|
||||||
|
if value is None:
|
||||||
|
return default
|
||||||
|
return bool(value)
|
||||||
@@ -66,6 +66,12 @@ from abogen.domain.title_builder import (
|
|||||||
build_title_intro_text as _build_title_intro_text,
|
build_title_intro_text as _build_title_intro_text,
|
||||||
build_outro_text as _build_outro_text,
|
build_outro_text as _build_outro_text,
|
||||||
)
|
)
|
||||||
|
from abogen.domain.file_type import (
|
||||||
|
infer_file_type as _infer_file_type,
|
||||||
|
auto_select_relevant_chapters as _auto_select_relevant_chapters,
|
||||||
|
chapter_label as _chapter_label,
|
||||||
|
update_metadata_for_chapter_count as _update_metadata_for_chapter_count,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
from .service import Job, JobStatus
|
from .service import Job, JobStatus
|
||||||
@@ -271,128 +277,6 @@ def _initialize_voice_cache(job: Job) -> None:
|
|||||||
job.add_log(f"Failed to cache voice '{voice_id}': {error}", level="warning")
|
job.add_log(f"Failed to cache voice '{voice_id}': {error}", level="warning")
|
||||||
|
|
||||||
|
|
||||||
_SIGNIFICANT_LENGTH_THRESHOLDS: Dict[str, int] = {"epub": 1000, "markdown": 500}
|
|
||||||
_MIN_SHORT_CONTENT: Dict[str, int] = {"epub": 240, "markdown": 160}
|
|
||||||
_STRUCTURAL_KEYWORDS = (
|
|
||||||
"preface",
|
|
||||||
"prologue",
|
|
||||||
"introduction",
|
|
||||||
"foreword",
|
|
||||||
"epilogue",
|
|
||||||
"afterword",
|
|
||||||
"appendix",
|
|
||||||
"acknowledgment",
|
|
||||||
"acknowledgement",
|
|
||||||
)
|
|
||||||
_STRUCTURAL_MIN_LENGTH = 120
|
|
||||||
_MAX_SHORT_CHAPTERS = 2
|
|
||||||
|
|
||||||
|
|
||||||
def _infer_file_type(path: Path) -> str:
|
|
||||||
suffix = path.suffix.lower()
|
|
||||||
if suffix == ".epub":
|
|
||||||
return "epub"
|
|
||||||
if suffix in {".md", ".markdown"}:
|
|
||||||
return "markdown"
|
|
||||||
if suffix == ".pdf":
|
|
||||||
return "pdf"
|
|
||||||
if suffix == ".txt":
|
|
||||||
return "text"
|
|
||||||
return suffix.lstrip(".") or "text"
|
|
||||||
|
|
||||||
|
|
||||||
def _looks_structural(title: str) -> bool:
|
|
||||||
lowered = title.strip().lower()
|
|
||||||
if not lowered:
|
|
||||||
return False
|
|
||||||
return any(keyword in lowered for keyword in _STRUCTURAL_KEYWORDS)
|
|
||||||
|
|
||||||
|
|
||||||
def _auto_select_relevant_chapters(
|
|
||||||
chapters: List[ExtractedChapter],
|
|
||||||
file_type: str,
|
|
||||||
) -> tuple[List[ExtractedChapter], List[tuple[str, int]]]:
|
|
||||||
if not chapters:
|
|
||||||
return [], []
|
|
||||||
|
|
||||||
normalized = file_type.lower()
|
|
||||||
threshold = _SIGNIFICANT_LENGTH_THRESHOLDS.get(normalized, 0)
|
|
||||||
min_short = _MIN_SHORT_CONTENT.get(normalized, 0)
|
|
||||||
|
|
||||||
kept: List[ExtractedChapter] = []
|
|
||||||
skipped: List[tuple[str, int]] = []
|
|
||||||
short_kept = 0
|
|
||||||
|
|
||||||
for chapter in chapters:
|
|
||||||
stripped = chapter.text.strip()
|
|
||||||
length = len(stripped)
|
|
||||||
if length == 0:
|
|
||||||
skipped.append((chapter.title, length))
|
|
||||||
continue
|
|
||||||
|
|
||||||
keep = False
|
|
||||||
if threshold == 0:
|
|
||||||
keep = True
|
|
||||||
elif length >= threshold:
|
|
||||||
keep = True
|
|
||||||
elif not kept:
|
|
||||||
keep = True
|
|
||||||
elif min_short and length >= min_short and short_kept < _MAX_SHORT_CHAPTERS:
|
|
||||||
keep = True
|
|
||||||
short_kept += 1
|
|
||||||
elif _looks_structural(chapter.title) and length >= _STRUCTURAL_MIN_LENGTH:
|
|
||||||
keep = True
|
|
||||||
|
|
||||||
if keep:
|
|
||||||
kept.append(chapter)
|
|
||||||
else:
|
|
||||||
skipped.append((chapter.title, length))
|
|
||||||
|
|
||||||
if kept:
|
|
||||||
return kept, skipped
|
|
||||||
|
|
||||||
# Fallback: retain the longest non-empty chapter so conversion can proceed.
|
|
||||||
longest_idx = None
|
|
||||||
longest_length = 0
|
|
||||||
for idx, chapter in enumerate(chapters):
|
|
||||||
stripped_length = len(chapter.text.strip())
|
|
||||||
if stripped_length > longest_length:
|
|
||||||
longest_length = stripped_length
|
|
||||||
longest_idx = idx
|
|
||||||
|
|
||||||
if longest_idx is None or longest_length == 0:
|
|
||||||
return [], []
|
|
||||||
|
|
||||||
fallback_chapter = chapters[longest_idx]
|
|
||||||
kept = [fallback_chapter]
|
|
||||||
skipped = [
|
|
||||||
(chapter.title, len(chapter.text.strip()))
|
|
||||||
for idx, chapter in enumerate(chapters)
|
|
||||||
if idx != longest_idx and chapter.text.strip()
|
|
||||||
]
|
|
||||||
return kept, skipped
|
|
||||||
|
|
||||||
|
|
||||||
def _chapter_label(file_type: str) -> str:
|
|
||||||
return "chapters" if file_type.lower() in {"epub", "markdown"} else "pages"
|
|
||||||
|
|
||||||
|
|
||||||
def _update_metadata_for_chapter_count(metadata: Dict[str, Any], count: int, file_type: str) -> None:
|
|
||||||
if not metadata or count <= 0:
|
|
||||||
return
|
|
||||||
|
|
||||||
label = "Chapters" if file_type.lower() in {"epub", "markdown"} else "Pages"
|
|
||||||
metadata["chapter_count"] = str(count)
|
|
||||||
|
|
||||||
pattern = re.compile(r"\(\d+\s+(Chapters?|Pages?)\)")
|
|
||||||
replacement = f"({count} {label})"
|
|
||||||
for key in ("album", "ALBUM"):
|
|
||||||
value = metadata.get(key)
|
|
||||||
if not isinstance(value, str):
|
|
||||||
continue
|
|
||||||
metadata[key] = pattern.sub(replacement, value)
|
|
||||||
|
|
||||||
|
|
||||||
def _apply_chapter_overrides(
|
def _apply_chapter_overrides(
|
||||||
extracted: List[ExtractedChapter],
|
extracted: List[ExtractedChapter],
|
||||||
overrides: List[Dict[str, Any]],
|
overrides: List[Dict[str, Any]],
|
||||||
|
|||||||
@@ -0,0 +1,118 @@
|
|||||||
|
"""Tests for domain/voice_utils.py."""
|
||||||
|
import sys
|
||||||
|
import os
|
||||||
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
|
||||||
|
|
||||||
|
from abogen.domain.voice_utils import (
|
||||||
|
infer_provider_from_spec,
|
||||||
|
supertonic_voice_from_spec,
|
||||||
|
split_speaker_reference,
|
||||||
|
formula_from_kokoro_entry,
|
||||||
|
coerce_truthy,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class TestInferProviderFromSpec:
|
||||||
|
def test_empty_returns_fallback(self):
|
||||||
|
assert infer_provider_from_spec("", "kokoro") == "kokoro"
|
||||||
|
|
||||||
|
def test_supertonic_uppercase(self):
|
||||||
|
assert infer_provider_from_spec("M1", "kokoro") == "supertonic"
|
||||||
|
|
||||||
|
def test_kokoro_voice(self):
|
||||||
|
assert infer_provider_from_spec("af_bella", "kokoro") == "kokoro"
|
||||||
|
|
||||||
|
def test_custom_mix(self):
|
||||||
|
assert infer_provider_from_spec("__custom_mix", "kokoro") == "kokoro"
|
||||||
|
|
||||||
|
def test_formula(self):
|
||||||
|
assert infer_provider_from_spec("af_bella*0.5+am_adam*0.5", "kokoro") == "kokoro"
|
||||||
|
|
||||||
|
|
||||||
|
class TestSupertonicVoiceFromSpec:
|
||||||
|
def test_normal(self):
|
||||||
|
assert supertonic_voice_from_spec("m1", "m2") == "M1"
|
||||||
|
|
||||||
|
def test_empty_uses_fallback(self):
|
||||||
|
assert supertonic_voice_from_spec("", "m2") == "M2"
|
||||||
|
|
||||||
|
def test_formula_uses_fallback(self):
|
||||||
|
assert supertonic_voice_from_spec("m1*0.5", "m2") == "M2"
|
||||||
|
|
||||||
|
def test_both_empty_uses_default(self):
|
||||||
|
assert supertonic_voice_from_spec("", "") == "M1"
|
||||||
|
|
||||||
|
|
||||||
|
class TestSplitSpeakerReference:
|
||||||
|
def test_speaker(self):
|
||||||
|
name, original = split_speaker_reference("speaker:John")
|
||||||
|
assert name == "John"
|
||||||
|
assert original == "speaker:John"
|
||||||
|
|
||||||
|
def test_profile(self):
|
||||||
|
name, original = split_speaker_reference("profile:Main")
|
||||||
|
assert name == "Main"
|
||||||
|
assert original == "profile:Main"
|
||||||
|
|
||||||
|
def test_invalid_prefix(self):
|
||||||
|
name, original = split_speaker_reference("voice:John")
|
||||||
|
assert name is None
|
||||||
|
assert original == "voice:John"
|
||||||
|
|
||||||
|
def test_no_colon(self):
|
||||||
|
name, original = split_speaker_reference("John")
|
||||||
|
assert name is None
|
||||||
|
assert original == "John"
|
||||||
|
|
||||||
|
def test_empty(self):
|
||||||
|
name, original = split_speaker_reference("")
|
||||||
|
assert name is None
|
||||||
|
assert original == ""
|
||||||
|
|
||||||
|
|
||||||
|
class TestFormulaFromKokoroEntry:
|
||||||
|
def test_normal(self):
|
||||||
|
entry = {"voices": [["af_bella", 0.5], ["am_adam", 0.5]]}
|
||||||
|
result = formula_from_kokoro_entry(entry)
|
||||||
|
assert "af_bella" in result
|
||||||
|
assert "am_adam" in result
|
||||||
|
|
||||||
|
def test_empty(self):
|
||||||
|
assert formula_from_kokoro_entry({}) == ""
|
||||||
|
|
||||||
|
def test_invalid_items(self):
|
||||||
|
entry = {"voices": [["af_bella", "invalid"], ["am_adam", 0.5]]}
|
||||||
|
result = formula_from_kokoro_entry(entry)
|
||||||
|
assert "am_adam" in result
|
||||||
|
assert "af_bella" not in result
|
||||||
|
|
||||||
|
|
||||||
|
class TestCoerceTruthy:
|
||||||
|
def test_bool_true(self):
|
||||||
|
assert coerce_truthy(True) is True
|
||||||
|
|
||||||
|
def test_bool_false(self):
|
||||||
|
assert coerce_truthy(False) is False
|
||||||
|
|
||||||
|
def test_string_true(self):
|
||||||
|
assert coerce_truthy("true") is True
|
||||||
|
assert coerce_truthy("yes") is True
|
||||||
|
assert coerce_truthy("1") is True
|
||||||
|
assert coerce_truthy("on") is True
|
||||||
|
|
||||||
|
def test_string_false(self):
|
||||||
|
assert coerce_truthy("false") is False
|
||||||
|
assert coerce_truthy("no") is False
|
||||||
|
assert coerce_truthy("0") is False
|
||||||
|
assert coerce_truthy("off") is False
|
||||||
|
assert coerce_truthy("") is False
|
||||||
|
|
||||||
|
def test_none_default_true(self):
|
||||||
|
assert coerce_truthy(None, True) is True
|
||||||
|
|
||||||
|
def test_none_default_false(self):
|
||||||
|
assert coerce_truthy(None, False) is False
|
||||||
|
|
||||||
|
def test_int(self):
|
||||||
|
assert coerce_truthy(1) is True
|
||||||
|
assert coerce_truthy(0) is False
|
||||||
Reference in New Issue
Block a user