diff --git a/abogen/domain/metadata_extraction.py b/abogen/domain/metadata_extraction.py new file mode 100644 index 0000000..1383b09 --- /dev/null +++ b/abogen/domain/metadata_extraction.py @@ -0,0 +1,191 @@ +"""Metadata extraction and processing utilities. + +This module provides functions for extracting metadata from text content +and generating ffmpeg metadata arguments. +""" + +from __future__ import annotations + +import datetime +import os +import re +from pathlib import Path +from typing import Dict, List, Optional, Tuple + + +def extract_metadata_from_text(text: str) -> Dict[str, Optional[str]]: + """Extract metadata tags from text content. + + Looks for tags in format: <> + + Supported tags: + - TITLE, ARTIST, ALBUM, YEAR + - ALBUM_ARTIST, COMPOSER, GENRE + - COVER_PATH + + Args: + text: Text content to search for metadata tags. + + Returns: + Dictionary with extracted metadata values (None if not found). + """ + metadata = {} + + patterns = { + "title": r"<]*)>>", + "artist": r"<]*)>>", + "album": r"<]*)>>", + "year": r"<]*)>>", + "album_artist": r"<]*)>>", + "composer": r"<]*)>>", + "genre": r"<]*)>>", + "cover_path": r"<]*)>>", + } + + for key, pattern in patterns.items(): + match = re.search(pattern, text) + if match: + metadata[key] = match.group(1).strip() + else: + metadata[key] = None + + return metadata + + +def get_filename_from_path( + file_path: str, + display_path: Optional[str] = None, + from_queue: bool = False, +) -> str: + """Extract filename (without extension) from path. + + Args: + file_path: The file path to extract from. + display_path: Optional display path (used if from_queue is False). + from_queue: Whether the file is from queue. + + Returns: + Filename without extension. + """ + if from_queue: + base_path = file_path + else: + base_path = display_path if display_path else file_path + + filename = os.path.splitext(os.path.basename(base_path))[0] + return filename + + +def build_ffmpeg_metadata_args( + metadata: Dict[str, Optional[str]], + filename: str, +) -> List[str]: + """Build ffmpeg metadata arguments from metadata dictionary. + + Args: + metadata: Dictionary with metadata keys and values. + filename: Fallback filename for title/album if not specified. + + Returns: + List of ffmpeg metadata arguments. + """ + args = [] + + # Default values + defaults = { + "title": filename, + "artist": "Unknown", + "album": filename, + "date": str(datetime.datetime.now().year), + "album_artist": "Unknown", + "composer": "Narrator", + "genre": "Audiobook", + } + + # Map of metadata keys to ffmpeg metadata keys + key_mapping = { + "title": "title", + "artist": "artist", + "album": "album", + "year": "date", # year -> date for ffmpeg + "album_artist": "album_artist", + "composer": "composer", + "genre": "genre", + } + + for metadata_key, ffmpeg_key in key_mapping.items(): + value = metadata.get(metadata_key) + if value is None: + value = defaults.get(metadata_key, "") + if value: + args.extend(["-metadata", f"{ffmpeg_key}={value}"]) + + return args + + +def extract_metadata_and_build_args( + text: str, + filename: str, + display_path: Optional[str] = None, + from_queue: bool = False, +) -> Tuple[List[str], Optional[str]]: + """Extract metadata from text and build ffmpeg arguments. + + Convenience function that combines extract_metadata_from_text and + build_ffmpeg_metadata_args. + + Args: + text: Text content to search for metadata tags. + filename: Fallback filename for title/album. + display_path: Optional display path. + from_queue: Whether the file is from queue. + + Returns: + Tuple of (ffmpeg_metadata_args, cover_path). + """ + metadata = extract_metadata_from_text(text) + cover_path = metadata.get("cover_path") + + # Get actual filename from path + actual_filename = get_filename_from_path( + file_path=filename, + display_path=display_path, + from_queue=from_queue, + ) + + args = build_ffmpeg_metadata_args(metadata, actual_filename) + return args, cover_path + + +def read_text_for_metadata( + file_path: str, + is_direct_text: bool, + direct_text: Optional[str] = None, + encoding: Optional[str] = None, +) -> str: + """Read text content for metadata extraction. + + Args: + file_path: Path to file (or text if is_direct_text). + is_direct_text: Whether file_path contains direct text. + direct_text: Optional direct text (used if is_direct_text). + encoding: File encoding (detected if not provided). + + Returns: + Text content for metadata extraction. + """ + if is_direct_text: + return direct_text or file_path + + # Read from file + actual_path = direct_text if direct_text else file_path + + try: + if encoding is None: + from abogen.utils import detect_encoding + encoding = detect_encoding(actual_path) + + with open(actual_path, "r", encoding=encoding, errors="replace") as f: + return f.read() + except Exception: + return "" diff --git a/abogen/pyqt/conversion.py b/abogen/pyqt/conversion.py index e28b0f0..caf30c1 100644 --- a/abogen/pyqt/conversion.py +++ b/abogen/pyqt/conversion.py @@ -37,6 +37,10 @@ from abogen.domain.audio_buffer import ( ) from abogen.domain.subtitle_generation import process_subtitle_tokens from abogen.domain.voice_loader import load_voice_cached +from abogen.domain.metadata_extraction import ( + extract_metadata_and_build_args, + read_text_for_metadata, +) import abogen.hf_tracker as hf_tracker import static_ffmpeg import threading # for efficient waiting @@ -1890,99 +1894,37 @@ class ConversionThread(QThread): def _extract_and_add_metadata_tags_to_ffmpeg_cmd(self): """Extract metadata tags from text content and add them to ffmpeg command""" - metadata_options = [] - - # Get the input text (either direct or from file) - text = "" - if self.is_direct_text: - text = self.file_name - else: - try: - encoding = detect_encoding(self.file_name) - with open( - self.file_name, "r", encoding=encoding, errors="replace" - ) as file: - text = file.read() - except Exception as e: - self.log_updated.emit( - f"Warning: Could not read file for metadata extraction: {e}" - ) - return [] - - # Extract metadata tags using regex - title_match = re.search(r"<]*)>>", text) - artist_match = re.search(r"<]*)>>", text) - album_match = re.search(r"<]*)>>", text) - year_match = re.search(r"<]*)>>", text) - album_artist_match = re.search(r"<]*)>>", text) - composer_match = re.search(r"<]*)>>", text) - genre_match = re.search(r"<]*)>>", text) - cover_match = re.search(r"<]*)>>", text) - cover_path = cover_match.group(1) if cover_match else None - - # Use display path or filename as fallback for title - - # Use file_name for logs if from_queue, otherwise use display_path if available - if getattr(self, "from_queue", False): - filename = os.path.splitext(os.path.basename(self.file_name))[0] - else: - filename = os.path.splitext( - os.path.basename( - self.display_path if self.display_path else self.file_name - ) - )[0] - - if title_match: - metadata_options.extend(["-metadata", f"title={title_match.group(1)}"]) - else: - metadata_options.extend(["-metadata", f"title={filename}"]) - - # Add artist metadata - if artist_match: - metadata_options.extend(["-metadata", f"artist={artist_match.group(1)}"]) - else: - metadata_options.extend(["-metadata", f"artist=Unknown"]) - - # Add album metadata - if album_match: - metadata_options.extend(["-metadata", f"album={album_match.group(1)}"]) - else: - metadata_options.extend(["-metadata", f"album={filename}"]) - - # Add year metadata - if year_match: - metadata_options.extend(["-metadata", f"date={year_match.group(1)}"]) - else: - # Use current year if year is not specified - import datetime - - current_year = datetime.datetime.now().year - metadata_options.extend(["-metadata", f"date={current_year}"]) - - # Add album artist metadata - if album_artist_match: - metadata_options.extend( - ["-metadata", f"album_artist={album_artist_match.group(1)}"] + # 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: + self.log_updated.emit( + ("Warning: Could not read file for metadata extraction", "orange") ) - else: - metadata_options.extend(["-metadata", f"album_artist=Unknown"]) - - # Add composer metadata - if composer_match: - metadata_options.extend( - ["-metadata", f"composer={composer_match.group(1)}"] + 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, + display_path=getattr(self, "display_path", None), + from_queue=getattr(self, "from_queue", False), ) - else: - metadata_options.extend(["-metadata", f"composer=Narrator"]) - - # Add genre metadata - if genre_match: - metadata_options.extend(["-metadata", f"genre={genre_match.group(1)}"]) - else: - metadata_options.extend(["-metadata", f"genre=Audiobook"]) - - # Add these to ffmpeg command - return metadata_options, cover_path + return metadata_options, cover_path + except Exception as e: + self.log_updated.emit( + (f"Warning: Metadata extraction error: {e}", "orange") + ) + return [], None def _process_subtitle_tokens( self, diff --git a/abogen/webui/conversion_runner.py b/abogen/webui/conversion_runner.py index 42dbfc2..5dcf912 100644 --- a/abogen/webui/conversion_runner.py +++ b/abogen/webui/conversion_runner.py @@ -114,7 +114,6 @@ from abogen.domain.device import select_device as _select_device from abogen.domain.audio_helpers import ( build_ffmpeg_command as _build_ffmpeg_command, to_float32 as _to_float32, - apply_m4b_chapters_with_mutagen as _apply_m4b_chapters_with_mutagen, ) from abogen.domain.audio_buffer import ( create_silence as _create_silence, @@ -144,112 +143,6 @@ class AudioSink: _APOSTROPHE_CONFIG = ApostropheConfig() -def _apply_m4b_chapters_with_mutagen( - audio_path: Path, - chapters: List[Dict[str, Any]], - job: Job, -) -> bool: - try: - return _apply_m4b_chapters_with_mutagen(audio_path, chapters) - except ImportError: - job.add_log( - "Unable to write MP4 chapter atoms because mutagen is not installed.", - level="warning", - ) - return False - except Exception as exc: - job.add_log(f"Failed to write MP4 chapter atoms: {exc}", level="warning") - return False - - -def _embed_m4b_metadata( - audio_path: Path, - metadata_payload: Dict[str, Any], - job: Job, -) -> None: - metadata_map = dict(metadata_payload.get("metadata") or {}) - chapter_entries = list(metadata_payload.get("chapters") or []) - ffmetadata_path = _export_svc.write_ffmetadata_file(audio_path, metadata_map, chapter_entries) - cover_path: Optional[Path] = None - if job.cover_image_path: - candidate = Path(job.cover_image_path) - if candidate.exists(): - cover_path = candidate - - metadata_args = _export_svc._metadata_to_ffmpeg_args(metadata_map) - - if not ffmetadata_path and not cover_path and not metadata_args: - return - - job.add_log("Embedding metadata into m4b output") - - command: List[str] = ["ffmpeg", "-y", "-i", str(audio_path)] - metadata_index: Optional[int] = None - cover_index: Optional[int] = None - next_index = 1 - - if ffmetadata_path: - command += ["-f", "ffmetadata", "-i", str(ffmetadata_path)] - metadata_index = next_index - next_index += 1 - - if cover_path: - command += ["-i", str(cover_path)] - cover_index = next_index - next_index += 1 - - command += ["-map", "0:a"] - command += ["-c:a", "copy"] - - if cover_index is not None: - command += ["-map", f"{cover_index}:v:0"] - command += ["-c:v:0", "mjpeg"] - command += ["-disposition:v:0", "attached_pic"] - command += ["-metadata:s:v:0", "title=Cover Art"] - if job.cover_image_mime: - command += ["-metadata:s:v:0", f"mimetype={job.cover_image_mime}"] - - if metadata_index is not None: - command += ["-map_metadata", str(metadata_index)] - command += ["-map_chapters", str(metadata_index)] - else: - command += ["-map_metadata", "0"] - - if metadata_args: - command.extend(metadata_args) - - command += ["-movflags", "+faststart+use_metadata_tags"] - - temp_output = audio_path.with_suffix(audio_path.suffix + ".tmp") - if audio_path.suffix.lower() in {".m4b", ".mp4", ".m4a"}: - command += ["-f", "mp4"] - command.append(str(temp_output)) - - process = create_process(command, text=True) - try: - return_code = process.wait() - finally: - if ffmetadata_path and ffmetadata_path.exists(): - try: - ffmetadata_path.unlink() - except OSError: - pass - - if return_code != 0: - if temp_output.exists(): - temp_output.unlink(missing_ok=True) - raise RuntimeError(f"ffmpeg failed to embed metadata (exit code {return_code})") - - temp_output.replace(audio_path) - job.add_log("Embedded metadata and chapters into m4b output", level="info") - - mutagen_applied = _apply_m4b_chapters_with_mutagen(audio_path, chapter_entries, job) - if mutagen_applied: - job.add_log( - f"Applied {len(chapter_entries)} chapter markers via mutagen", level="info" - ) - - def run_conversion_job(job: Job) -> None: job.add_log("Preparing conversion pipeline") canceller = _make_canceller(job) @@ -1141,7 +1034,20 @@ def run_conversion_job(job: Job) -> None: and job.status not in {JobStatus.FAILED, JobStatus.CANCELLED} ): try: - _embed_m4b_metadata(audio_output_path, metadata_payload, job) + cover_path = None + if job.cover_image_path: + candidate = Path(job.cover_image_path) + if candidate.exists(): + cover_path = candidate + + _export_svc.embed_m4b_metadata( + audio_path=audio_output_path, + metadata=metadata_payload.get("metadata") or {}, + chapters=metadata_payload.get("chapters") or [], + cover_path=cover_path, + cover_mime=job.cover_image_mime, + log_callback=lambda msg, level="info": job.add_log(msg, level=level), + ) except Exception as exc: # pragma: no cover - ensure failure propagates job.add_log( f"Failed to embed metadata into m4b output: {exc}", diff --git a/tests/test_metadata_extraction.py b/tests/test_metadata_extraction.py new file mode 100644 index 0000000..2b3383e --- /dev/null +++ b/tests/test_metadata_extraction.py @@ -0,0 +1,199 @@ +"""Tests for abogen.domain.metadata_extraction module.""" + +import pytest + +from abogen.domain.metadata_extraction import ( + extract_metadata_from_text, + get_filename_from_path, + build_ffmpeg_metadata_args, + extract_metadata_and_build_args, + read_text_for_metadata, +) + + +class TestExtractMetadataFromText: + """Tests for extract_metadata_from_text function.""" + + def test_extract_all_metadata(self): + """Test extracting all metadata tags.""" + text = """ + <> + <> + <> + <> + <> + <> + <> + <> + """ + metadata = extract_metadata_from_text(text) + + assert metadata["title"] == "Test Book" + assert metadata["artist"] == "Test Author" + assert metadata["album"] == "Test Album" + assert metadata["year"] == "2024" + assert metadata["album_artist"] == "Album Artist" + assert metadata["composer"] == "Composer Name" + assert metadata["genre"] == "Fiction" + assert metadata["cover_path"] == "/path/to/cover.jpg" + + def test_extract_partial_metadata(self): + """Test extracting partial metadata.""" + text = "<>" + metadata = extract_metadata_from_text(text) + + assert metadata["title"] == "Only Title" + assert metadata["artist"] is None + assert metadata["cover_path"] is None + + def test_empty_text(self): + """Test extracting from empty text.""" + metadata = extract_metadata_from_text("") + + for key in metadata: + assert metadata[key] is None + + def test_no_tags(self): + """Test text without metadata tags.""" + text = "This is just regular text without any metadata tags." + metadata = extract_metadata_from_text(text) + + for key in metadata: + assert metadata[key] is None + + def test_strip_whitespace(self): + """Test that values are stripped of whitespace.""" + text = "<>" + metadata = extract_metadata_from_text(text) + + assert metadata["title"] == "Test Book" + + +class TestGetFilenameFromPath: + """Tests for get_filename_from_path function.""" + + def test_simple_path(self): + """Test extracting filename from simple path.""" + filename = get_filename_from_path("/path/to/file.txt") + assert filename == "file" + + def test_path_with_multiple_extensions(self): + """Test extracting filename from path with multiple extensions.""" + filename = get_filename_from_path("/path/to/file.tar.gz") + assert filename == "file.tar" + + def test_windows_path(self): + """Test extracting filename from Windows path.""" + # Note: This test may behave differently on Windows vs Unix + # but should work correctly on the current platform + filename = get_filename_from_path("C:\\path\\to\\file.txt") + assert "file" in filename + + def test_with_display_path(self): + """Test using display_path when not from_queue.""" + filename = get_filename_from_path( + file_path="/original/path/file.txt", + display_path="/display/path/display_file.txt", + from_queue=False, + ) + assert filename == "display_file" + + def test_with_display_path_from_queue(self): + """Test ignoring display_path when from_queue.""" + filename = get_filename_from_path( + file_path="/original/path/file.txt", + display_path="/display/path/display_file.txt", + from_queue=True, + ) + assert filename == "file" + + +class TestBuildFfmpegMetadataArgs: + """Tests for build_ffmpeg_metadata_args function.""" + + def test_all_metadata_provided(self): + """Test building args with all metadata provided.""" + metadata = { + "title": "Test Title", + "artist": "Test Artist", + "album": "Test Album", + "year": "2024", + "album_artist": "Album Artist", + "composer": "Composer", + "genre": "Fiction", + } + args = build_ffmpeg_metadata_args(metadata, "fallback") + + assert "-metadata" in args + assert "title=Test Title" in args + assert "artist=Test Artist" in args + assert "album=Test Album" in args + assert "date=2024" in args # year -> date + + def test_use_defaults(self): + """Test that defaults are used for missing metadata.""" + metadata = {} # Empty metadata + args = build_ffmpeg_metadata_args(metadata, "mybook") + + # Should use defaults + assert any("title=mybook" in arg for arg in args) + assert any("artist=Unknown" in arg for arg in args) + assert any("genre=Audiobook" in arg for arg in args) + + def test_empty_values_skipped(self): + """Test that empty values are skipped.""" + metadata = { + "title": "Test", + "artist": "", + "album": None, + } + args = build_ffmpeg_metadata_args(metadata, "fallback") + + # Should have title but not artist/album (empty) + assert any("title=Test" in arg for arg in args) + + +class TestExtractMetadataAndBuildArgs: + """Tests for extract_metadata_and_build_args function.""" + + def test_full_workflow(self): + """Test full metadata extraction and arg building.""" + text = "<>\n<>" + args, cover_path = extract_metadata_and_build_args( + text=text, + filename="mybook.txt", + ) + + assert any("title=My Book" in arg for arg in args) + assert any("artist=My Author" in arg for arg in args) + assert cover_path is None + + def test_with_cover_path(self): + """Test extraction with cover path.""" + text = "<>" + args, cover_path = extract_metadata_and_build_args( + text=text, + filename="mybook.txt", + ) + + assert cover_path == "/covers/cover.jpg" + + +class TestReadTextForMetadata: + """Tests for read_text_for_metadata function.""" + + def test_direct_text(self): + """Test reading direct text.""" + text = read_text_for_metadata( + file_path="This is direct text", + is_direct_text=True, + ) + assert text == "This is direct text" + + def test_file_not_found(self): + """Test handling of file not found.""" + text = read_text_for_metadata( + file_path="/nonexistent/file.txt", + is_direct_text=False, + ) + assert text == ""