refactor: add build_metadata_payload() to domain, unify metadata assembly in both UIs

abogen/domain/metadata_helpers.py

@@ -403,3 +403,40 @@ def load_audiobookshelf_chapters(
         if title and start is not None and end is not None:
             cleaned.append({"title": str(title), "start": start, "end": end})
     return cleaned or None
+
+
+def build_metadata_payload(
+    metadata: Optional[Dict[str, Any]] = None,
+    chapter_markers: Optional[List[Dict[str, Any]]] = None,
+    chunk_markers: Optional[List[Dict[str, Any]]] = None,
+    chunk_level: Optional[str] = None,
+    speaker_mode: Optional[str] = None,
+    speakers: Optional[Dict[str, Any]] = None,
+    generate_epub3: bool = False,
+) -> Dict[str, Any]:
+    """Build the canonical metadata payload dict for persistence and downstream use.
+
+    This is the single source of truth for metadata assembly. Both PyQt and WebUI
+    runners should call this instead of building the dict manually.
+
+    Args:
+        metadata: Normalized metadata tags dict.
+        chapter_markers: List of chapter marker dicts with title/start/end.
+        chunk_markers: List of chunk marker dicts.
+        chunk_level: Chunk granularity level (e.g. 'chapter', 'chunk').
+        speaker_mode: Speaker mode ('single', 'multi', etc.).
+        speakers: Speaker profile mapping.
+        generate_epub3: Whether EPUB3 generation is enabled.
+
+    Returns:
+        Complete metadata payload dict.
+    """
+    return {
+        "metadata": dict(metadata or {}),
+        "chapters": chapter_markers or [],
+        "chunks": chunk_markers or [],
+        "chunk_level": chunk_level,
+        "speaker_mode": speaker_mode,
+        "speakers": dict(speakers or {}),
+        "generate_epub3": generate_epub3,
+    }

abogen/pyqt/conversion.py

@@ -46,16 +46,8 @@ from abogen.domain.audio_buffer import (
 )
 from abogen.domain.subtitle_generation import process_subtitle_tokens
 from abogen.domain.voice_loader import VoiceCache, load_voice_cached, resolve_voice
-from abogen.domain.progress import calc_etr_str
-from abogen.domain.normalization import TTSContext
-from abogen.domain.pronunciation import (
-    compile_pronunciation_rules,
-    compile_heteronym_sentence_rules,
-    merge_pronunciation_overrides,
-)
 from abogen.domain.metadata_extraction import (
-    extract_metadata_and_build_args,
-    extract_metadata_for_file,
+    extract_metadata_from_text,
 )
 from abogen.domain.text_chapters import parse_chapters_from_text
 from abogen.infrastructure.exporters import ExportService
@@ -520,6 +512,9 @@ class ConversionThread(QThread):
                 ) as file:
                     text = file.read()
 
+            # Extract metadata BEFORE clean_text strips the tags
+            self._extracted_metadata = extract_metadata_from_text(text)
+
             # Clean up text using utility function
             text = clean_text(text)
 
@@ -545,18 +540,15 @@ class ConversionThread(QThread):
                 )
 
             # --- Compile normalization rules (heteronym + pronunciation) ---
