mirror of
https://github.com/denizsafak/abogen.git
synced 2026-07-22 07:10:28 +02:00
refactor: wire up domain/chapter_overrides.py and domain/metadata_merge.py
- Update chapter_overrides.py to return tuple matching original signature - Import apply_chapter_overrides and merge_metadata from domain modules - Remove old function bodies from conversion_runner.py - Add tests/test_chapter_merge_normalize.py (19 tests) - conversion_runner.py: 1677 → 1574 lines - All tests pass
This commit is contained in:
@@ -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)
|
||||
return selected, metadata_updates, diagnostics
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user