diff --git a/abogen/domain/pronunciation.py b/abogen/domain/pronunciation.py new file mode 100644 index 0000000..2b1f05a --- /dev/null +++ b/abogen/domain/pronunciation.py @@ -0,0 +1,261 @@ +"""Pronunciation rule compilation and application. + +Pure functions for compiling token-level and sentence-level pronunciation +overrides into regex patterns, applying them to text, and merging multiple +override sources with precedence rules. +""" + +from __future__ import annotations + +import re +from typing import Any, Dict, Iterable, List, Mapping, Optional + +from abogen.entity_analysis import normalize_token as normalize_entity_token +from abogen.entity_analysis import normalize_manual_override_token + + +def compile_pronunciation_rules( + overrides: Optional[Iterable[Mapping[str, Any]]], +) -> List[Dict[str, Any]]: + if not overrides: + return [] + + candidates: List[Dict[str, Any]] = [] + seen: set[str] = set() + + for entry in overrides: + if not isinstance(entry, Mapping): + continue + pronunciation_value = str(entry.get("pronunciation") or "").strip() + if not pronunciation_value: + continue + + token_values: List[str] = [] + token_raw = entry.get("token") + if token_raw: + token_value = str(token_raw).strip() + if token_value: + token_values.append(token_value) + normalized_raw = entry.get("normalized") + if normalized_raw: + normalized_value = str(normalized_raw).strip() + if normalized_value: + token_values.append(normalized_value) + if token_raw and not token_values: + fallback = normalize_entity_token(str(token_raw)) + if fallback: + token_values.append(fallback) + + if not token_values: + continue + + usage_normalized = str(entry.get("normalized") or "").strip() + if not usage_normalized and token_values: + usage_normalized = normalize_entity_token(token_values[0]) or token_values[0] + usage_token = str(entry.get("token") or token_values[0]) + + for token_value in token_values: + key = token_value.casefold() + if key in seen: + continue + seen.add(key) + candidates.append( + { + "token": token_value, + "normalized": usage_normalized, + "replacement": pronunciation_value, + } + ) + + if not candidates: + return [] + + candidates.sort(key=lambda item: len(item["token"]), reverse=True) + compiled: List[Dict[str, Any]] = [] + for candidate in candidates: + token_value = candidate["token"] + pronunciation_value = candidate["replacement"] + escaped = re.escape(token_value) + pattern = re.compile(rf"(?i)(?'s|\u2019s|\u2019)?(?!\w)") + compiled.append( + { + "pattern": pattern, + "replacement": pronunciation_value, + "normalized": candidate.get("normalized") or token_value, + "token": candidate.get("token") or token_value, + } + ) + + return compiled + + +def compile_heteronym_sentence_rules( + overrides: Optional[Iterable[Mapping[str, Any]]], +) -> List[Dict[str, Any]]: + if not overrides: + return [] + + compiled: List[Dict[str, Any]] = [] + seen: set[str] = set() + + for entry in overrides: + if not isinstance(entry, Mapping): + continue + sentence = str(entry.get("sentence") or "").strip() + if not sentence: + continue + choice = str(entry.get("choice") or "").strip() + if not choice: + continue + + replacement_sentence = "" + options = entry.get("options") + if isinstance(options, list): + for opt in options: + if not isinstance(opt, Mapping): + continue + if str(opt.get("key") or "").strip() == choice: + replacement_sentence = str(opt.get("replacement_sentence") or "").strip() + break + if not replacement_sentence: + continue + + rule_key = f"{sentence}\n{choice}".casefold() + if rule_key in seen: + continue + seen.add(rule_key) + + parts = [p for p in re.split(r"\s+", sentence) if p] + if not parts: + continue + pattern_text = r"\s+".join(re.escape(p) for p in parts) + pattern = re.compile(pattern_text) + compiled.append({"pattern": pattern, "replacement": replacement_sentence}) + + compiled.sort(key=lambda item: len(item["pattern"].pattern), reverse=True) + return compiled + + +def apply_heteronym_sentence_rules(text: str, rules: List[Dict[str, Any]]) -> str: + if not text or not rules: + return text + result = text + for rule in rules: + pattern = rule["pattern"] + replacement = rule["replacement"] + result = pattern.sub(replacement, result) + return result + + +def apply_pronunciation_rules( + text: str, + rules: List[Dict[str, Any]], + usage_counter: Optional[Dict[str, int]] = None, +) -> str: + if not text or not rules: + return text + + result = text + for rule in rules: + pattern = rule["pattern"] + pronunciation_value = rule["replacement"] + usage_key = str(rule.get("normalized") or "").strip() + + def _replacement(match: re.Match[str]) -> str: + suffix = match.group("possessive") or "" + if usage_counter is not None and usage_key: + usage_counter[usage_key] = usage_counter.get(usage_key, 0) + 1 + return pronunciation_value + suffix + + result = pattern.sub(_replacement, result) + + return result + + +def merge_pronunciation_overrides(job: Any) -> List[Dict[str, Any]]: + """Return pronunciation override entries, ensuring manual overrides are included. + + Pending jobs keep both ``manual_overrides`` and ``pronunciation_overrides``, but the + latter can be stale if the UI didn't resync before enqueue. During conversion, + we must merge manual overrides so they always apply (before TTS). + + Precedence: manual overrides win over existing entries for the same normalized key. + """ + + collected: Dict[str, Dict[str, Any]] = {} + + existing = getattr(job, "pronunciation_overrides", None) + if isinstance(existing, list): + for entry in existing: + if not isinstance(entry, Mapping): + continue + token_value = str(entry.get("token") or "").strip() + pronunciation_value = str(entry.get("pronunciation") or "").strip() + if not token_value or not pronunciation_value: + continue + normalized = str(entry.get("normalized") or "").strip() or normalize_entity_token(token_value) + if not normalized: + continue + collected[normalized] = { + "token": token_value, + "normalized": normalized, + "pronunciation": pronunciation_value, + "voice": str(entry.get("voice") or "").strip() or None, + "notes": str(entry.get("notes") or "").strip() or None, + "context": str(entry.get("context") or "").strip() or None, + "source": str(entry.get("source") or "pronunciation"), + "language": getattr(job, "language", None), + } + + speakers = getattr(job, "speakers", None) + if isinstance(speakers, dict): + for payload in speakers.values(): + if not isinstance(payload, Mapping): + continue + token_value = str(payload.get("token") or "").strip() + pronunciation_value = str(payload.get("pronunciation") or "").strip() + if not token_value or not pronunciation_value: + continue + normalized = normalize_entity_token(token_value) + if not normalized: + continue + collected[normalized] = { + "token": token_value, + "normalized": normalized, + "pronunciation": pronunciation_value, + "voice": str( + payload.get("resolved_voice") + or payload.get("voice") + or getattr(job, "voice", "") + ).strip() + or None, + "notes": None, + "context": None, + "source": "speaker", + "language": getattr(job, "language", None), + } + + manual = getattr(job, "manual_overrides", None) + if isinstance(manual, list): + for entry in manual: + if not isinstance(entry, Mapping): + continue + token_value = str(entry.get("token") or "").strip() + pronunciation_value = str(entry.get("pronunciation") or "").strip() + if not token_value or not pronunciation_value: + continue + normalized = str(entry.get("normalized") or "").strip() or normalize_manual_override_token(token_value) + if not normalized: + continue + collected[normalized] = { + "token": token_value, + "normalized": normalized, + "pronunciation": pronunciation_value, + "voice": str(entry.get("voice") or "").strip() or None, + "notes": str(entry.get("notes") or "").strip() or None, + "context": str(entry.get("context") or "").strip() or None, + "source": str(entry.get("source") or "manual"), + "language": getattr(job, "language", None), + } + + return list(collected.values()) diff --git a/abogen/webui/conversion_runner.py b/abogen/webui/conversion_runner.py index c7311e6..be39f6c 100644 --- a/abogen/webui/conversion_runner.py +++ b/abogen/webui/conversion_runner.py @@ -72,6 +72,13 @@ from abogen.domain.file_type import ( chapter_label as _chapter_label, update_metadata_for_chapter_count as _update_metadata_for_chapter_count, ) +from abogen.domain.pronunciation import ( + compile_pronunciation_rules as _compile_pronunciation_rules, + compile_heteronym_sentence_rules as _compile_heteronym_sentence_rules, + apply_heteronym_sentence_rules as _apply_heteronym_sentence_rules, + apply_pronunciation_rules as _apply_pronunciation_rules, + merge_pronunciation_overrides as _merge_pronunciation_overrides, +) from .service import Job, JobStatus @@ -399,266 +406,6 @@ def _normalize_for_pipeline( return normalize_for_pipeline(text, config=apostrophe_config, settings=runtime_settings) -def _merge_pronunciation_overrides(job: Any) -> List[Dict[str, Any]]: - """Return pronunciation override entries, ensuring manual overrides are included. - - Pending jobs keep both `manual_overrides` and `pronunciation_overrides`, but the - latter can be stale if the UI didn't resync before enqueue. During conversion, - we must merge manual overrides so they always apply (before TTS). - - Precedence: manual overrides win over existing entries for the same normalized key. - """ - - collected: Dict[str, Dict[str, Any]] = {} - - existing = getattr(job, "pronunciation_overrides", None) - if isinstance(existing, list): - for entry in existing: - if not isinstance(entry, Mapping): - continue - token_value = str(entry.get("token") or "").strip() - pronunciation_value = str(entry.get("pronunciation") or "").strip() - if not token_value or not pronunciation_value: - continue - normalized = str(entry.get("normalized") or "").strip() or normalize_entity_token(token_value) - if not normalized: - continue - collected[normalized] = { - "token": token_value, - "normalized": normalized, - "pronunciation": pronunciation_value, - "voice": str(entry.get("voice") or "").strip() or None, - "notes": str(entry.get("notes") or "").strip() or None, - "context": str(entry.get("context") or "").strip() or None, - "source": str(entry.get("source") or "pronunciation"), - "language": getattr(job, "language", None), - } - - # Speaker pronunciation entries (optional), mirrored from the pending-job collector. - speakers = getattr(job, "speakers", None) - if isinstance(speakers, dict): - for payload in speakers.values(): - if not isinstance(payload, Mapping): - continue - token_value = str(payload.get("token") or "").strip() - pronunciation_value = str(payload.get("pronunciation") or "").strip() - if not token_value or not pronunciation_value: - continue - normalized = normalize_entity_token(token_value) - if not normalized: - continue - collected[normalized] = { - "token": token_value, - "normalized": normalized, - "pronunciation": pronunciation_value, - "voice": str( - payload.get("resolved_voice") - or payload.get("voice") - or getattr(job, "voice", "") - ).strip() - or None, - "notes": None, - "context": None, - "source": "speaker", - "language": getattr(job, "language", None), - } - - # Manual overrides should take precedence. - manual = getattr(job, "manual_overrides", None) - if isinstance(manual, list): - for entry in manual: - if not isinstance(entry, Mapping): - continue - token_value = str(entry.get("token") or "").strip() - pronunciation_value = str(entry.get("pronunciation") or "").strip() - if not token_value or not pronunciation_value: - continue - normalized = str(entry.get("normalized") or "").strip() or normalize_manual_override_token(token_value) - if not normalized: - continue - collected[normalized] = { - "token": token_value, - "normalized": normalized, - "pronunciation": pronunciation_value, - "voice": str(entry.get("voice") or "").strip() or None, - "notes": str(entry.get("notes") or "").strip() or None, - "context": str(entry.get("context") or "").strip() or None, - "source": str(entry.get("source") or "manual"), - "language": getattr(job, "language", None), - } - - return list(collected.values()) - - -def _compile_pronunciation_rules( - overrides: Optional[Iterable[Mapping[str, Any]]], -) -> List[Dict[str, Any]]: - if not overrides: - return [] - - candidates: List[Dict[str, Any]] = [] - seen: set[str] = set() - - for entry in overrides: - if not isinstance(entry, Mapping): - continue - pronunciation_value = str(entry.get("pronunciation") or "").strip() - if not pronunciation_value: - continue - - token_values: List[str] = [] - token_raw = entry.get("token") - if token_raw: - token_value = str(token_raw).strip() - if token_value: - token_values.append(token_value) - normalized_raw = entry.get("normalized") - if normalized_raw: - normalized_value = str(normalized_raw).strip() - if normalized_value: - token_values.append(normalized_value) - if token_raw and not token_values: - fallback = normalize_entity_token(str(token_raw)) - if fallback: - token_values.append(fallback) - - if not token_values: - continue - - usage_normalized = str(entry.get("normalized") or "").strip() - if not usage_normalized and token_values: - usage_normalized = normalize_entity_token(token_values[0]) or token_values[0] - usage_token = str(entry.get("token") or token_values[0]) - - for token_value in token_values: - key = token_value.casefold() - if key in seen: - continue - seen.add(key) - candidates.append( - { - "token": token_value, - "normalized": usage_normalized, - "replacement": pronunciation_value, - } - ) - - if not candidates: - return [] - - candidates.sort(key=lambda item: len(item["token"]), reverse=True) - compiled: List[Dict[str, Any]] = [] - for candidate in candidates: - token_value = candidate["token"] - pronunciation_value = candidate["replacement"] - escaped = re.escape(token_value) - pattern = re.compile(rf"(?i)(?'s|\u2019s|\u2019)?(?!\w)") - compiled.append( - { - "pattern": pattern, - "replacement": pronunciation_value, - "normalized": candidate.get("normalized") or token_value, - "token": candidate.get("token") or token_value, - } - ) - - return compiled - - -def _compile_heteronym_sentence_rules( - overrides: Optional[Iterable[Mapping[str, Any]]], -) -> List[Dict[str, Any]]: - """Compile sentence-level replacements for heteronym disambiguation. - - These are intentionally scoped to a specific sentence string rather than a token, - so we can apply different pronunciations for the same word in different contexts. - - Expected override entry shape (from pending/job): - - sentence: original sentence text - - choice: selected option key - - options: [{key, replacement_sentence, ...}] - """ - if not overrides: - return [] - - compiled: List[Dict[str, Any]] = [] - seen: set[str] = set() - - for entry in overrides: - if not isinstance(entry, Mapping): - continue - sentence = str(entry.get("sentence") or "").strip() - if not sentence: - continue - choice = str(entry.get("choice") or "").strip() - if not choice: - continue - - replacement_sentence = "" - options = entry.get("options") - if isinstance(options, list): - for opt in options: - if not isinstance(opt, Mapping): - continue - if str(opt.get("key") or "").strip() == choice: - replacement_sentence = str(opt.get("replacement_sentence") or "").strip() - break - if not replacement_sentence: - continue - - rule_key = f"{sentence}\n{choice}".casefold() - if rule_key in seen: - continue - seen.add(rule_key) - - parts = [p for p in re.split(r"\s+", sentence) if p] - if not parts: - continue - pattern_text = r"\s+".join(re.escape(p) for p in parts) - pattern = re.compile(pattern_text) - compiled.append({"pattern": pattern, "replacement": replacement_sentence}) - - # Replace longer sentences first to avoid partial matches. - compiled.sort(key=lambda item: len(item["pattern"].pattern), reverse=True) - return compiled - - -def _apply_heteronym_sentence_rules(text: str, rules: List[Dict[str, Any]]) -> str: - if not text or not rules: - return text - result = text - for rule in rules: - pattern = rule["pattern"] - replacement = rule["replacement"] - result = pattern.sub(replacement, result) - return result - - -def _apply_pronunciation_rules( - text: str, - rules: List[Dict[str, Any]], - usage_counter: Optional[Dict[str, int]] = None, -) -> str: - if not text or not rules: - return text - - result = text - for rule in rules: - pattern = rule["pattern"] - pronunciation_value = rule["replacement"] - usage_key = str(rule.get("normalized") or "").strip() - - def _replacement(match: re.Match[str]) -> str: - suffix = match.group("possessive") or "" - if usage_counter is not None and usage_key: - usage_counter[usage_key] = usage_counter.get(usage_key, 0) + 1 - return pronunciation_value + suffix - - result = pattern.sub(_replacement, result) - - return result - - def _chapter_voice_spec(job: Job, override: Optional[Dict[str, Any]]) -> str: if not override: return _job_voice_fallback(job) diff --git a/tests/test_pronunciation.py b/tests/test_pronunciation.py new file mode 100644 index 0000000..f0b6bc1 --- /dev/null +++ b/tests/test_pronunciation.py @@ -0,0 +1,334 @@ +"""Tests for abogen.domain.pronunciation — compile/apply pronunciation rules.""" + +from __future__ import annotations + +import re +import pytest + + +# --------------------------------------------------------------------------- +# We import the domain functions. The module must be created first. +# For now the tests are written against the expected public API so they can +# serve as the contract during extraction. +# --------------------------------------------------------------------------- + + +class TestCompilePronunciationRules: + """compile_pronunciation_rules turns override dicts into regex-based rules.""" + + def test_empty_input(self): + from abogen.domain.pronunciation import compile_pronunciation_rules + + assert compile_pronunciation_rules(None) == [] + assert compile_pronunciation_rules([]) == [] + + def test_single_entry(self): + from abogen.domain.pronunciation import compile_pronunciation_rules + + overrides = [{"token": "albeit", "pronunciation": "all be it"}] + rules = compile_pronunciation_rules(overrides) + assert len(rules) == 1 + assert rules[0]["replacement"] == "all be it" + assert rules[0]["pattern"].search("albeit") + + def test_skips_entries_without_pronunciation(self): + from abogen.domain.pronunciation import compile_pronunciation_rules + + overrides = [{"token": "hello"}] + assert compile_pronunciation_rules(overrides) == [] + + def test_skips_entries_without_token(self): + from abogen.domain.pronunciation import compile_pronunciation_rules + + overrides = [{"pronunciation": "foo"}] + assert compile_pronunciation_rules(overrides) == [] + + def test_deduplication_by_casefold(self): + from abogen.domain.pronunciation import compile_pronunciation_rules + + overrides = [ + {"token": "Albeit", "pronunciation": "all be it"}, + {"token": "ALBEIT", "pronunciation": "all be it"}, + ] + rules = compile_pronunciation_rules(overrides) + assert len(rules) == 1 + + def test_longer_token_sorted_first(self): + from abogen.domain.pronunciation import compile_pronunciation_rules + + overrides = [ + {"token": "ice cream", "pronunciation": "ice cream"}, + {"token": "ice", "pronunciation": "ais"}, + ] + rules = compile_pronunciation_rules(overrides) + assert len(rules) == 2 + assert len(rules[0]["token"]) >= len(rules[1]["token"]) + + def test_normalized_fallback_to_entity_token(self): + from abogen.domain.pronunciation import compile_pronunciation_rules + + overrides = [{"normalized": "USA", "pronunciation": "you ess ay"}] + rules = compile_pronunciation_rules(overrides) + assert len(rules) == 1 + + def test_pattern_is_case_insensitive(self): + from abogen.domain.pronunciation import compile_pronunciation_rules + + overrides = [{"token": "hello", "pronunciation": "hi"}] + rules = compile_pronunciation_rules(overrides) + assert rules[0]["pattern"].search("Hello") + assert rules[0]["pattern"].search("HELLO") + + def test_non_mapping_items_skipped(self): + from abogen.domain.pronunciation import compile_pronunciation_rules + + overrides = ["bad", None, 42] + assert compile_pronunciation_rules(overrides) == [] + + +class TestCompileHeteronymSentenceRules: + """compile_heteronym_sentence_rules builds sentence-level replacements.""" + + def test_empty_input(self): + from abogen.domain.pronunciation import compile_heteronym_sentence_rules + + assert compile_heteronym_sentence_rules(None) == [] + assert compile_heteronym_sentence_rules([]) == [] + + def test_basic_replacement(self): + from abogen.domain.pronunciation import compile_heteronym_sentence_rules + + overrides = [ + { + "sentence": "I read the book", + "choice": "past", + "options": [ + {"key": "present", "replacement_sentence": "I read the book"}, + {"key": "past", "replacement_sentence": "I read the book"}, + ], + } + ] + rules = compile_heteronym_sentence_rules(overrides) + assert len(rules) == 1 + assert rules[0]["replacement"] == "I read the book" + + def test_skips_without_sentence(self): + from abogen.domain.pronunciation import compile_heteronym_sentence_rules + + overrides = [{"choice": "a", "options": []}] + assert compile_heteronym_sentence_rules(overrides) == [] + + def test_skips_without_choice(self): + from abogen.domain.pronunciation import compile_heteronym_sentence_rules + + overrides = [{"sentence": "hello", "options": []}] + assert compile_heteronym_sentence_rules(overrides) == [] + + def test_skips_when_no_matching_option(self): + from abogen.domain.pronunciation import compile_heteronym_sentence_rules + + overrides = [ + { + "sentence": "I read the book", + "choice": "past", + "options": [{"key": "present", "replacement_sentence": "I read the book"}], + } + ] + assert compile_heteronym_sentence_rules(overrides) == [] + + def test_deduplication(self): + from abogen.domain.pronunciation import compile_heteronym_sentence_rules + + entry = { + "sentence": "I read the book", + "choice": "past", + "options": [{"key": "past", "replacement_sentence": "I red the book"}], + } + rules = compile_heteronym_sentence_rules([entry, entry]) + assert len(rules) == 1 + + def test_longer_sentence_sorted_first(self): + from abogen.domain.pronunciation import compile_heteronym_sentence_rules + + overrides = [ + { + "sentence": "short", + "choice": "a", + "options": [{"key": "a", "replacement_sentence": "s"}], + }, + { + "sentence": "a longer sentence here", + "choice": "b", + "options": [{"key": "b", "replacement_sentence": "l"}], + }, + ] + rules = compile_heteronym_sentence_rules(overrides) + assert len(rules[0]["pattern"].pattern) >= len(rules[1]["pattern"].pattern) + + +class TestApplyPronunciationRules: + """apply_pronunciation_rules applies compiled token-level rules.""" + + def test_empty_text(self): + from abogen.domain.pronunciation import apply_pronunciation_rules + + assert apply_pronunciation_rules("", []) == "" + + def test_no_rules(self): + from abogen.domain.pronunciation import apply_pronunciation_rules + + assert apply_pronunciation_rules("hello", []) == "hello" + + def test_basic_replacement(self): + from abogen.domain.pronunciation import compile_pronunciation_rules, apply_pronunciation_rules + + rules = compile_pronunciation_rules([{"token": "albeit", "pronunciation": "all be it"}]) + result = apply_pronunciation_rules("albeit it was raining", rules) + assert result == "all be it it was raining" + + def test_possessive_preserved(self): + from abogen.domain.pronunciation import compile_pronunciation_rules, apply_pronunciation_rules + + rules = compile_pronunciation_rules([{"token": "dog", "pronunciation": "dawg"}]) + result = apply_pronunciation_rules("the dog's bone", rules) + assert result == "the dawg's bone" + + def test_usage_counter_increments(self): + from abogen.domain.pronunciation import compile_pronunciation_rules, apply_pronunciation_rules + + rules = compile_pronunciation_rules([{"token": "hello", "pronunciation": "hi"}]) + counter: dict[str, int] = {} + apply_pronunciation_rules("hello hello", rules, usage_counter=counter) + assert counter.get("hello", 0) == 2 + + def test_case_insensitive_match(self): + from abogen.domain.pronunciation import compile_pronunciation_rules, apply_pronunciation_rules + + rules = compile_pronunciation_rules([{"token": "test", "pronunciation": "tst"}]) + result = apply_pronunciation_rules("This is a Test", rules) + assert "tst" in result.lower() + + +class TestApplyHeteronymSentenceRules: + """apply_heteronym_sentence_rules applies sentence-level replacements.""" + + def test_empty_text(self): + from abogen.domain.pronunciation import apply_heteronym_sentence_rules + + assert apply_heteronym_sentence_rules("", []) == "" + + def test_no_rules(self): + from abogen.domain.pronunciation import apply_heteronym_sentence_rules + + assert apply_heteronym_sentence_rules("hello", []) == "hello" + + def test_basic_replacement(self): + from abogen.domain.pronunciation import ( + compile_heteronym_sentence_rules, + apply_heteronym_sentence_rules, + ) + + rules = compile_heteronym_sentence_rules( + [ + { + "sentence": "I read the book", + "choice": "past", + "options": [{"key": "past", "replacement_sentence": "I read the book"}], + } + ] + ) + result = apply_heteronym_sentence_rules("I read the book.", rules) + assert result == "I read the book." + + def test_no_match_left_unchanged(self): + from abogen.domain.pronunciation import ( + compile_heteronym_sentence_rules, + apply_heteronym_sentence_rules, + ) + + rules = compile_heteronym_sentence_rules( + [ + { + "sentence": "I read the book", + "choice": "past", + "options": [{"key": "past", "replacement_sentence": "I red the book"}], + } + ] + ) + result = apply_heteronym_sentence_rules("something else entirely", rules) + assert result == "something else entirely" + + +class TestMergePronunciationOverrides: + """merge_pronunciation_overrides consolidates override sources.""" + + def test_empty_job(self): + from abogen.domain.pronunciation import merge_pronunciation_overrides + + class FakeJob: + pronunciation_overrides = None + speakers = None + manual_overrides = None + language = "en" + + result = merge_pronunciation_overrides(FakeJob()) + assert result == [] + + def test_pronunciation_overrides_included(self): + from abogen.domain.pronunciation import merge_pronunciation_overrides + + class FakeJob: + pronunciation_overrides = [ + {"token": "hello", "pronunciation": "hi", "normalized": "hello"} + ] + speakers = None + manual_overrides = None + language = "en" + + result = merge_pronunciation_overrides(FakeJob()) + assert len(result) == 1 + assert result[0]["token"] == "hello" + assert result[0]["source"] == "pronunciation" + + def test_manual_overrides_win(self): + from abogen.domain.pronunciation import merge_pronunciation_overrides + + class FakeJob: + pronunciation_overrides = [ + {"token": "hello", "pronunciation": "hi old", "normalized": "hello"} + ] + speakers = None + manual_overrides = [ + {"token": "hello", "pronunciation": "hi new", "normalized": "hello"} + ] + language = "en" + + result = merge_pronunciation_overrides(FakeJob()) + assert len(result) == 1 + assert result[0]["pronunciation"] == "hi new" + assert result[0]["source"] == "manual" + + def test_speaker_entries_included(self): + from abogen.domain.pronunciation import merge_pronunciation_overrides + + class FakeJob: + pronunciation_overrides = None + speakers = {"narrator": {"token": "war", "pronunciation": "wɔːr"}} + manual_overrides = None + language = "en" + + result = merge_pronunciation_overrides(FakeJob()) + assert len(result) == 1 + assert result[0]["source"] == "speaker" + + def test_skips_empty_tokens(self): + from abogen.domain.pronunciation import merge_pronunciation_overrides + + class FakeJob: + pronunciation_overrides = [{"token": "", "pronunciation": "foo"}] + speakers = None + manual_overrides = None + language = "en" + + result = merge_pronunciation_overrides(FakeJob()) + assert result == []