diff --git a/abogen/domain/speaker_metadata.py b/abogen/domain/speaker_metadata.py new file mode 100644 index 0000000..6481efb --- /dev/null +++ b/abogen/domain/speaker_metadata.py @@ -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 diff --git a/abogen/webui/routes/utils/form.py b/abogen/webui/routes/utils/form.py index ecb24bd..77398f2 100644 --- a/abogen/webui/routes/utils/form.py +++ b/abogen/webui/routes/utils/form.py @@ -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 = {} diff --git a/abogen/webui/routes/utils/voice.py b/abogen/webui/routes/utils/voice.py index 17ec2ec..b6bbad3 100644 --- a/abogen/webui/routes/utils/voice.py +++ b/abogen/webui/routes/utils/voice.py @@ -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() diff --git a/tests/test_domain_speaker_metadata.py b/tests/test_domain_speaker_metadata.py new file mode 100644 index 0000000..049297e --- /dev/null +++ b/tests/test_domain_speaker_metadata.py @@ -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"]