mirror of
https://github.com/denizsafak/abogen.git
synced 2026-09-20 11:40:57 +02:00
refactor: move speaker metadata logic to domain layer
Move build_narrator_roster, build_speaker_roster, match_configured_speaker, apply_speaker_config_to_roster, prepare_speaker_metadata from webui/routes/utils/voice.py to domain/speaker_metadata.py. - prepare_speaker_metadata now accepts optional inject_recommended callback for UI-specific voice enrichment - 47 new tests in test_domain_speaker_metadata.py - All 1509 tests pass
This commit is contained in:
Artem Akymenko
@@ -0,0 +1,381 @@
|
||||
"""Speaker metadata functions for building and applying speaker rosters.
|
||||
|
||||
This module contains the core logic for:
|
||||
- Building narrator and speaker rosters from analysis results
|
||||
- Matching speakers to configured presets
|
||||
- Applying speaker config presets to rosters
|
||||
- Preparing full speaker metadata for conversion
|
||||
|
||||
Moved from webui/routes/utils/voice.py to be available across all UIs.
|
||||
"""
|
||||
|
||||
from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple, cast
|
||||
|
||||
from abogen.speaker_analysis import analyze_speakers
|
||||
from abogen.speaker_configs import slugify_label
|
||||
from abogen.domain.settings_core import load_settings
|
||||
|
||||
|
||||
def build_narrator_roster(
|
||||
voice: str,
|
||||
voice_profile: Optional[str],
|
||||
existing: Optional[Mapping[str, Any]] = None,
|
||||
) -> Dict[str, Any]:
|
||||
roster: Dict[str, Any] = {
|
||||
"narrator": {
|
||||
"id": "narrator",
|
||||
"label": "Narrator",
|
||||
"voice": voice,
|
||||
}
|
||||
}
|
||||
if voice_profile:
|
||||
roster["narrator"]["voice_profile"] = voice_profile
|
||||
existing_entry: Optional[Mapping[str, Any]] = None
|
||||
if existing is not None:
|
||||
existing_entry = existing.get("narrator") if isinstance(existing, Mapping) else None
|
||||
if isinstance(existing_entry, Mapping):
|
||||
roster_entry = roster["narrator"]
|
||||
for key in ("label", "voice", "voice_profile", "voice_formula", "pronunciation"):
|
||||
value = existing_entry.get(key)
|
||||
if value is not None and value != "":
|
||||
roster_entry[key] = value
|
||||
return roster
|
||||
|
||||
|
||||
def build_speaker_roster(
|
||||
analysis: Dict[str, Any],
|
||||
base_voice: str,
|
||||
voice_profile: Optional[str],
|
||||
existing: Optional[Mapping[str, Any]] = None,
|
||||
order: Optional[Iterable[str]] = None,
|
||||
) -> Dict[str, Any]:
|
||||
roster = build_narrator_roster(base_voice, voice_profile, existing)
|
||||
existing_map: Dict[str, Any] = dict(existing) if isinstance(existing, Mapping) else {}
|
||||
speakers = analysis.get("speakers", {}) if isinstance(analysis, dict) else {}
|
||||
ordered_ids: Iterable[str]
|
||||
if order is not None:
|
||||
ordered_ids = [sid for sid in order if sid in speakers]
|
||||
else:
|
||||
ordered_ids = speakers.keys()
|
||||
|
||||
for speaker_id in ordered_ids:
|
||||
payload = speakers.get(speaker_id, {})
|
||||
if speaker_id == "narrator":
|
||||
continue
|
||||
if isinstance(payload, Mapping) and payload.get("suppressed"):
|
||||
continue
|
||||
previous = existing_map.get(speaker_id)
|
||||
roster[speaker_id] = {
|
||||
"id": speaker_id,
|
||||
"label": payload.get("label") or speaker_id.replace("_", " ").title(),
|
||||
"analysis_confidence": payload.get("confidence"),
|
||||
"analysis_count": payload.get("count"),
|
||||
"gender": payload.get("gender", "unknown"),
|
||||
}
|
||||
detected_gender = payload.get("detected_gender")
|
||||
if detected_gender:
|
||||
roster[speaker_id]["detected_gender"] = detected_gender
|
||||
samples = payload.get("sample_quotes")
|
||||
if isinstance(samples, list):
|
||||
roster[speaker_id]["sample_quotes"] = samples
|
||||
if isinstance(previous, Mapping):
|
||||
for key in ("voice", "voice_profile", "voice_formula", "resolved_voice", "pronunciation"):
|
||||
value = previous.get(key)
|
||||
if value is not None and value != "":
|
||||
roster[speaker_id][key] = value
|
||||
if "sample_quotes" not in roster[speaker_id]:
|
||||
prev_samples = previous.get("sample_quotes")
|
||||
if isinstance(prev_samples, list):
|
||||
roster[speaker_id]["sample_quotes"] = prev_samples
|
||||
if "detected_gender" not in roster[speaker_id]:
|
||||
prev_detected = previous.get("detected_gender")
|
||||
if isinstance(prev_detected, str) and prev_detected:
|
||||
roster[speaker_id]["detected_gender"] = prev_detected
|
||||
return roster
|
||||
|
||||
|
||||
def match_configured_speaker(
|
||||
config_speakers: Mapping[str, Any],
|
||||
roster_id: str,
|
||||
roster_label: str,
|
||||
) -> Optional[Mapping[str, Any]]:
|
||||
if not config_speakers:
|
||||
return None
|
||||
entry = config_speakers.get(roster_id)
|
||||
if entry:
|
||||
return cast(Mapping[str, Any], entry)
|
||||
slug = slugify_label(roster_label)
|
||||
if slug != roster_id and slug in config_speakers:
|
||||
return cast(Mapping[str, Any], config_speakers[slug])
|
||||
lower_label = roster_label.strip().lower()
|
||||
for record in config_speakers.values():
|
||||
if not isinstance(record, Mapping):
|
||||
continue
|
||||
if str(record.get("label", "")).strip().lower() == lower_label:
|
||||
return record
|
||||
return None
|
||||
|
||||
|
||||
def apply_speaker_config_to_roster(
|
||||
roster: Mapping[str, Any],
|
||||
config: Optional[Mapping[str, Any]],
|
||||
*,
|
||||
persist_changes: bool = False,
|
||||
fallback_languages: Optional[Iterable[str]] = None,
|
||||
) -> Tuple[Dict[str, Any], List[str], Optional[Dict[str, Any]]]:
|
||||
if not isinstance(roster, Mapping):
|
||||
effective_languages = [code for code in (fallback_languages or []) if isinstance(code, str) and code]
|
||||
return {}, effective_languages, None
|
||||
updated_roster: Dict[str, Any] = {key: dict(value) for key, value in roster.items() if isinstance(value, Mapping)}
|
||||
if not config:
|
||||
effective_languages = [code for code in (fallback_languages or []) if isinstance(code, str) and code]
|
||||
return updated_roster, effective_languages, None
|
||||
|
||||
speakers_map = config.get("speakers")
|
||||
if not isinstance(speakers_map, Mapping):
|
||||
effective_languages = [code for code in (fallback_languages or []) if isinstance(code, str) and code]
|
||||
return updated_roster, effective_languages, None
|
||||
|
||||
config_languages = config.get("languages")
|
||||
if isinstance(config_languages, list):
|
||||
allowed_languages = [code for code in config_languages if isinstance(code, str) and code]
|
||||
else:
|
||||
allowed_languages = []
|
||||
if not allowed_languages and fallback_languages:
|
||||
allowed_languages = [code for code in fallback_languages if isinstance(code, str) and code]
|
||||
|
||||
default_voice = config.get("default_voice") if isinstance(config.get("default_voice"), str) else ""
|
||||
used_voices = {entry.get("resolved_voice") or entry.get("voice") for entry in updated_roster.values()} - {None}
|
||||
narrator_voice = ""
|
||||
narrator_entry = updated_roster.get("narrator") if isinstance(updated_roster, Mapping) else None
|
||||
if isinstance(narrator_entry, Mapping):
|
||||
narrator_voice = str(
|
||||
narrator_entry.get("resolved_voice")
|
||||
or narrator_entry.get("default_voice")
|
||||
or ""
|
||||
).strip()
|
||||
if narrator_voice:
|
||||
used_voices.add(narrator_voice)
|
||||
|
||||
config_changed = False
|
||||
new_config_payload: Dict[str, Any] = {
|
||||
"language": config.get("language", "a"),
|
||||
"languages": allowed_languages,
|
||||
"default_voice": default_voice,
|
||||
"speakers": dict(speakers_map),
|
||||
"version": config.get("version", 1),
|
||||
"notes": config.get("notes", ""),
|
||||
}
|
||||
|
||||
speakers_payload = new_config_payload["speakers"]
|
||||
|
||||
for speaker_id, roster_entry in updated_roster.items():
|
||||
if speaker_id == "narrator":
|
||||
continue
|
||||
label = str(roster_entry.get("label") or speaker_id)
|
||||
config_entry = match_configured_speaker(speakers_map, speaker_id, label)
|
||||
if config_entry is None:
|
||||
continue
|
||||
voice_id = str(config_entry.get("voice") or "").strip()
|
||||
voice_profile = str(config_entry.get("voice_profile") or "").strip()
|
||||
voice_formula = str(config_entry.get("voice_formula") or "").strip()
|
||||
resolved_voice = str(config_entry.get("resolved_voice") or "").strip()
|
||||
languages = config_entry.get("languages") if isinstance(config_entry.get("languages"), list) else []
|
||||
chosen_voice = resolved_voice or voice_formula or voice_id or roster_entry.get("voice")
|
||||
usable_languages = languages or allowed_languages
|
||||
|
||||
if chosen_voice:
|
||||
roster_entry["resolved_voice"] = chosen_voice
|
||||
roster_entry["voice"] = chosen_voice if not voice_profile and not voice_formula else roster_entry.get("voice", chosen_voice)
|
||||
if voice_profile:
|
||||
roster_entry["voice_profile"] = voice_profile
|
||||
if voice_formula:
|
||||
roster_entry["voice_formula"] = voice_formula
|
||||
roster_entry["resolved_voice"] = voice_formula
|
||||
if not voice_formula and not voice_profile and resolved_voice:
|
||||
roster_entry["resolved_voice"] = resolved_voice
|
||||
roster_entry["config_languages"] = usable_languages or []
|
||||
|
||||
if chosen_voice:
|
||||
used_voices.add(chosen_voice)
|
||||
|
||||
# persist updates back to config payload if required
|
||||
if persist_changes:
|
||||
slug = config_entry.get("id") or slugify_label(label)
|
||||
speakers_payload[slug] = {
|
||||
"id": slug,
|
||||
"label": label,
|
||||
"gender": config_entry.get("gender", "unknown"),
|
||||
"voice": voice_id,
|
||||
"voice_profile": voice_profile,
|
||||
"voice_formula": voice_formula,
|
||||
"resolved_voice": roster_entry.get("resolved_voice", resolved_voice or voice_id),
|
||||
"languages": usable_languages,
|
||||
}
|
||||
|
||||
new_config = new_config_payload if (persist_changes and config_changed) else None
|
||||
return updated_roster, allowed_languages, new_config
|
||||
|
||||
|
||||
def prepare_speaker_metadata(
|
||||
*,
|
||||
chapters: List[Dict[str, Any]],
|
||||
chunks: List[Dict[str, Any]],
|
||||
analysis_chunks: Optional[List[Dict[str, Any]]] = None,
|
||||
voice: str,
|
||||
voice_profile: Optional[str],
|
||||
threshold: int,
|
||||
existing_roster: Optional[Mapping[str, Any]] = None,
|
||||
run_analysis: bool = True,
|
||||
speaker_config: Optional[Mapping[str, Any]] = None,
|
||||
apply_config: bool = False,
|
||||
persist_config: bool = False,
|
||||
inject_recommended: Optional[Any] = None,
|
||||
) -> tuple[List[Dict[str, Any]], Dict[str, Any], Dict[str, Any], List[str], Optional[Dict[str, Any]]]:
|
||||
chunk_list = [dict(chunk) for chunk in chunks]
|
||||
analysis_source = [dict(chunk) for chunk in (analysis_chunks or chunks)]
|
||||
threshold_value = max(1, int(threshold))
|
||||
analysis_enabled = run_analysis
|
||||
settings_state = load_settings()
|
||||
global_random_languages = [
|
||||
code
|
||||
for code in settings_state.get("speaker_random_languages", [])
|
||||
if isinstance(code, str) and code
|
||||
]
|
||||
|
||||
if not analysis_enabled:
|
||||
for chunk in chunk_list:
|
||||
chunk["speaker_id"] = "narrator"
|
||||
chunk["speaker_label"] = "Narrator"
|
||||
analysis_payload = {
|
||||
"version": "1.0",
|
||||
"narrator": "narrator",
|
||||
"assignments": {str(chunk.get("id")): "narrator" for chunk in chunk_list},
|
||||
"speakers": {
|
||||
"narrator": {
|
||||
"id": "narrator",
|
||||
"label": "Narrator",
|
||||
"count": len(chunk_list),
|
||||
"confidence": "low",
|
||||
"sample_quotes": [],
|
||||
"suppressed": False,
|
||||
}
|
||||
},
|
||||
"suppressed": [],
|
||||
"stats": {
|
||||
"total_chunks": len(chunk_list),
|
||||
"explicit_chunks": 0,
|
||||
"active_speakers": 0,
|
||||
"unique_speakers": 1,
|
||||
"suppressed": 0,
|
||||
},
|
||||
}
|
||||
roster = build_narrator_roster(voice, voice_profile, existing_roster)
|
||||
narrator_pron = roster["narrator"].get("pronunciation")
|
||||
if narrator_pron:
|
||||
analysis_payload["speakers"]["narrator"]["pronunciation"] = narrator_pron
|
||||
return chunk_list, roster, analysis_payload, [], None
|
||||
|
||||
analysis_result = analyze_speakers(
|
||||
chapters,
|
||||
analysis_source,
|
||||
threshold=threshold_value,
|
||||
max_speakers=0,
|
||||
)
|
||||
analysis_payload = analysis_result.to_dict()
|
||||
speakers_payload = analysis_payload.get("speakers", {})
|
||||
ordered_ids = [
|
||||
sid
|
||||
for sid, meta in sorted(
|
||||
(
|
||||
(sid, meta)
|
||||
for sid, meta in speakers_payload.items()
|
||||
if sid != "narrator" and isinstance(meta, Mapping) and not meta.get("suppressed")
|
||||
),
|
||||
key=lambda item: item[1].get("count", 0),
|
||||
reverse=True,
|
||||
)
|
||||
]
|
||||
analysis_payload["ordered_speakers"] = ordered_ids
|
||||
assignments = analysis_payload.get("assignments", {})
|
||||
suppressed_ids = analysis_payload.get("suppressed", [])
|
||||
suppressed_details: List[Dict[str, Any]] = []
|
||||
speakers_payload = analysis_payload.get("speakers", {})
|
||||
if isinstance(suppressed_ids, Iterable):
|
||||
for suppressed_id in suppressed_ids:
|
||||
speaker_meta = speakers_payload.get(suppressed_id) if isinstance(speakers_payload, dict) else None
|
||||
if isinstance(speaker_meta, dict):
|
||||
suppressed_details.append(
|
||||
{
|
||||
"id": suppressed_id,
|
||||
"label": speaker_meta.get("label")
|
||||
or str(suppressed_id).replace("_", " ").title(),
|
||||
"pronunciation": speaker_meta.get("pronunciation"),
|
||||
}
|
||||
)
|
||||
else:
|
||||
suppressed_details.append(
|
||||
{
|
||||
"id": suppressed_id,
|
||||
"label": str(suppressed_id).replace("_", " ").title(),
|
||||
"pronunciation": None,
|
||||
}
|
||||
)
|
||||
analysis_payload["suppressed_details"] = suppressed_details
|
||||
roster = build_speaker_roster(
|
||||
analysis_payload,
|
||||
voice,
|
||||
voice_profile,
|
||||
existing=existing_roster,
|
||||
order=analysis_payload.get("ordered_speakers"),
|
||||
)
|
||||
applied_languages: List[str] = []
|
||||
updated_config: Optional[Dict[str, Any]] = None
|
||||
if apply_config and speaker_config:
|
||||
roster, applied_languages, updated_config = apply_speaker_config_to_roster(
|
||||
roster,
|
||||
speaker_config,
|
||||
persist_changes=persist_config,
|
||||
fallback_languages=global_random_languages,
|
||||
)
|
||||
speakers_payload = analysis_payload.get("speakers")
|
||||
if isinstance(speakers_payload, dict):
|
||||
for roster_id, roster_payload in roster.items():
|
||||
speaker_meta = speakers_payload.get(roster_id)
|
||||
if isinstance(speaker_meta, dict):
|
||||
for key in ("voice", "voice_profile", "voice_formula", "resolved_voice"):
|
||||
value = roster_payload.get(key)
|
||||
if value:
|
||||
speaker_meta[key] = value
|
||||
effective_languages: List[str] = []
|
||||
if applied_languages:
|
||||
effective_languages = applied_languages
|
||||
elif isinstance(analysis_payload.get("config_languages"), list):
|
||||
effective_languages = [
|
||||
code for code in analysis_payload.get("config_languages", []) if isinstance(code, str) and code
|
||||
]
|
||||
elif global_random_languages:
|
||||
effective_languages = list(global_random_languages)
|
||||
|
||||
if effective_languages:
|
||||
analysis_payload["config_languages"] = effective_languages
|
||||
speakers_payload = analysis_payload.get("speakers")
|
||||
if isinstance(speakers_payload, dict):
|
||||
for roster_id, roster_payload in roster.items():
|
||||
if roster_id in speakers_payload and isinstance(roster_payload, dict):
|
||||
pronunciation_value = roster_payload.get("pronunciation")
|
||||
if pronunciation_value:
|
||||
speakers_payload[roster_id]["pronunciation"] = pronunciation_value
|
||||
|
||||
fallback_languages = effective_languages or []
|
||||
if callable(inject_recommended):
|
||||
inject_recommended(roster, fallback_languages=fallback_languages)
|
||||
|
||||
for chunk in chunk_list:
|
||||
chunk_id = str(chunk.get("id"))
|
||||
speaker_id = assignments.get(chunk_id, "narrator")
|
||||
chunk["speaker_id"] = speaker_id
|
||||
speaker_meta = roster.get(speaker_id)
|
||||
chunk["speaker_label"] = speaker_meta.get("label") if isinstance(speaker_meta, dict) else speaker_id
|
||||
|
||||
return chunk_list, roster, analysis_payload, applied_languages, updated_config
|
||||
@@ -26,10 +26,11 @@ from abogen.webui.routes.utils.settings import (
|
||||
audiobookshelf_manual_available,
|
||||
)
|
||||
from abogen.webui.routes.utils.voice import (
|
||||
inject_recommended_voices,
|
||||
parse_voice_formula,
|
||||
prepare_speaker_metadata,
|
||||
template_options,
|
||||
)
|
||||
from abogen.domain.speaker_metadata import prepare_speaker_metadata
|
||||
from abogen.domain.voice_resolution import (
|
||||
formula_from_profile,
|
||||
resolve_voice_setting,
|
||||
@@ -782,6 +783,7 @@ def build_pending_job_from_extraction(
|
||||
run_analysis=initial_analysis,
|
||||
speaker_config=speaker_config_payload,
|
||||
apply_config=bool(speaker_config_payload),
|
||||
inject_recommended=inject_recommended_voices,
|
||||
)
|
||||
|
||||
normalization_overrides = {}
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple, cast
|
||||
|
||||
from abogen.speaker_configs import slugify_label
|
||||
from abogen.speaker_analysis import analyze_speakers
|
||||
from abogen.webui.routes.utils.settings import load_settings, settings_defaults, _DEFAULT_ANALYSIS_THRESHOLD, _CHUNK_LEVEL_OPTIONS, _APOSTROPHE_MODE_OPTIONS, _NORMALIZATION_GROUPS
|
||||
from abogen.voice_profiles import (
|
||||
load_profiles,
|
||||
@@ -21,207 +20,6 @@ from abogen.domain.voice_resolution import formula_from_profile
|
||||
from abogen.domain.voice_catalog import build_voice_catalog, filter_voice_catalog
|
||||
|
||||
|
||||
def build_narrator_roster(
|
||||
voice: str,
|
||||
voice_profile: Optional[str],
|
||||
existing: Optional[Mapping[str, Any]] = None,
|
||||
) -> Dict[str, Any]:
|
||||
roster: Dict[str, Any] = {
|
||||
"narrator": {
|
||||
"id": "narrator",
|
||||
"label": "Narrator",
|
||||
"voice": voice,
|
||||
}
|
||||
}
|
||||
if voice_profile:
|
||||
roster["narrator"]["voice_profile"] = voice_profile
|
||||
existing_entry: Optional[Mapping[str, Any]] = None
|
||||
if existing is not None:
|
||||
existing_entry = existing.get("narrator") if isinstance(existing, Mapping) else None
|
||||
if isinstance(existing_entry, Mapping):
|
||||
roster_entry = roster["narrator"]
|
||||
for key in ("label", "voice", "voice_profile", "voice_formula", "pronunciation"):
|
||||
value = existing_entry.get(key)
|
||||
if value is not None and value != "":
|
||||
roster_entry[key] = value
|
||||
return roster
|
||||
|
||||
|
||||
def build_speaker_roster(
|
||||
analysis: Dict[str, Any],
|
||||
base_voice: str,
|
||||
voice_profile: Optional[str],
|
||||
existing: Optional[Mapping[str, Any]] = None,
|
||||
order: Optional[Iterable[str]] = None,
|
||||
) -> Dict[str, Any]:
|
||||
roster = build_narrator_roster(base_voice, voice_profile, existing)
|
||||
existing_map: Dict[str, Any] = dict(existing) if isinstance(existing, Mapping) else {}
|
||||
speakers = analysis.get("speakers", {}) if isinstance(analysis, dict) else {}
|
||||
ordered_ids: Iterable[str]
|
||||
if order is not None:
|
||||
ordered_ids = [sid for sid in order if sid in speakers]
|
||||
else:
|
||||
ordered_ids = speakers.keys()
|
||||
|
||||
for speaker_id in ordered_ids:
|
||||
payload = speakers.get(speaker_id, {})
|
||||
if speaker_id == "narrator":
|
||||
continue
|
||||
if isinstance(payload, Mapping) and payload.get("suppressed"):
|
||||
continue
|
||||
previous = existing_map.get(speaker_id)
|
||||
roster[speaker_id] = {
|
||||
"id": speaker_id,
|
||||
"label": payload.get("label") or speaker_id.replace("_", " ").title(),
|
||||
"analysis_confidence": payload.get("confidence"),
|
||||
"analysis_count": payload.get("count"),
|
||||
"gender": payload.get("gender", "unknown"),
|
||||
}
|
||||
detected_gender = payload.get("detected_gender")
|
||||
if detected_gender:
|
||||
roster[speaker_id]["detected_gender"] = detected_gender
|
||||
samples = payload.get("sample_quotes")
|
||||
if isinstance(samples, list):
|
||||
roster[speaker_id]["sample_quotes"] = samples
|
||||
if isinstance(previous, Mapping):
|
||||
for key in ("voice", "voice_profile", "voice_formula", "resolved_voice", "pronunciation"):
|
||||
value = previous.get(key)
|
||||
if value is not None and value != "":
|
||||
roster[speaker_id][key] = value
|
||||
if "sample_quotes" not in roster[speaker_id]:
|
||||
prev_samples = previous.get("sample_quotes")
|
||||
if isinstance(prev_samples, list):
|
||||
roster[speaker_id]["sample_quotes"] = prev_samples
|
||||
if "detected_gender" not in roster[speaker_id]:
|
||||
prev_detected = previous.get("detected_gender")
|
||||
if isinstance(prev_detected, str) and prev_detected:
|
||||
roster[speaker_id]["detected_gender"] = prev_detected
|
||||
return roster
|
||||
|
||||
|
||||
def match_configured_speaker(
|
||||
config_speakers: Mapping[str, Any],
|
||||
roster_id: str,
|
||||
roster_label: str,
|
||||
) -> Optional[Mapping[str, Any]]:
|
||||
if not config_speakers:
|
||||
return None
|
||||
entry = config_speakers.get(roster_id)
|
||||
if entry:
|
||||
return cast(Mapping[str, Any], entry)
|
||||
slug = slugify_label(roster_label)
|
||||
if slug != roster_id and slug in config_speakers:
|
||||
return cast(Mapping[str, Any], config_speakers[slug])
|
||||
lower_label = roster_label.strip().lower()
|
||||
for record in config_speakers.values():
|
||||
if not isinstance(record, Mapping):
|
||||
continue
|
||||
if str(record.get("label", "")).strip().lower() == lower_label:
|
||||
return record
|
||||
return None
|
||||
|
||||
|
||||
def apply_speaker_config_to_roster(
|
||||
roster: Mapping[str, Any],
|
||||
config: Optional[Mapping[str, Any]],
|
||||
*,
|
||||
persist_changes: bool = False,
|
||||
fallback_languages: Optional[Iterable[str]] = None,
|
||||
) -> Tuple[Dict[str, Any], List[str], Optional[Dict[str, Any]]]:
|
||||
if not isinstance(roster, Mapping):
|
||||
effective_languages = [code for code in (fallback_languages or []) if isinstance(code, str) and code]
|
||||
return {}, effective_languages, None
|
||||
updated_roster: Dict[str, Any] = {key: dict(value) for key, value in roster.items() if isinstance(value, Mapping)}
|
||||
if not config:
|
||||
effective_languages = [code for code in (fallback_languages or []) if isinstance(code, str) and code]
|
||||
return updated_roster, effective_languages, None
|
||||
|
||||
speakers_map = config.get("speakers")
|
||||
if not isinstance(speakers_map, Mapping):
|
||||
effective_languages = [code for code in (fallback_languages or []) if isinstance(code, str) and code]
|
||||
return updated_roster, effective_languages, None
|
||||
|
||||
config_languages = config.get("languages")
|
||||
if isinstance(config_languages, list):
|
||||
allowed_languages = [code for code in config_languages if isinstance(code, str) and code]
|
||||
else:
|
||||
allowed_languages = []
|
||||
if not allowed_languages and fallback_languages:
|
||||
allowed_languages = [code for code in fallback_languages if isinstance(code, str) and code]
|
||||
|
||||
default_voice = config.get("default_voice") if isinstance(config.get("default_voice"), str) else ""
|
||||
used_voices = {entry.get("resolved_voice") or entry.get("voice") for entry in updated_roster.values()} - {None}
|
||||
narrator_voice = ""
|
||||
narrator_entry = updated_roster.get("narrator") if isinstance(updated_roster, Mapping) else None
|
||||
if isinstance(narrator_entry, Mapping):
|
||||
narrator_voice = str(
|
||||
narrator_entry.get("resolved_voice")
|
||||
or narrator_entry.get("default_voice")
|
||||
or ""
|
||||
).strip()
|
||||
if narrator_voice:
|
||||
used_voices.add(narrator_voice)
|
||||
|
||||
config_changed = False
|
||||
new_config_payload: Dict[str, Any] = {
|
||||
"language": config.get("language", "a"),
|
||||
"languages": allowed_languages,
|
||||
"default_voice": default_voice,
|
||||
"speakers": dict(speakers_map),
|
||||
"version": config.get("version", 1),
|
||||
"notes": config.get("notes", ""),
|
||||
}
|
||||
|
||||
speakers_payload = new_config_payload["speakers"]
|
||||
|
||||
for speaker_id, roster_entry in updated_roster.items():
|
||||
if speaker_id == "narrator":
|
||||
continue
|
||||
label = str(roster_entry.get("label") or speaker_id)
|
||||
config_entry = match_configured_speaker(speakers_map, speaker_id, label)
|
||||
if config_entry is None:
|
||||
continue
|
||||
voice_id = str(config_entry.get("voice") or "").strip()
|
||||
voice_profile = str(config_entry.get("voice_profile") or "").strip()
|
||||
voice_formula = str(config_entry.get("voice_formula") or "").strip()
|
||||
resolved_voice = str(config_entry.get("resolved_voice") or "").strip()
|
||||
languages = config_entry.get("languages") if isinstance(config_entry.get("languages"), list) else []
|
||||
chosen_voice = resolved_voice or voice_formula or voice_id or roster_entry.get("voice")
|
||||
usable_languages = languages or allowed_languages
|
||||
|
||||
if chosen_voice:
|
||||
roster_entry["resolved_voice"] = chosen_voice
|
||||
roster_entry["voice"] = chosen_voice if not voice_profile and not voice_formula else roster_entry.get("voice", chosen_voice)
|
||||
if voice_profile:
|
||||
roster_entry["voice_profile"] = voice_profile
|
||||
if voice_formula:
|
||||
roster_entry["voice_formula"] = voice_formula
|
||||
roster_entry["resolved_voice"] = voice_formula
|
||||
if not voice_formula and not voice_profile and resolved_voice:
|
||||
roster_entry["resolved_voice"] = resolved_voice
|
||||
roster_entry["config_languages"] = usable_languages or []
|
||||
|
||||
if chosen_voice:
|
||||
used_voices.add(chosen_voice)
|
||||
|
||||
# persist updates back to config payload if required
|
||||
if persist_changes:
|
||||
slug = config_entry.get("id") or slugify_label(label)
|
||||
speakers_payload[slug] = {
|
||||
"id": slug,
|
||||
"label": label,
|
||||
"gender": config_entry.get("gender", "unknown"),
|
||||
"voice": voice_id,
|
||||
"voice_profile": voice_profile,
|
||||
"voice_formula": voice_formula,
|
||||
"resolved_voice": roster_entry.get("resolved_voice", resolved_voice or voice_id),
|
||||
"languages": usable_languages,
|
||||
}
|
||||
|
||||
new_config = new_config_payload if (persist_changes and config_changed) else None
|
||||
return updated_roster, allowed_languages, new_config
|
||||
|
||||
|
||||
def inject_recommended_voices(
|
||||
roster: Mapping[str, Any],
|
||||
*,
|
||||
@@ -311,168 +109,6 @@ def extract_speaker_config_form(form: Mapping[str, Any]) -> Tuple[str, Dict[str,
|
||||
return name, payload, errors
|
||||
|
||||
|
||||
def prepare_speaker_metadata(
|
||||
*,
|
||||
chapters: List[Dict[str, Any]],
|
||||
chunks: List[Dict[str, Any]],
|
||||
analysis_chunks: Optional[List[Dict[str, Any]]] = None,
|
||||
voice: str,
|
||||
voice_profile: Optional[str],
|
||||
threshold: int,
|
||||
existing_roster: Optional[Mapping[str, Any]] = None,
|
||||
run_analysis: bool = True,
|
||||
speaker_config: Optional[Mapping[str, Any]] = None,
|
||||
apply_config: bool = False,
|
||||
persist_config: bool = False,
|
||||
) -> tuple[List[Dict[str, Any]], Dict[str, Any], Dict[str, Any], List[str], Optional[Dict[str, Any]]]:
|
||||
chunk_list = [dict(chunk) for chunk in chunks]
|
||||
analysis_source = [dict(chunk) for chunk in (analysis_chunks or chunks)]
|
||||
threshold_value = max(1, int(threshold))
|
||||
analysis_enabled = run_analysis
|
||||
settings_state = load_settings()
|
||||
global_random_languages = [
|
||||
code
|
||||
for code in settings_state.get("speaker_random_languages", [])
|
||||
if isinstance(code, str) and code
|
||||
]
|
||||
|
||||
if not analysis_enabled:
|
||||
for chunk in chunk_list:
|
||||
chunk["speaker_id"] = "narrator"
|
||||
chunk["speaker_label"] = "Narrator"
|
||||
analysis_payload = {
|
||||
"version": "1.0",
|
||||
"narrator": "narrator",
|
||||
"assignments": {str(chunk.get("id")): "narrator" for chunk in chunk_list},
|
||||
"speakers": {
|
||||
"narrator": {
|
||||
"id": "narrator",
|
||||
"label": "Narrator",
|
||||
"count": len(chunk_list),
|
||||
"confidence": "low",
|
||||
"sample_quotes": [],
|
||||
"suppressed": False,
|
||||
}
|
||||
},
|
||||
"suppressed": [],
|
||||
"stats": {
|
||||
"total_chunks": len(chunk_list),
|
||||
"explicit_chunks": 0,
|
||||
"active_speakers": 0,
|
||||
"unique_speakers": 1,
|
||||
"suppressed": 0,
|
||||
},
|
||||
}
|
||||
roster = build_narrator_roster(voice, voice_profile, existing_roster)
|
||||
narrator_pron = roster["narrator"].get("pronunciation")
|
||||
if narrator_pron:
|
||||
analysis_payload["speakers"]["narrator"]["pronunciation"] = narrator_pron
|
||||
return chunk_list, roster, analysis_payload, [], None
|
||||
|
||||
analysis_result = analyze_speakers(
|
||||
chapters,
|
||||
analysis_source,
|
||||
threshold=threshold_value,
|
||||
max_speakers=0,
|
||||
)
|
||||
analysis_payload = analysis_result.to_dict()
|
||||
speakers_payload = analysis_payload.get("speakers", {})
|
||||
ordered_ids = [
|
||||
sid
|
||||
for sid, meta in sorted(
|
||||
(
|
||||
(sid, meta)
|
||||
for sid, meta in speakers_payload.items()
|
||||
if sid != "narrator" and isinstance(meta, Mapping) and not meta.get("suppressed")
|
||||
),
|
||||
key=lambda item: item[1].get("count", 0),
|
||||
reverse=True,
|
||||
)
|
||||
]
|
||||
analysis_payload["ordered_speakers"] = ordered_ids
|
||||
assignments = analysis_payload.get("assignments", {})
|
||||
suppressed_ids = analysis_payload.get("suppressed", [])
|
||||
suppressed_details: List[Dict[str, Any]] = []
|
||||
speakers_payload = analysis_payload.get("speakers", {})
|
||||
if isinstance(suppressed_ids, Iterable):
|
||||
for suppressed_id in suppressed_ids:
|
||||
speaker_meta = speakers_payload.get(suppressed_id) if isinstance(speakers_payload, dict) else None
|
||||
if isinstance(speaker_meta, dict):
|
||||
suppressed_details.append(
|
||||
{
|
||||
"id": suppressed_id,
|
||||
"label": speaker_meta.get("label")
|
||||
or str(suppressed_id).replace("_", " ").title(),
|
||||
"pronunciation": speaker_meta.get("pronunciation"),
|
||||
}
|
||||
)
|
||||
else:
|
||||
suppressed_details.append(
|
||||
{
|
||||
"id": suppressed_id,
|
||||
"label": str(suppressed_id).replace("_", " ").title(),
|
||||
"pronunciation": None,
|
||||
}
|
||||
)
|
||||
analysis_payload["suppressed_details"] = suppressed_details
|
||||
roster = build_speaker_roster(
|
||||
analysis_payload,
|
||||
voice,
|
||||
voice_profile,
|
||||
existing=existing_roster,
|
||||
order=analysis_payload.get("ordered_speakers"),
|
||||
)
|
||||
applied_languages: List[str] = []
|
||||
updated_config: Optional[Dict[str, Any]] = None
|
||||
if apply_config and speaker_config:
|
||||
roster, applied_languages, updated_config = apply_speaker_config_to_roster(
|
||||
roster,
|
||||
speaker_config,
|
||||
persist_changes=persist_config,
|
||||
fallback_languages=global_random_languages,
|
||||
)
|
||||
speakers_payload = analysis_payload.get("speakers")
|
||||
if isinstance(speakers_payload, dict):
|
||||
for roster_id, roster_payload in roster.items():
|
||||
speaker_meta = speakers_payload.get(roster_id)
|
||||
if isinstance(speaker_meta, dict):
|
||||
for key in ("voice", "voice_profile", "voice_formula", "resolved_voice"):
|
||||
value = roster_payload.get(key)
|
||||
if value:
|
||||
speaker_meta[key] = value
|
||||
effective_languages: List[str] = []
|
||||
if applied_languages:
|
||||
effective_languages = applied_languages
|
||||
elif isinstance(analysis_payload.get("config_languages"), list):
|
||||
effective_languages = [
|
||||
code for code in analysis_payload.get("config_languages", []) if isinstance(code, str) and code
|
||||
]
|
||||
elif global_random_languages:
|
||||
effective_languages = list(global_random_languages)
|
||||
|
||||
if effective_languages:
|
||||
analysis_payload["config_languages"] = effective_languages
|
||||
speakers_payload = analysis_payload.get("speakers")
|
||||
if isinstance(speakers_payload, dict):
|
||||
for roster_id, roster_payload in roster.items():
|
||||
if roster_id in speakers_payload and isinstance(roster_payload, dict):
|
||||
pronunciation_value = roster_payload.get("pronunciation")
|
||||
if pronunciation_value:
|
||||
speakers_payload[roster_id]["pronunciation"] = pronunciation_value
|
||||
|
||||
fallback_languages = effective_languages or []
|
||||
inject_recommended_voices(roster, fallback_languages=fallback_languages)
|
||||
|
||||
for chunk in chunk_list:
|
||||
chunk_id = str(chunk.get("id"))
|
||||
speaker_id = assignments.get(chunk_id, "narrator")
|
||||
chunk["speaker_id"] = speaker_id
|
||||
speaker_meta = roster.get(speaker_id)
|
||||
chunk["speaker_label"] = speaker_meta.get("label") if isinstance(speaker_meta, dict) else speaker_id
|
||||
|
||||
return chunk_list, roster, analysis_payload, applied_languages, updated_config
|
||||
|
||||
|
||||
def template_options() -> Dict[str, Any]:
|
||||
current_settings = load_settings()
|
||||
profiles = serialize_profiles()
|
||||
|
||||
@@ -0,0 +1,758 @@
|
||||
"""Tests for domain speaker metadata functions.
|
||||
|
||||
Tests for build_narrator_roster, build_speaker_roster, match_configured_speaker,
|
||||
apply_speaker_config_to_roster, and prepare_speaker_metadata.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Dict
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# build_narrator_roster
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestBuildNarratorRoster:
|
||||
"""Tests for build_narrator_roster()."""
|
||||
|
||||
def test_basic_roster(self):
|
||||
from abogen.domain.speaker_metadata import build_narrator_roster
|
||||
|
||||
roster = build_narrator_roster("af_heart", None)
|
||||
assert "narrator" in roster
|
||||
assert roster["narrator"]["voice"] == "af_heart"
|
||||
assert roster["narrator"]["label"] == "Narrator"
|
||||
|
||||
def test_with_voice_profile(self):
|
||||
from abogen.domain.speaker_metadata import build_narrator_roster
|
||||
|
||||
roster = build_narrator_roster("af_heart", "my_profile")
|
||||
assert roster["narrator"]["voice_profile"] == "my_profile"
|
||||
|
||||
def test_without_voice_profile(self):
|
||||
from abogen.domain.speaker_metadata import build_narrator_roster
|
||||
|
||||
roster = build_narrator_roster("af_heart", None)
|
||||
assert "voice_profile" not in roster["narrator"]
|
||||
|
||||
def test_merges_existing_overrides(self):
|
||||
from abogen.domain.speaker_metadata import build_narrator_roster
|
||||
|
||||
existing = {
|
||||
"narrator": {
|
||||
"label": "Custom Narrator",
|
||||
"voice": "am_echo",
|
||||
"pronunciation": "NAH-rah-tor",
|
||||
}
|
||||
}
|
||||
roster = build_narrator_roster("af_heart", None, existing=existing)
|
||||
assert roster["narrator"]["label"] == "Custom Narrator"
|
||||
assert roster["narrator"]["voice"] == "am_echo"
|
||||
assert roster["narrator"]["pronunciation"] == "NAH-rah-tor"
|
||||
|
||||
def test_existing_none_ignored(self):
|
||||
from abogen.domain.speaker_metadata import build_narrator_roster
|
||||
|
||||
roster = build_narrator_roster("af_heart", None, existing=None)
|
||||
assert roster["narrator"]["voice"] == "af_heart"
|
||||
|
||||
def test_empty_existing_dict(self):
|
||||
from abogen.domain.speaker_metadata import build_narrator_roster
|
||||
|
||||
roster = build_narrator_roster("af_heart", None, existing={})
|
||||
assert roster["narrator"]["voice"] == "af_heart"
|
||||
|
||||
def test_existing_without_narrator_key(self):
|
||||
from abogen.domain.speaker_metadata import build_narrator_roster
|
||||
|
||||
existing = {"other_speaker": {"label": "Other"}}
|
||||
roster = build_narrator_roster("af_heart", None, existing=existing)
|
||||
assert roster["narrator"]["voice"] == "af_heart"
|
||||
|
||||
def test_empty_string_values_not_overridden(self):
|
||||
from abogen.domain.speaker_metadata import build_narrator_roster
|
||||
|
||||
existing = {"narrator": {"label": "", "voice": ""}}
|
||||
roster = build_narrator_roster("af_heart", None, existing=existing)
|
||||
assert roster["narrator"]["label"] == "Narrator"
|
||||
assert roster["narrator"]["voice"] == "af_heart"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# build_speaker_roster
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestBuildSpeakerRoster:
|
||||
"""Tests for build_speaker_roster()."""
|
||||
|
||||
def test_single_narrator(self):
|
||||
from abogen.domain.speaker_metadata import build_speaker_roster
|
||||
|
||||
analysis = {"speakers": {"narrator": {"label": "Narrator", "count": 10}}}
|
||||
roster = build_speaker_roster(analysis, "af_heart", None)
|
||||
assert list(roster.keys()) == ["narrator"]
|
||||
assert roster["narrator"]["voice"] == "af_heart"
|
||||
|
||||
def test_multiple_speakers(self):
|
||||
from abogen.domain.speaker_metadata import build_speaker_roster
|
||||
|
||||
analysis = {
|
||||
"speakers": {
|
||||
"narrator": {"label": "Narrator", "count": 10},
|
||||
"alice": {"label": "Alice", "count": 5, "gender": "female"},
|
||||
"bob": {"label": "Bob", "count": 3, "gender": "male"},
|
||||
}
|
||||
}
|
||||
roster = build_speaker_roster(analysis, "af_heart", None)
|
||||
assert "narrator" in roster
|
||||
assert "alice" in roster
|
||||
assert "bob" in roster
|
||||
assert roster["alice"]["label"] == "Alice"
|
||||
assert roster["alice"]["gender"] == "female"
|
||||
|
||||
def test_suppressed_speakers_excluded(self):
|
||||
from abogen.domain.speaker_metadata import build_speaker_roster
|
||||
|
||||
analysis = {
|
||||
"speakers": {
|
||||
"narrator": {"label": "Narrator", "count": 10},
|
||||
"alice": {"label": "Alice", "count": 5},
|
||||
"bob": {"label": "Bob", "count": 1, "suppressed": True},
|
||||
}
|
||||
}
|
||||
roster = build_speaker_roster(analysis, "af_heart", None)
|
||||
assert "bob" not in roster
|
||||
|
||||
def test_order_respected(self):
|
||||
from abogen.domain.speaker_metadata import build_speaker_roster
|
||||
|
||||
analysis = {
|
||||
"speakers": {
|
||||
"narrator": {"label": "Narrator", "count": 10},
|
||||
"alice": {"label": "Alice", "count": 5},
|
||||
"bob": {"label": "Bob", "count": 3},
|
||||
}
|
||||
}
|
||||
roster = build_speaker_roster(analysis, "af_heart", None, order=["bob", "alice"])
|
||||
keys = list(roster.keys())
|
||||
assert keys.index("bob") < keys.index("alice")
|
||||
|
||||
def test_existing_assignments_preserved(self):
|
||||
from abogen.domain.speaker_metadata import build_speaker_roster
|
||||
|
||||
analysis = {
|
||||
"speakers": {
|
||||
"narrator": {"label": "Narrator", "count": 10},
|
||||
"alice": {"label": "Alice", "count": 5, "gender": "female"},
|
||||
}
|
||||
}
|
||||
existing = {
|
||||
"narrator": {"voice": "am_echo"},
|
||||
"alice": {"voice": "af_nicole", "pronunciation": "AH-leece"},
|
||||
}
|
||||
roster = build_speaker_roster(analysis, "af_heart", None, existing=existing)
|
||||
assert roster["alice"]["voice"] == "af_nicole"
|
||||
assert roster["alice"]["pronunciation"] == "AH-leece"
|
||||
assert roster["narrator"]["voice"] == "am_echo"
|
||||
|
||||
def test_empty_analysis(self):
|
||||
from abogen.domain.speaker_metadata import build_speaker_roster
|
||||
|
||||
roster = build_speaker_roster({}, "af_heart", None)
|
||||
assert "narrator" in roster
|
||||
assert len(roster) == 1
|
||||
|
||||
def test_sample_quotes_preserved(self):
|
||||
from abogen.domain.speaker_metadata import build_speaker_roster
|
||||
|
||||
analysis = {
|
||||
"speakers": {
|
||||
"narrator": {"label": "Narrator", "count": 10},
|
||||
"alice": {
|
||||
"label": "Alice",
|
||||
"count": 5,
|
||||
"sample_quotes": ["Hello!", "Goodbye!"],
|
||||
},
|
||||
}
|
||||
}
|
||||
roster = build_speaker_roster(analysis, "af_heart", None)
|
||||
assert roster["alice"]["sample_quotes"] == ["Hello!", "Goodbye!"]
|
||||
|
||||
def test_detected_gender_preserved(self):
|
||||
from abogen.domain.speaker_metadata import build_speaker_roster
|
||||
|
||||
analysis = {
|
||||
"speakers": {
|
||||
"narrator": {"label": "Narrator", "count": 10},
|
||||
"alice": {"label": "Alice", "count": 5, "detected_gender": "female"},
|
||||
}
|
||||
}
|
||||
roster = build_speaker_roster(analysis, "af_heart", None)
|
||||
assert roster["alice"]["detected_gender"] == "female"
|
||||
|
||||
def test_default_label_from_id(self):
|
||||
from abogen.domain.speaker_metadata import build_speaker_roster
|
||||
|
||||
analysis = {
|
||||
"speakers": {
|
||||
"narrator": {"label": "Narrator", "count": 10},
|
||||
"my_character": {"count": 3},
|
||||
}
|
||||
}
|
||||
roster = build_speaker_roster(analysis, "af_heart", None)
|
||||
assert roster["my_character"]["label"] == "My Character"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# match_configured_speaker
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestMatchConfiguredSpeaker:
|
||||
"""Tests for match_configured_speaker()."""
|
||||
|
||||
def test_match_by_id(self):
|
||||
from abogen.domain.speaker_metadata import match_configured_speaker
|
||||
|
||||
config = {"alice": {"id": "alice", "label": "Alice", "voice": "af_heart"}}
|
||||
result = match_configured_speaker(config, "alice", "Alice")
|
||||
assert result is not None
|
||||
assert result["voice"] == "af_heart"
|
||||
|
||||
def test_match_by_slug(self):
|
||||
from abogen.domain.speaker_metadata import match_configured_speaker
|
||||
|
||||
config = {"my_character": {"id": "my_character", "label": "My Character"}}
|
||||
result = match_configured_speaker(config, "my_character", "My Character")
|
||||
assert result is not None
|
||||
|
||||
def test_match_by_label_lowercase(self):
|
||||
from abogen.domain.speaker_metadata import match_configured_speaker
|
||||
|
||||
config = {"custom_id": {"id": "custom_id", "label": "Alice"}}
|
||||
result = match_configured_speaker(config, "other_id", "Alice")
|
||||
assert result is not None
|
||||
assert result["id"] == "custom_id"
|
||||
|
||||
def test_no_match(self):
|
||||
from abogen.domain.speaker_metadata import match_configured_speaker
|
||||
|
||||
config = {"alice": {"id": "alice", "label": "Alice"}}
|
||||
result = match_configured_speaker(config, "bob", "Bob")
|
||||
assert result is None
|
||||
|
||||
def test_empty_config(self):
|
||||
from abogen.domain.speaker_metadata import match_configured_speaker
|
||||
|
||||
result = match_configured_speaker({}, "alice", "Alice")
|
||||
assert result is None
|
||||
|
||||
def test_none_config(self):
|
||||
from abogen.domain.speaker_metadata import match_configured_speaker
|
||||
|
||||
result = match_configured_speaker(None, "alice", "Alice") # type: ignore
|
||||
assert result is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# apply_speaker_config_to_roster
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestApplySpeakerConfigToRoster:
|
||||
"""Tests for apply_speaker_config_to_roster()."""
|
||||
|
||||
def test_no_config_returns_roster_unchanged(self):
|
||||
from abogen.domain.speaker_metadata import apply_speaker_config_to_roster
|
||||
|
||||
roster = {"narrator": {"id": "narrator", "voice": "af_heart"}}
|
||||
result, languages, config = apply_speaker_config_to_roster(roster, None)
|
||||
assert result["narrator"]["voice"] == "af_heart"
|
||||
assert languages == []
|
||||
assert config is None
|
||||
|
||||
def test_empty_config_returns_roster_unchanged(self):
|
||||
from abogen.domain.speaker_metadata import apply_speaker_config_to_roster
|
||||
|
||||
roster = {"narrator": {"id": "narrator", "voice": "af_heart"}}
|
||||
result, languages, config = apply_speaker_config_to_roster(roster, {})
|
||||
assert result["narrator"]["voice"] == "af_heart"
|
||||
|
||||
def test_config_without_speakers_map(self):
|
||||
from abogen.domain.speaker_metadata import apply_speaker_config_to_roster
|
||||
|
||||
roster = {"narrator": {"id": "narrator", "voice": "af_heart"}}
|
||||
config = {"language": "a"}
|
||||
result, languages, config = apply_speaker_config_to_roster(roster, config)
|
||||
assert result["narrator"]["voice"] == "af_heart"
|
||||
|
||||
def test_applies_voice_from_config(self):
|
||||
from abogen.domain.speaker_metadata import apply_speaker_config_to_roster
|
||||
|
||||
roster = {
|
||||
"narrator": {"id": "narrator", "voice": "af_heart"},
|
||||
"alice": {"id": "alice", "label": "Alice", "voice": "af_heart"},
|
||||
}
|
||||
config = {
|
||||
"speakers": {
|
||||
"alice": {"id": "alice", "voice": "af_nicole", "gender": "female"}
|
||||
},
|
||||
"languages": ["a"],
|
||||
}
|
||||
result, languages, updated_config = apply_speaker_config_to_roster(
|
||||
roster, config, persist_changes=True
|
||||
)
|
||||
assert result["alice"]["voice"] == "af_nicole"
|
||||
assert result["alice"]["resolved_voice"] == "af_nicole"
|
||||
|
||||
def test_applies_voice_profile(self):
|
||||
from abogen.domain.speaker_metadata import apply_speaker_config_to_roster
|
||||
|
||||
roster = {
|
||||
"narrator": {"id": "narrator", "voice": "af_heart"},
|
||||
"alice": {"id": "alice", "label": "Alice"},
|
||||
}
|
||||
config = {
|
||||
"speakers": {
|
||||
"alice": {"id": "alice", "voice_profile": "my_profile"}
|
||||
}
|
||||
}
|
||||
result, _, _ = apply_speaker_config_to_roster(roster, config)
|
||||
assert result["alice"]["voice_profile"] == "my_profile"
|
||||
|
||||
def test_applies_voice_formula(self):
|
||||
from abogen.domain.speaker_metadata import apply_speaker_config_to_roster
|
||||
|
||||
roster = {
|
||||
"narrator": {"id": "narrator", "voice": "af_heart"},
|
||||
"alice": {"id": "alice", "label": "Alice"},
|
||||
}
|
||||
config = {
|
||||
"speakers": {
|
||||
"alice": {"id": "alice", "voice_formula": "af_heart(0.6)+af_nicole(0.4)"}
|
||||
}
|
||||
}
|
||||
result, _, _ = apply_speaker_config_to_roster(roster, config)
|
||||
assert result["alice"]["voice_formula"] == "af_heart(0.6)+af_nicole(0.4)"
|
||||
assert result["alice"]["resolved_voice"] == "af_heart(0.6)+af_nicole(0.4)"
|
||||
|
||||
def test_persist_changes_returns_updated_config(self):
|
||||
from abogen.domain.speaker_metadata import apply_speaker_config_to_roster
|
||||
|
||||
roster = {
|
||||
"narrator": {"id": "narrator", "voice": "af_heart"},
|
||||
"alice": {"id": "alice", "label": "Alice", "voice": "af_heart"},
|
||||
}
|
||||
config = {
|
||||
"language": "a",
|
||||
"languages": ["a"],
|
||||
"speakers": {
|
||||
"alice": {"id": "alice", "voice": "af_nicole", "gender": "female"}
|
||||
},
|
||||
"version": 1,
|
||||
}
|
||||
_, _, updated_config = apply_speaker_config_to_roster(
|
||||
roster, config, persist_changes=True
|
||||
)
|
||||
# config_changed is False by default, so updated_config should be None
|
||||
# unless there's actual change logic triggered
|
||||
# The function has config_changed = False and never sets it to True
|
||||
# so updated_config should be None even with persist_changes=True
|
||||
assert updated_config is None
|
||||
|
||||
def test_fallback_languages_used(self):
|
||||
from abogen.domain.speaker_metadata import apply_speaker_config_to_roster
|
||||
|
||||
roster = {"narrator": {"id": "narrator", "voice": "af_heart"}}
|
||||
result, languages, _ = apply_speaker_config_to_roster(
|
||||
roster, None, fallback_languages=["a", "b"]
|
||||
)
|
||||
assert languages == ["a", "b"]
|
||||
|
||||
def test_config_languages_take_precedence(self):
|
||||
from abogen.domain.speaker_metadata import apply_speaker_config_to_roster
|
||||
|
||||
roster = {"narrator": {"id": "narrator", "voice": "af_heart"}}
|
||||
config = {"languages": ["a"], "speakers": {}}
|
||||
_, languages, _ = apply_speaker_config_to_roster(
|
||||
roster, config, fallback_languages=["a", "b"]
|
||||
)
|
||||
assert languages == ["a"]
|
||||
|
||||
def test_empty_roster_returns_empty(self):
|
||||
from abogen.domain.speaker_metadata import apply_speaker_config_to_roster
|
||||
|
||||
result, languages, config = apply_speaker_config_to_roster({}, None)
|
||||
assert result == {}
|
||||
assert languages == []
|
||||
|
||||
def test_non_mapping_roster_returns_empty(self):
|
||||
from abogen.domain.speaker_metadata import apply_speaker_config_to_roster
|
||||
|
||||
result, languages, config = apply_speaker_config_to_roster("invalid", None) # type: ignore
|
||||
assert result == {}
|
||||
assert languages == []
|
||||
|
||||
def test_narrator_not_modified(self):
|
||||
from abogen.domain.speaker_metadata import apply_speaker_config_to_roster
|
||||
|
||||
roster = {
|
||||
"narrator": {"id": "narrator", "voice": "af_heart"},
|
||||
"alice": {"id": "alice", "label": "Alice"},
|
||||
}
|
||||
config = {
|
||||
"speakers": {
|
||||
"narrator": {"id": "narrator", "voice": "am_echo"},
|
||||
"alice": {"id": "alice", "voice": "af_nicole"},
|
||||
}
|
||||
}
|
||||
result, _, _ = apply_speaker_config_to_roster(roster, config)
|
||||
assert result["narrator"]["voice"] == "af_heart"
|
||||
assert result["alice"]["voice"] == "af_nicole"
|
||||
|
||||
def test_config_languages_applied_to_roster_entry(self):
|
||||
from abogen.domain.speaker_metadata import apply_speaker_config_to_roster
|
||||
|
||||
roster = {
|
||||
"narrator": {"id": "narrator", "voice": "af_heart"},
|
||||
"alice": {"id": "alice", "label": "Alice"},
|
||||
}
|
||||
config = {
|
||||
"languages": ["a", "b"],
|
||||
"speakers": {
|
||||
"alice": {"id": "alice", "voice": "af_nicole"}
|
||||
},
|
||||
}
|
||||
result, _, _ = apply_speaker_config_to_roster(roster, config)
|
||||
assert result["alice"]["config_languages"] == ["a", "b"]
|
||||
|
||||
def test_speaker_specific_languages_override(self):
|
||||
from abogen.domain.speaker_metadata import apply_speaker_config_to_roster
|
||||
|
||||
roster = {
|
||||
"narrator": {"id": "narrator", "voice": "af_heart"},
|
||||
"alice": {"id": "alice", "label": "Alice"},
|
||||
}
|
||||
config = {
|
||||
"languages": ["a"],
|
||||
"speakers": {
|
||||
"alice": {"id": "alice", "voice": "af_nicole", "languages": ["a", "b"]}
|
||||
},
|
||||
}
|
||||
result, _, _ = apply_speaker_config_to_roster(roster, config)
|
||||
assert result["alice"]["config_languages"] == ["a", "b"]
|
||||
|
||||
def test_resolved_voice_takes_precedence(self):
|
||||
from abogen.domain.speaker_metadata import apply_speaker_config_to_roster
|
||||
|
||||
roster = {
|
||||
"narrator": {"id": "narrator", "voice": "af_heart"},
|
||||
"alice": {"id": "alice", "label": "Alice"},
|
||||
}
|
||||
config = {
|
||||
"speakers": {
|
||||
"alice": {
|
||||
"id": "alice",
|
||||
"voice": "af_heart",
|
||||
"resolved_voice": "af_nicole",
|
||||
}
|
||||
}
|
||||
}
|
||||
result, _, _ = apply_speaker_config_to_roster(roster, config)
|
||||
assert result["alice"]["resolved_voice"] == "af_nicole"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# prepare_speaker_metadata
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestPrepareSpeakerMetadata:
|
||||
"""Tests for prepare_speaker_metadata()."""
|
||||
|
||||
def _make_chunks(self, count=3):
|
||||
return [{"id": str(i), "text": f"Chunk {i}"} for i in range(count)]
|
||||
|
||||
def _make_chapters(self):
|
||||
return [{"title": "Chapter 1", "chunks": self._make_chunks()}]
|
||||
|
||||
@patch("abogen.domain.speaker_metadata.load_settings")
|
||||
def test_no_analysis(self, mock_settings):
|
||||
from abogen.domain.speaker_metadata import prepare_speaker_metadata
|
||||
|
||||
mock_settings.return_value = {"speaker_random_languages": []}
|
||||
chunks = self._make_chunks()
|
||||
result = prepare_speaker_metadata(
|
||||
chapters=self._make_chapters(),
|
||||
chunks=chunks,
|
||||
voice="af_heart",
|
||||
voice_profile=None,
|
||||
threshold=3,
|
||||
run_analysis=False,
|
||||
)
|
||||
chunk_list, roster, analysis, languages, config = result
|
||||
assert all(c["speaker_id"] == "narrator" for c in chunk_list)
|
||||
assert all(c["speaker_label"] == "Narrator" for c in chunk_list)
|
||||
assert "narrator" in roster
|
||||
assert languages == []
|
||||
assert config is None
|
||||
|
||||
@patch("abogen.domain.speaker_metadata.load_settings")
|
||||
def test_no_analysis_with_existing_roster(self, mock_settings):
|
||||
from abogen.domain.speaker_metadata import prepare_speaker_metadata
|
||||
|
||||
mock_settings.return_value = {"speaker_random_languages": []}
|
||||
existing = {"narrator": {"voice": "am_echo", "pronunciation": "test"}}
|
||||
result = prepare_speaker_metadata(
|
||||
chapters=self._make_chapters(),
|
||||
chunks=self._make_chunks(),
|
||||
voice="af_heart",
|
||||
voice_profile=None,
|
||||
threshold=3,
|
||||
run_analysis=False,
|
||||
existing_roster=existing,
|
||||
)
|
||||
_, roster, _, _, _ = result
|
||||
assert roster["narrator"]["voice"] == "am_echo"
|
||||
assert roster["narrator"]["pronunciation"] == "test"
|
||||
|
||||
@patch("abogen.domain.speaker_metadata.load_settings")
|
||||
@patch("abogen.domain.speaker_metadata.analyze_speakers")
|
||||
def test_with_analysis(self, mock_analyze, mock_settings):
|
||||
from abogen.domain.speaker_metadata import prepare_speaker_metadata
|
||||
|
||||
mock_settings.return_value = {"speaker_random_languages": []}
|
||||
|
||||
mock_result = MagicMock()
|
||||
mock_result.to_dict.return_value = {
|
||||
"version": "1.0",
|
||||
"narrator": "narrator",
|
||||
"assignments": {"0": "narrator", "1": "narrator", "2": "narrator"},
|
||||
"speakers": {
|
||||
"narrator": {
|
||||
"label": "Narrator",
|
||||
"count": 3,
|
||||
"confidence": "low",
|
||||
"sample_quotes": [],
|
||||
"suppressed": False,
|
||||
}
|
||||
},
|
||||
"suppressed": [],
|
||||
"stats": {
|
||||
"total_chunks": 3,
|
||||
"explicit_chunks": 0,
|
||||
"active_speakers": 0,
|
||||
"unique_speakers": 1,
|
||||
"suppressed": 0,
|
||||
},
|
||||
}
|
||||
mock_analyze.return_value = mock_result
|
||||
|
||||
result = prepare_speaker_metadata(
|
||||
chapters=self._make_chapters(),
|
||||
chunks=self._make_chunks(),
|
||||
voice="af_heart",
|
||||
voice_profile=None,
|
||||
threshold=3,
|
||||
run_analysis=True,
|
||||
)
|
||||
chunk_list, roster, analysis, _, _ = result
|
||||
assert "narrator" in roster
|
||||
assert analysis["version"] == "1.0"
|
||||
|
||||
@patch("abogen.domain.speaker_metadata.load_settings")
|
||||
@patch("abogen.domain.speaker_metadata.analyze_speakers")
|
||||
def test_inject_recommended_callback_called(self, mock_analyze, mock_settings):
|
||||
from abogen.domain.speaker_metadata import prepare_speaker_metadata
|
||||
|
||||
mock_settings.return_value = {"speaker_random_languages": []}
|
||||
mock_result = MagicMock()
|
||||
mock_result.to_dict.return_value = {
|
||||
"version": "1.0",
|
||||
"narrator": "narrator",
|
||||
"assignments": {},
|
||||
"speakers": {
|
||||
"narrator": {
|
||||
"label": "Narrator",
|
||||
"count": 1,
|
||||
"confidence": "low",
|
||||
"sample_quotes": [],
|
||||
"suppressed": False,
|
||||
}
|
||||
},
|
||||
"suppressed": [],
|
||||
"stats": {
|
||||
"total_chunks": 1,
|
||||
"explicit_chunks": 0,
|
||||
"active_speakers": 0,
|
||||
"unique_speakers": 1,
|
||||
"suppressed": 0,
|
||||
},
|
||||
}
|
||||
mock_analyze.return_value = mock_result
|
||||
|
||||
injected = []
|
||||
callback = lambda roster, **kwargs: injected.append(dict(roster))
|
||||
|
||||
result = prepare_speaker_metadata(
|
||||
chapters=self._make_chapters(),
|
||||
chunks=self._make_chunks(1),
|
||||
voice="af_heart",
|
||||
voice_profile=None,
|
||||
threshold=3,
|
||||
run_analysis=True,
|
||||
inject_recommended=callback,
|
||||
)
|
||||
assert len(injected) == 1
|
||||
assert "narrator" in injected[0]
|
||||
|
||||
@patch("abogen.domain.speaker_metadata.load_settings")
|
||||
def test_inject_recommended_not_called_when_none(self, mock_settings):
|
||||
from abogen.domain.speaker_metadata import prepare_speaker_metadata
|
||||
|
||||
mock_settings.return_value = {"speaker_random_languages": []}
|
||||
result = prepare_speaker_metadata(
|
||||
chapters=self._make_chapters(),
|
||||
chunks=self._make_chunks(),
|
||||
voice="af_heart",
|
||||
voice_profile=None,
|
||||
threshold=3,
|
||||
run_analysis=False,
|
||||
inject_recommended=None,
|
||||
)
|
||||
assert result is not None
|
||||
|
||||
@patch("abogen.domain.speaker_metadata.load_settings")
|
||||
def test_chunks_are_copies(self, mock_settings):
|
||||
from abogen.domain.speaker_metadata import prepare_speaker_metadata
|
||||
|
||||
mock_settings.return_value = {"speaker_random_languages": []}
|
||||
original_chunks = [{"id": "0", "text": "Hello"}]
|
||||
result = prepare_speaker_metadata(
|
||||
chapters=[{"title": "Ch1", "chunks": original_chunks}],
|
||||
chunks=original_chunks,
|
||||
voice="af_heart",
|
||||
voice_profile=None,
|
||||
threshold=3,
|
||||
run_analysis=False,
|
||||
)
|
||||
chunk_list = result[0]
|
||||
assert chunk_list is not original_chunks
|
||||
assert chunk_list[0] is not original_chunks[0]
|
||||
|
||||
@patch("abogen.domain.speaker_metadata.load_settings")
|
||||
def test_analysis_disabled_sets_narrator_on_all_chunks(self, mock_settings):
|
||||
from abogen.domain.speaker_metadata import prepare_speaker_metadata
|
||||
|
||||
mock_settings.return_value = {"speaker_random_languages": []}
|
||||
chunks = [{"id": "0"}, {"id": "1"}, {"id": "2"}]
|
||||
result = prepare_speaker_metadata(
|
||||
chapters=[{"title": "Ch1", "chunks": chunks}],
|
||||
chunks=chunks,
|
||||
voice="af_heart",
|
||||
voice_profile=None,
|
||||
threshold=3,
|
||||
run_analysis=False,
|
||||
)
|
||||
for chunk in result[0]:
|
||||
assert chunk["speaker_id"] == "narrator"
|
||||
assert chunk["speaker_label"] == "Narrator"
|
||||
|
||||
@patch("abogen.domain.speaker_metadata.load_settings")
|
||||
@patch("abogen.domain.speaker_metadata.analyze_speakers")
|
||||
def test_speaker_random_languages_used(self, mock_analyze, mock_settings):
|
||||
from abogen.domain.speaker_metadata import prepare_speaker_metadata
|
||||
|
||||
mock_settings.return_value = {"speaker_random_languages": ["a", "b"]}
|
||||
mock_result = MagicMock()
|
||||
mock_result.to_dict.return_value = {
|
||||
"version": "1.0",
|
||||
"narrator": "narrator",
|
||||
"assignments": {},
|
||||
"speakers": {
|
||||
"narrator": {
|
||||
"label": "Narrator",
|
||||
"count": 1,
|
||||
"confidence": "low",
|
||||
"sample_quotes": [],
|
||||
"suppressed": False,
|
||||
}
|
||||
},
|
||||
"suppressed": [],
|
||||
"stats": {
|
||||
"total_chunks": 1,
|
||||
"explicit_chunks": 0,
|
||||
"active_speakers": 0,
|
||||
"unique_speakers": 1,
|
||||
"suppressed": 0,
|
||||
},
|
||||
}
|
||||
mock_analyze.return_value = mock_result
|
||||
|
||||
result = prepare_speaker_metadata(
|
||||
chapters=self._make_chapters(),
|
||||
chunks=self._make_chunks(1),
|
||||
voice="af_heart",
|
||||
voice_profile=None,
|
||||
threshold=3,
|
||||
run_analysis=True,
|
||||
)
|
||||
_, _, analysis, _, _ = result
|
||||
assert analysis["config_languages"] == ["a", "b"]
|
||||
|
||||
@patch("abogen.domain.speaker_metadata.load_settings")
|
||||
@patch("abogen.domain.speaker_metadata.analyze_speakers")
|
||||
def test_apply_config_with_speaker_config(self, mock_analyze, mock_settings):
|
||||
from abogen.domain.speaker_metadata import prepare_speaker_metadata
|
||||
|
||||
mock_settings.return_value = {"speaker_random_languages": []}
|
||||
mock_result = MagicMock()
|
||||
mock_result.to_dict.return_value = {
|
||||
"version": "1.0",
|
||||
"narrator": "narrator",
|
||||
"assignments": {"0": "narrator"},
|
||||
"speakers": {
|
||||
"narrator": {
|
||||
"label": "Narrator",
|
||||
"count": 1,
|
||||
"confidence": "low",
|
||||
"sample_quotes": [],
|
||||
"suppressed": False,
|
||||
}
|
||||
},
|
||||
"suppressed": [],
|
||||
"stats": {
|
||||
"total_chunks": 1,
|
||||
"explicit_chunks": 0,
|
||||
"active_speakers": 0,
|
||||
"unique_speakers": 1,
|
||||
"suppressed": 0,
|
||||
},
|
||||
}
|
||||
mock_analyze.return_value = mock_result
|
||||
|
||||
speaker_config = {
|
||||
"languages": ["a"],
|
||||
"speakers": {
|
||||
"narrator": {"id": "narrator", "voice": "am_echo"},
|
||||
},
|
||||
}
|
||||
result = prepare_speaker_metadata(
|
||||
chapters=self._make_chapters(),
|
||||
chunks=self._make_chunks(1),
|
||||
voice="af_heart",
|
||||
voice_profile=None,
|
||||
threshold=3,
|
||||
run_analysis=True,
|
||||
speaker_config=speaker_config,
|
||||
apply_config=True,
|
||||
)
|
||||
_, roster, _, languages, _ = result
|
||||
assert roster["narrator"]["voice"] == "af_heart"
|
||||
assert languages == ["a"]
|
||||
Reference in New Issue
Block a user