fix: create missing domain/file_type.py from previous incomplete refactoring

abogen/domain/chapter_overrides.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional
+
+from abogen.text_extractor import ExtractedChapter
+
+
+@dataclass
+class ChapterOverrideResult:
+    selected: List[ExtractedChapter]
+    metadata_updates: Dict[str, str]
+    diagnostics: List[str]
+
+
+def apply_chapter_overrides(
+    extracted: List[ExtractedChapter],
+    overrides: List[Dict[str, Any]],
+    coerce_truthy_fn,
+) -> ChapterOverrideResult:
+    if not overrides:
+        return ChapterOverrideResult(selected=[], metadata_updates={}, diagnostics=[])
+
+    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_fn(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 ChapterOverrideResult(selected=selected, metadata_updates=metadata_updates, diagnostics=diagnostics)

abogen/domain/file_type.py

@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Dict, List, Tuple
+
+from abogen.text_extractor import ExtractedChapter
+
+
+_SIGNIFICANT_LENGTH_THRESHOLDS: Dict[str, int] = {"epub": 1000, "markdown": 500}
+_MIN_SHORT_CONTENT: Dict[str, int] = {"epub": 240, "markdown": 160}
+_STRUCTURAL_KEYWORDS = (
+    "preface",
+    "prologue",
+    "introduction",
+    "foreword",
+    "epilogue",
+    "afterword",
+    "appendix",
+    "acknowledgment",
+    "acknowledgement",
+)
+_STRUCTURAL_MIN_LENGTH = 120
+_MAX_SHORT_CHAPTERS = 2
+
+
+@dataclass
+class ChapterFilterResult:
+    kept: List[ExtractedChapter]
+    skipped: List[Tuple[str, int]]
+
+
+def infer_file_type(path: Path) -> str:
+    suffix = path.suffix.lower()
+    if suffix == ".epub":
+        return "epub"
+    if suffix in {".md", ".markdown"}:
+        return "markdown"
+    if suffix == ".pdf":
+        return "pdf"
+    if suffix == ".txt":
+        return "text"
+    return suffix.lstrip(".") or "text"
+
+
+def looks_structural(title: str) -> bool:
+    lowered = title.strip().lower()
+    if not lowered:
+        return False
+    return any(keyword in lowered for keyword in _STRUCTURAL_KEYWORDS)
+
+
+def chapter_label(file_type: str) -> str:
+    return "chapters" if file_type.lower() in {"epub", "markdown"} else "pages"
+
+
+def auto_select_relevant_chapters(
+    chapters: List[ExtractedChapter],
+    file_type: str,
+) -> ChapterFilterResult:
+    if not chapters:
+        return ChapterFilterResult(kept=[], skipped=[])
+
+    normalized = file_type.lower()
+    threshold = _SIGNIFICANT_LENGTH_THRESHOLDS.get(normalized, 0)
+    min_short = _MIN_SHORT_CONTENT.get(normalized, 0)
+
+    kept: List[ExtractedChapter] = []
+    skipped: List[Tuple[str, int]] = []
+    short_kept = 0
+
+    for chapter in chapters:
+        stripped = chapter.text.strip()
+        length = len(stripped)
+        if length == 0:
+            skipped.append((chapter.title, length))
+            continue
+
+        keep = False
+        if threshold == 0:
+            keep = True
+        elif length >= threshold:
+            keep = True
+        elif not kept:
+            keep = True
+        elif min_short and length >= min_short and short_kept < _MAX_SHORT_CHAPTERS:
+            keep = True
+            short_kept += 1
+        elif looks_structural(chapter.title) and length >= _STRUCTURAL_MIN_LENGTH:
+            keep = True
+
+        if keep:
+            kept.append(chapter)
+        else:
+            skipped.append((chapter.title, length))
+
+    if kept:
+        return ChapterFilterResult(kept=kept, skipped=skipped)
+
+    longest_idx = None
+    longest_length = 0
+    for idx, chapter in enumerate(chapters):
+        stripped = chapter.text.strip()
+        if stripped and len(stripped) > longest_length:
+            longest_length = len(stripped)
+            longest_idx = idx
+
+    if longest_idx is not None:
+        longest = chapters[longest_idx]
+        fallback_skipped = [
+            (chapter.title, len(chapter.text.strip()))
+            for idx, chapter in enumerate(chapters)
+            if idx != longest_idx and chapter.text.strip()
+        ]
+        return ChapterFilterResult(kept=[longest], skipped=fallback_skipped)
+
+    return ChapterFilterResult(kept=[], skipped=skipped)
+
+
+def update_metadata_for_chapter_count(
+    metadata: Dict[str, Any], count: int, file_type: str
+) -> None:
+    if not metadata or count <= 0:
+        return
+
+    label = "Chapters" if file_type.lower() in {"epub", "markdown"} else "Pages"
+    metadata["chapter_count"] = str(count)
+
+    pattern = re.compile(r"\(\d+\s+(Chapters?|Pages?)\)")
+    replacement = f"({count} {label})"
+    for key in ("album", "ALBUM"):
+        value = metadata.get(key)
+        if not isinstance(value, str):
+            continue
+        metadata[key] = pattern.sub(replacement, value)

abogen/domain/metadata_merge.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+from typing import Any, Dict, Optional
+
+
+def merge_metadata(
+    extracted: Optional[Dict[str, Any]],
+    overrides: Optional[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)
+    if overrides:
+        for key, value in overrides.items():
+            key_str = str(key)
+            if value is None:
+                merged.pop(key_str, None)
+            else:
+                merged[key_str] = str(value)
+    return merged