feat: Implement chapter overrides and metadata merging in conversion process

- Added `_coerce_truthy` function to handle truthy value coercion.
- Introduced `_apply_chapter_overrides` to apply chapter modifications based on provided overrides.
- Implemented `_merge_metadata` to combine extracted metadata with overrides, ensuring proper handling of None values.
- Updated `run_conversion_job` to utilize new chapter override and metadata merging functionalities.
- Modified `Job` class to store chapters as dictionaries for better flexibility.
- Enhanced `ConversionService` to normalize chapter input and metadata tags.
- Added comprehensive tests for chapter overrides and metadata merging to ensure functionality and correctness.

abogen/web/conversion_runner.py

@@ -9,7 +9,7 @@ import sys
 from contextlib import ExitStack
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Callable, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional
 
 import numpy as np
 import soundfile as sf
@@ -43,6 +43,126 @@ class AudioSink:
     write: Callable[[np.ndarray], None]
 
 
+def _coerce_truthy(value: Any, default: bool = True) -> bool:
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        lowered = value.strip().lower()
+        if lowered in {"true", "1", "yes", "on"}:
+            return True
+        if lowered in {"false", "0", "no", "off"}:
+            return False
+        return default
+    if value is None:
+        return default
+    return bool(value)
+
+
+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
+
+
 def run_conversion_job(job: Job) -> None:
     job.add_log("Preparing conversion pipeline")
     canceller = _make_canceller(job)
@@ -53,11 +173,29 @@ def run_conversion_job(job: Job) -> None:
     try:
         pipeline = _load_pipeline(job)
         extraction = extract_from_path(job.stored_path)
-        job.metadata_tags = extraction.metadata or {}
+
+        metadata_overrides: Dict[str, Any] = dict(job.metadata_tags or {})
+        if job.chapters:
+            selected_chapters, chapter_metadata, diagnostics = _apply_chapter_overrides(
+                extraction.chapters,
+                job.chapters,
+            )
+            for message in diagnostics:
+                job.add_log(message, level="warning")
+            if selected_chapters:
+                extraction.chapters = selected_chapters
+                metadata_overrides.update(chapter_metadata)
+                job.add_log(
+                    f"Chapter overrides applied: {len(selected_chapters)} selected.",
+                    level="info",
+                )
+            else:
+                raise ValueError("No chapters were enabled in the requested job.")
+
+        job.metadata_tags = _merge_metadata(extraction.metadata, metadata_overrides)
 
         total_characters = extraction.total_characters or calculate_text_length(extraction.combined_text)
-        if job.total_characters == 0:
-            job.total_characters = total_characters
+        job.total_characters = total_characters
         job.add_log(f"Total characters: {job.total_characters:,}")
 
         _apply_newline_policy(extraction.chapters, job.replace_single_newlines)
@@ -237,9 +375,9 @@ def _select_device() -> str:
 
 
 def _prepare_output_dir(job: Job) -> Path:
-    from platformdirs import user_desktop_dir
+    from platformdirs import user_desktop_dir  # type: ignore[import-not-found]
 
-    default_output = Path(get_user_cache_path("outputs"))
+    default_output = Path(str(get_user_cache_path("outputs")))
     if job.save_mode == "Save to Desktop":
         directory = Path(user_desktop_dir())
     elif job.save_mode == "Save next to input file":

abogen/web/service.py

@@ -6,7 +6,7 @@ import uuid
 from dataclasses import dataclass, field
 from enum import Enum
 from pathlib import Path
-from typing import Callable, Dict, Iterable, List, Optional
+from typing import Any, Callable, Dict, Iterable, List, Optional, Mapping
 
 
 class JobStatus(str, Enum):
@@ -64,7 +64,7 @@ class Job:
     logs: List[JobLog] = field(default_factory=list)
     error: Optional[str] = None
     result: JobResult = field(default_factory=JobResult)
-    chapters: List[str] = field(default_factory=list)
+    chapters: List[Dict[str, Any]] = field(default_factory=list)
     queue_position: Optional[int] = None
     cancel_requested: bool = False
 
@@ -100,6 +100,18 @@ class Job:
                 "voice_profile": self.voice_profile,
                 "max_subtitle_words": self.max_subtitle_words,
             },
+            "metadata_tags": dict(self.metadata_tags),
+            "chapters": [
+                {
+                    "id": entry.get("id"),
+                    "index": entry.get("index"),
+                    "order": entry.get("order"),
+                    "title": entry.get("title"),
+                    "enabled": bool(entry.get("enabled", True)),
+                    "characters": len(str(entry.get("text", ""))),
+                }
+                for entry in self.chapters
+            ],
         }
 
 
@@ -149,7 +161,7 @@ class ConversionService:
         replace_single_newlines: bool,
         subtitle_format: str,
         total_characters: int,
-        chapters: Optional[Iterable[str]] = None,
+        chapters: Optional[Iterable[Any]] = None,
         save_chapters_separately: bool = False,
         merge_chapters_at_end: bool = True,
         separate_chapters_format: str = "wav",
@@ -157,8 +169,13 @@ class ConversionService:
         save_as_project: bool = False,
         voice_profile: Optional[str] = None,
         max_subtitle_words: int = 50,
