mirror of
https://github.com/denizsafak/abogen.git
synced 2026-07-22 07:10:28 +02:00
refactor: extract metadata processing logic to domain layer
- Replace manual metadata extraction with regex in pyqt/conversion.py with calls to domain/metadata_extraction.py functions - Remove duplicate _embed_m4b_metadata and _apply_m4b_chapters_with_mutagen functions from webui/conversion_runner.py - Use ExportService.embed_m4b_metadata for m4b metadata embedding - Reduce code duplication between PyQt and WebUI interfaces
This commit is contained in:
Artem Akymenko
@@ -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: <<METADATA_KEY:value>>
|
||||
|
||||
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"<<METADATA_TITLE:([^>]*)>>",
|
||||
"artist": r"<<METADATA_ARTIST:([^>]*)>>",
|
||||
"album": r"<<METADATA_ALBUM:([^>]*)>>",
|
||||
"year": r"<<METADATA_YEAR:([^>]*)>>",
|
||||
"album_artist": r"<<METADATA_ALBUM_ARTIST:([^>]*)>>",
|
||||
"composer": r"<<METADATA_COMPOSER:([^>]*)>>",
|
||||
"genre": r"<<METADATA_GENRE:([^>]*)>>",
|
||||
"cover_path": r"<<METADATA_COVER_PATH:([^>]*)>>",
|
||||
}
|
||||
|
||||
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 ""
|
||||
+33
-91
@@ -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"<<METADATA_TITLE:([^>]*)>>", text)
|
||||
artist_match = re.search(r"<<METADATA_ARTIST:([^>]*)>>", text)
|
||||
album_match = re.search(r"<<METADATA_ALBUM:([^>]*)>>", text)
|
||||
year_match = re.search(r"<<METADATA_YEAR:([^>]*)>>", text)
|
||||
album_artist_match = re.search(r"<<METADATA_ALBUM_ARTIST:([^>]*)>>", text)
|
||||
composer_match = re.search(r"<<METADATA_COMPOSER:([^>]*)>>", text)
|
||||
genre_match = re.search(r"<<METADATA_GENRE:([^>]*)>>", text)
|
||||
cover_match = re.search(r"<<METADATA_COVER_PATH:([^>]*)>>", 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,
|
||||
|
||||
@@ -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}",
|
||||
|
||||
Reference in New Issue
Block a user