-            from abogen.domain.normalization import TTSContext
-            pronunciation_overrides = merge_pronunciation_overrides(
-                getattr(self, "pronunciation_overrides", None),
-                getattr(self, "manual_overrides", None),
-            )
-            self._tts_context = TTSContext(
-                split_pattern=self.split_pattern,
-                pronunciation_rules=compile_pronunciation_rules(pronunciation_overrides),
-                heteronym_rules=compile_heteronym_sentence_rules(
-                    getattr(self, "heteronym_overrides", None)
-                ),
+            from abogen.domain.normalization import build_tts_context
+            self._tts_context = build_tts_context(
+                language=self.lang_code,
+                subtitle_mode=self.subtitle_mode,
+                pronunciation_overrides=getattr(self, "pronunciation_overrides", None),
+                manual_overrides=getattr(self, "manual_overrides", None),
+                heteronym_overrides=getattr(self, "heteronym_overrides", None),
                 normalization_overrides=getattr(self, "normalization_overrides", None),
+                log_callback=lambda level, msg: self.log_updated.emit((msg, "grey" if level == "info" else "orange")),
             )
 
             # --- Chapter splitting logic ---
@@ -755,7 +747,7 @@ class ConversionThread(QThread):
             intro_emitted = False
             if merge_chapters_at_end:
                 intro_spec = resolve_intro(
-                    extract_metadata_for_file(self.file_name, self.is_direct_text),
+                    self._extracted_metadata,
                     os.path.basename(self.file_name) if self.file_name else "",
                     getattr(self, "read_title_intro", False),
                     self.voice, self.voice, list(self.voice_cache._cache.keys()),
@@ -1091,7 +1083,7 @@ class ConversionThread(QThread):
             # --- Outro synthesis ---
             if merge_chapters_at_end:
                 outro_spec = resolve_outro(
-                    extract_metadata_for_file(self.file_name, self.is_direct_text),
+                    self._extracted_metadata,
                     os.path.basename(self.file_name) if self.file_name else "",
                     getattr(self, "read_closing_outro", True),
                     self.voice, self.voice, list(self.voice_cache._cache.keys()),
@@ -1132,12 +1124,7 @@ class ConversionThread(QThread):
                     # Add chapters via ExportService (unified with WebUI)
                     if total_chapters > 1:
                         export_svc = ExportService()
-                        metadata_text = read_text_for_metadata(
-                            file_path=self.file_name,
-                            is_direct_text=self.is_direct_text,
-                            direct_text=self.file_name if self.is_direct_text else None,
-                        )
-                        metadata = extract_metadata_from_text(metadata_text) if metadata_text else {}
+                        metadata = dict(getattr(self, "_extracted_metadata", {}))
                         # Convert cover_path from metadata to Path if present
                         cover_path_raw = metadata.pop("cover_path", None)
                         cover_path = Path(cover_path_raw) if cover_path_raw and os.path.exists(cover_path_raw) else None
@@ -1376,33 +1363,28 @@ class ConversionThread(QThread):
             raise ValueError(f"Unsupported output format: {self.output_format}")
 
     def _extract_and_add_metadata_tags_to_ffmpeg_cmd(self):
-        """Extract metadata tags from text content and add them to ffmpeg command"""
-        # Read text for metadata extraction
-        text = read_text_for_metadata(
-            file_path=self.file_name,
-            is_direct_text=self.is_direct_text,
-            direct_text=self.file_name if self.is_direct_text else None,
-        )
-        
-        if not text:
+        """Build ffmpeg metadata args from previously extracted metadata."""
+        metadata = getattr(self, "_extracted_metadata", None)
+        if not metadata or not any(metadata.values()):
             self.log_updated.emit(
-                ("Warning: Could not read file for metadata extraction", "orange")
+                ("Warning: No metadata tags found in text", "orange")
             )
             return [], None
         
-        # Extract metadata and build ffmpeg args
         filename = self.file_name if self.is_direct_text else (
             self.display_path if self.display_path else self.file_name
         )
         
         try:
-            metadata_options, cover_path = extract_metadata_and_build_args(
-                text=text,
-                filename=filename,
+            from abogen.domain.metadata_extraction import build_ffmpeg_metadata_args, get_filename_from_path
+            actual_filename = get_filename_from_path(
+                file_path=filename,
                 display_path=getattr(self, "display_path", None),
                 from_queue=getattr(self, "from_queue", False),
             )
-            return metadata_options, cover_path
+            args = build_ffmpeg_metadata_args(metadata, actual_filename)
+            cover_path = metadata.get("cover_path")
+            return args, cover_path
         except Exception as e:
             self.log_updated.emit(
                 (f"Warning: Metadata extraction error: {e}", "orange")

abogen/webui/conversion_runner.py

@@ -50,6 +50,7 @@ from abogen.domain.metadata_helpers import (
     normalize_series_number as _normalize_series_number,
     extract_series_metadata as _extract_series_metadata,
     format_series_sentence as _format_series_sentence,
+    build_metadata_payload as _build_metadata_payload,
 )
 from abogen.domain.intro_outro import resolve_intro, resolve_outro
 from abogen.domain.title_builder import (
@@ -852,15 +853,15 @@ def run_conversion_job(job: Job) -> None:
         if not audio_path and chapter_paths:
             job.result.audio_path = chapter_paths[0]
 
-        metadata_payload = {
-            "metadata": dict(job.metadata_tags or {}),
-            "chapters": chapter_markers,
-            "chunks": chunk_markers,
-            "chunk_level": job.chunk_level,
-            "speaker_mode": job.speaker_mode,
-            "speakers": dict(getattr(job, "speakers", {}) or {}),
-            "generate_epub3": job.generate_epub3,
-        }
+        metadata_payload = _build_metadata_payload(
+            metadata=job.metadata_tags,
+            chapter_markers=chapter_markers,
+            chunk_markers=chunk_markers,
+            chunk_level=job.chunk_level,
+            speaker_mode=job.speaker_mode,
+            speakers=getattr(job, "speakers", None),
+            generate_epub3=job.generate_epub3,
+        )
 
         if tts_context.usage_counter:
             _record_override_usage(job, tts_context.usage_counter, override_token_map)

tests/test_metadata_helpers.py

@@ -10,6 +10,7 @@ from abogen.domain.metadata_helpers import (
     normalize_series_number,
     extract_series_metadata,
     format_series_sentence,
+    build_metadata_payload,
 )
 
 
@@ -131,3 +132,56 @@ class TestFormatSeriesSentence:
 
     def test_with_the(self):
         assert format_series_sentence("The Lord of the Rings", "1") == "Book 1 of The Lord of the Rings"
+
+
+class TestBuildMetadataPayload:
+    def test_empty(self):
+        result = build_metadata_payload()
+        assert result == {
+            "metadata": {},
+            "chapters": [],
+            "chunks": [],
+            "chunk_level": None,
+            "speaker_mode": None,
+            "speakers": {},
+            "generate_epub3": False,
+        }
+
+    def test_with_metadata(self):
+        result = build_metadata_payload(metadata={"title": "Book", "artist": "Author"})
+        assert result["metadata"] == {"title": "Book", "artist": "Author"}
+
+    def test_metadata_is_copy(self):
+        original = {"title": "Book"}
+        result = build_metadata_payload(metadata=original)
+        result["metadata"]["title"] = "Changed"
+        assert original["title"] == "Book"
+
+    def test_with_chapters(self):
+        chapters = [{"title": "Ch1", "start": 0.0, "end": 10.0}]
+        result = build_metadata_payload(chapter_markers=chapters)
+        assert result["chapters"] == chapters
+
+    def test_with_all_fields(self):
+        result = build_metadata_payload(
+            metadata={"title": "Book"},
+            chapter_markers=[{"title": "Ch1", "start": 0.0, "end": 10.0}],
+            chunk_markers=[{"start": 0.0, "end": 5.0}],
+            chunk_level="chunk",
+            speaker_mode="multi",
+            speakers={"narrator": "M1"},
+            generate_epub3=True,
+        )
+        assert result["metadata"] == {"title": "Book"}
+        assert len(result["chapters"]) == 1
+        assert len(result["chunks"]) == 1
+        assert result["chunk_level"] == "chunk"
+        assert result["speaker_mode"] == "multi"
+        assert result["speakers"] == {"narrator": "M1"}
+        assert result["generate_epub3"] is True
+
+    def test_speakers_is_copy(self):
+        original = {"narrator": "M1"}
+        result = build_metadata_payload(speakers=original)
+        result["speakers"]["narrator"] = "M2"
+        assert original["narrator"] == "M1"