+        metadata_tags: Optional[Mapping[str, Any]] = None,
     ) -> Job:
         job_id = uuid.uuid4().hex
+        normalized_metadata = self._normalize_metadata_tags(metadata_tags)
+        normalized_chapters = self._normalize_chapters(chapters)
+        if total_characters <= 0 and normalized_chapters:
+            total_characters = sum(len(str(entry.get("text", ""))) for entry in normalized_chapters)
         job = Job(
             id=job_id,
             original_filename=original_filename,
@@ -180,9 +197,10 @@ class ConversionService:
             save_as_project=save_as_project,
             voice_profile=voice_profile,
             max_subtitle_words=max_subtitle_words,
+            metadata_tags=normalized_metadata,
             created_at=time.time(),
             total_characters=total_characters,
-            chapters=list(chapters or []),
+            chapters=normalized_chapters,
         )
         with self._lock:
             self._jobs[job_id] = job
@@ -322,6 +340,128 @@ class ConversionService:
             if job:
                 job.queue_position = index
 
+    @staticmethod
+    def _coerce_bool(value: Any, default: bool = True) -> bool:
+        if isinstance(value, bool):
+            return value
+        if isinstance(value, str):
+            lowered = value.strip().lower()
+            if lowered in {"true", "1", "yes", "on"}:
+                return True
+            if lowered in {"false", "0", "no", "off"}:
+                return False
+            return default
+        if value is None:
+            return default
+        return bool(value)
+
+    @staticmethod
+    def _coerce_optional_int(value: Any) -> Optional[int]:
+        if value is None:
+            return None
+        try:
+            return int(value)
+        except (TypeError, ValueError):
+            return None
+
+    @staticmethod
+    def _normalize_metadata_tags(values: Optional[Mapping[str, Any]]) -> Dict[str, str]:
+        if not values:
+            return {}
+        normalized: Dict[str, str] = {}
+        for key, raw_value in values.items():
+            if raw_value is None:
+                continue
+            key_str = str(key).strip()
+            if not key_str:
+                continue
+            normalized[key_str] = str(raw_value)
+        return normalized
+
+    @classmethod
+    def _normalize_chapters(cls, chapters: Optional[Iterable[Any]]) -> List[Dict[str, Any]]:
+        if not chapters:
+            return []
+
+        normalized: List[Dict[str, Any]] = []
+        for order, raw in enumerate(chapters):
+            if raw is None:
+                continue
+
+            if isinstance(raw, str):
+                raw_dict: Dict[str, Any] = {"title": raw}
+            elif isinstance(raw, dict):
+                raw_dict = dict(raw)
+            else:
+                continue
+
+            entry: Dict[str, Any] = {}
+
+            id_value = raw_dict.get("id") or raw_dict.get("chapter_id") or raw_dict.get("key")
+            if id_value is not None:
+                entry["id"] = str(id_value)
+
+            index_value = (
+                cls._coerce_optional_int(raw_dict.get("index"))
+                or cls._coerce_optional_int(raw_dict.get("original_index"))
+                or cls._coerce_optional_int(raw_dict.get("source_index"))
+                or cls._coerce_optional_int(raw_dict.get("chapter_index"))
+            )
+            if index_value is not None:
+                entry["index"] = index_value
+
+            order_value = (
+                cls._coerce_optional_int(raw_dict.get("order"))
+                or cls._coerce_optional_int(raw_dict.get("position"))
+                or cls._coerce_optional_int(raw_dict.get("sort"))
+                or cls._coerce_optional_int(raw_dict.get("sort_order"))
+            )
+            entry["order"] = order_value if order_value is not None else order
+
+            source_title = (
+                raw_dict.get("source_title")
+                or raw_dict.get("original_title")
+                or raw_dict.get("base_title")
+            )
+            if source_title:
+                entry["source_title"] = str(source_title)
+
+            title_value = (
+                raw_dict.get("title")
+                or raw_dict.get("name")
+                or raw_dict.get("label")
+                or raw_dict.get("chapter")
+            )
+            if title_value is not None:
+                entry["title"] = str(title_value)
+            elif source_title:
+                entry["title"] = str(source_title)
+            else:
+                entry["title"] = f"Chapter {order + 1}"
+
+            text_value = raw_dict.get("text")
+            if text_value is None:
+                text_value = raw_dict.get("content") or raw_dict.get("body") or raw_dict.get("value")
+            if text_value is not None:
+                entry["text"] = str(text_value)
+
+            enabled = cls._coerce_bool(
+                raw_dict.get("enabled", raw_dict.get("include", raw_dict.get("selected", True))),
+                True,
+            )
+            if "disabled" in raw_dict and cls._coerce_bool(raw_dict.get("disabled"), False):
+                enabled = False
+            entry["enabled"] = enabled
+
+            metadata_payload = raw_dict.get("metadata") or raw_dict.get("metadata_tags")
+            normalized_metadata = cls._normalize_metadata_tags(metadata_payload)
+            if normalized_metadata:
+                entry["metadata"] = normalized_metadata
+
+            normalized.append(entry)
+
+        return normalized
+
 
 def default_storage_root() -> Path:
     base = Path.cwd()