diff --git a/abogen/domain/chapter_overrides.py b/abogen/domain/chapter_overrides.py index f9fda27..15d9866 100644 --- a/abogen/domain/chapter_overrides.py +++ b/abogen/domain/chapter_overrides.py @@ -1,25 +1,17 @@ from __future__ import annotations -from dataclasses import dataclass -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple from abogen.text_extractor import ExtractedChapter - - -@dataclass -class ChapterOverrideResult: - selected: List[ExtractedChapter] - metadata_updates: Dict[str, str] - diagnostics: List[str] +from abogen.domain.voice_utils import coerce_truthy def apply_chapter_overrides( extracted: List[ExtractedChapter], overrides: List[Dict[str, Any]], - coerce_truthy_fn, -) -> ChapterOverrideResult: +) -> Tuple[List[ExtractedChapter], Dict[str, str], List[str]]: if not overrides: - return ChapterOverrideResult(selected=[], metadata_updates={}, diagnostics=[]) + return [], {}, [] selected: List[ExtractedChapter] = [] metadata_updates: Dict[str, str] = {} @@ -32,7 +24,7 @@ def apply_chapter_overrides( ) continue - enabled = coerce_truthy_fn(payload.get("enabled", True)) + enabled = coerce_truthy(payload.get("enabled", True)) payload["enabled"] = enabled if not enabled: continue @@ -97,4 +89,4 @@ def apply_chapter_overrides( selected.append(ExtractedChapter(title=title_value, text=text_value)) - return ChapterOverrideResult(selected=selected, metadata_updates=metadata_updates, diagnostics=diagnostics) \ No newline at end of file + return selected, metadata_updates, diagnostics diff --git a/abogen/webui/conversion_runner.py b/abogen/webui/conversion_runner.py index 6de393e..e8a24b2 100644 --- a/abogen/webui/conversion_runner.py +++ b/abogen/webui/conversion_runner.py @@ -87,6 +87,8 @@ from abogen.domain.voice_resolution import ( chapter_voice_spec as _chapter_voice_spec, chunk_voice_spec as _chunk_voice_spec, ) +from abogen.domain.chapter_overrides import apply_chapter_overrides as _apply_chapter_overrides +from abogen.domain.metadata_merge import merge_metadata as _merge_metadata from .service import Job, JobStatus @@ -192,111 +194,6 @@ def _coerce_truthy(value: Any, default: bool = True) -> bool: _OUTPUT_SANITIZE_RE = re.compile(r"[^\w\-_.]+") -def _apply_chapter_overrides( - extracted: List[ExtractedChapter], - overrides: List[Dict[str, Any]], -) -> tuple[List[ExtractedChapter], Dict[str, str], List[str]]: - if not overrides: - return [], {}, [] - - selected: List[ExtractedChapter] = [] - metadata_updates: Dict[str, str] = {} - diagnostics: List[str] = [] - - for position, payload in enumerate(overrides): - if not isinstance(payload, dict): - diagnostics.append( - f"Skipped chapter override at position {position + 1}: unsupported payload type {type(payload).__name__}." - ) - continue - - enabled = _coerce_truthy(payload.get("enabled", True)) - payload["enabled"] = enabled - if not enabled: - continue - - metadata_payload = payload.get("metadata") or {} - if isinstance(metadata_payload, dict): - for key, value in metadata_payload.items(): - if value is None: - continue - metadata_updates[str(key)] = str(value) - - base: Optional[ExtractedChapter] = None - idx_candidate = payload.get("index") - idx_normalized: Optional[int] = None - if isinstance(idx_candidate, int): - idx_normalized = idx_candidate - elif isinstance(idx_candidate, str): - try: - idx_normalized = int(idx_candidate) - except ValueError: - idx_normalized = None - if idx_normalized is not None and 0 <= idx_normalized < len(extracted): - base = extracted[idx_normalized] - payload["index"] = idx_normalized - - if base is None: - source_title = payload.get("source_title") - if isinstance(source_title, str): - base = next((chapter for chapter in extracted if chapter.title == source_title), None) - - if base is None: - candidate_title = payload.get("title") - if isinstance(candidate_title, str): - base = next((chapter for chapter in extracted if chapter.title == candidate_title), None) - - text_override = payload.get("text") - if text_override is not None: - text_value = str(text_override) - elif base is not None: - text_value = base.text - else: - diagnostics.append( - f"Skipped chapter override at position {position + 1}: no text provided and no matching source chapter found." - ) - continue - - title_override = payload.get("title") - if title_override is not None: - title_value = str(title_override) - elif base is not None: - title_value = base.title - else: - title_value = f"Chapter {position + 1}" - - if base and not payload.get("source_title"): - payload["source_title"] = base.title - - payload["title"] = title_value - payload["text"] = text_value - payload["characters"] = len(text_value) - payload.setdefault("order", payload.get("order", position)) - - selected.append(ExtractedChapter(title=title_value, text=text_value)) - - return selected, metadata_updates, diagnostics - - -def _merge_metadata( - extracted: Optional[Dict[str, str]], - overrides: Dict[str, Any], -) -> Dict[str, str]: - merged: Dict[str, str] = {} - if extracted: - for key, value in extracted.items(): - if value is None: - continue - merged[str(key)] = str(value) - for key, value in (overrides or {}).items(): - key_str = str(key) - if value is None: - merged.pop(key_str, None) - else: - merged[key_str] = str(value) - return merged - - _APOSTROPHE_CONFIG = ApostropheConfig() diff --git a/tests/test_chapter_merge_normalize.py b/tests/test_chapter_merge_normalize.py new file mode 100644 index 0000000..89054a8 --- /dev/null +++ b/tests/test_chapter_merge_normalize.py @@ -0,0 +1,167 @@ +"""Tests for chapter_overrides, merge_metadata, normalize_for_pipeline. + +Tests import from domain modules (new location). +""" + +from __future__ import annotations + +import pytest +from abogen.text_extractor import ExtractedChapter + + +# --------------------------------------------------------------------------- +# apply_chapter_overrides +# --------------------------------------------------------------------------- + + +class TestApplyChapterOverrides: + """apply_chapter_overrides applies chapter overrides to extracted chapters.""" + + def test_empty_overrides(self): + from abogen.domain.chapter_overrides import apply_chapter_overrides + + result, updates, diags = apply_chapter_overrides([], []) + assert result == [] + assert updates == {} + assert diags == [] + + def test_basic_override_by_index(self): + from abogen.domain.chapter_overrides import apply_chapter_overrides + + chapters = [ExtractedChapter(title="Ch1", text="original")] + overrides = [{"index": 0, "title": "New Title", "text": "new text"}] + result, updates, diags = apply_chapter_overrides(chapters, overrides) + assert len(result) == 1 + assert result[0].title == "New Title" + assert result[0].text == "new text" + + def test_override_by_source_title(self): + from abogen.domain.chapter_overrides import apply_chapter_overrides + + chapters = [ExtractedChapter(title="Ch1", text="text1")] + overrides = [{"source_title": "Ch1", "title": "Renamed"}] + result, _, _ = apply_chapter_overrides(chapters, overrides) + assert result[0].title == "Renamed" + + def test_disabled_override_skipped(self): + from abogen.domain.chapter_overrides import apply_chapter_overrides + + chapters = [ExtractedChapter(title="Ch1", text="text1")] + overrides = [{"index": 0, "enabled": False}] + result, _, _ = apply_chapter_overrides(chapters, overrides) + assert len(result) == 0 + + def test_metadata_updates_collected(self): + from abogen.domain.chapter_overrides import apply_chapter_overrides + + chapters = [ExtractedChapter(title="Ch1", text="text1")] + overrides = [{"index": 0, "metadata": {"album": "New Album"}}] + _, updates, _ = apply_chapter_overrides(chapters, overrides) + assert updates["album"] == "New Album" + + def test_no_matching_chapter_diagnostic(self): + from abogen.domain.chapter_overrides import apply_chapter_overrides + + overrides = [{"index": 99, "title": "X"}] + _, _, diags = apply_chapter_overrides([], overrides) + assert len(diags) == 1 + assert "Skipped" in diags[0] + + def test_non_dict_override_skipped(self): + from abogen.domain.chapter_overrides import apply_chapter_overrides + + _, _, diags = apply_chapter_overrides([], ["bad"]) + assert len(diags) == 1 + + def test_text_from_base_when_not_provided(self): + from abogen.domain.chapter_overrides import apply_chapter_overrides + + chapters = [ExtractedChapter(title="Ch1", text="original text")] + overrides = [{"index": 0, "title": "New Title"}] + result, _, _ = apply_chapter_overrides(chapters, overrides) + assert result[0].text == "original text" + + def test_default_title_when_no_base(self): + from abogen.domain.chapter_overrides import apply_chapter_overrides + + overrides = [{"text": "some text"}] + result, _, _ = apply_chapter_overrides([], overrides) + assert result[0].title == "Chapter 1" + + +# --------------------------------------------------------------------------- +# merge_metadata +# --------------------------------------------------------------------------- + + +class TestMergeMetadata: + """merge_metadata merges extracted metadata with overrides.""" + + def test_both_empty(self): + from abogen.domain.metadata_merge import merge_metadata + + assert merge_metadata({}, {}) == {} + + def test_only_extracted(self): + from abogen.domain.metadata_merge import merge_metadata + + result = merge_metadata({"album": "Book"}, {}) + assert result == {"album": "Book"} + + def test_only_overrides(self): + from abogen.domain.metadata_merge import merge_metadata + + result = merge_metadata(None, {"album": "Override"}) + assert result == {"album": "Override"} + + def test_override_wins(self): + from abogen.domain.metadata_merge import merge_metadata + + result = merge_metadata({"album": "Old"}, {"album": "New"}) + assert result == {"album": "New"} + + def test_none_value_deletes_key(self): + from abogen.domain.metadata_merge import merge_metadata + + result = merge_metadata({"album": "Book"}, {"album": None}) + assert "album" not in result + + def test_none_values_in_extracted_skipped(self): + from abogen.domain.metadata_merge import merge_metadata + + result = merge_metadata({"album": None, "artist": "X"}, {}) + assert result == {"artist": "X"} + + def test_numeric_values_stringified(self): + from abogen.domain.metadata_merge import merge_metadata + + result = merge_metadata({"track": 1}, {}) + assert result["track"] == "1" + + +# --------------------------------------------------------------------------- +# normalize_for_pipeline (thin wrapper) +# --------------------------------------------------------------------------- + + +class TestNormalizeForPipeline: + """normalize_for_pipeline normalizes text with runtime settings.""" + + def test_basic_normalize(self): + from abogen.webui.conversion_runner import _normalize_for_pipeline + + result = _normalize_for_pipeline("Hello World") + assert isinstance(result, str) + assert len(result) > 0 + + def test_empty_string(self): + from abogen.webui.conversion_runner import _normalize_for_pipeline + + result = _normalize_for_pipeline("") + assert result == "" + + def test_with_overrides(self): + from abogen.webui.conversion_runner import _normalize_for_pipeline + + result = _normalize_for_pipeline("test", normalization_overrides={"number_format": "words"}) + assert isinstance(result, str)