From 7ed2addb114462b8258808fa00a8a3eb8936e7d9 Mon Sep 17 00:00:00 2001 From: Artem Akymenko Date: Thu, 23 Jul 2026 09:39:02 +0000 Subject: [PATCH] refactor: add build_metadata_payload() to domain, unify metadata assembly in both UIs --- abogen/domain/metadata_helpers.py | 37 +++++++++++++++++ abogen/pyqt/conversion.py | 68 ++++++++++++------------------- abogen/webui/conversion_runner.py | 19 +++++---- tests/test_metadata_helpers.py | 54 ++++++++++++++++++++++++ 4 files changed, 126 insertions(+), 52 deletions(-) diff --git a/abogen/domain/metadata_helpers.py b/abogen/domain/metadata_helpers.py index 87d1756..fe05863 100644 --- a/abogen/domain/metadata_helpers.py +++ b/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, + } diff --git a/abogen/pyqt/conversion.py b/abogen/pyqt/conversion.py index a8e75ba..f186c59 100644 --- a/abogen/pyqt/conversion.py +++ b/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") diff --git a/abogen/webui/conversion_runner.py b/abogen/webui/conversion_runner.py index f9051b9..333b17d 100644 --- a/abogen/webui/conversion_runner.py +++ b/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) diff --git a/tests/test_metadata_helpers.py b/tests/test_metadata_helpers.py index 151a1e5..dd6d962 100644 --- a/tests/test_metadata_helpers.py +++ b/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"