mirror of
https://github.com/denizsafak/abogen.git
synced 2026-09-20 11:40:57 +02:00
Artem Akymenko
2d18501839
- Add expand_metadata_aliases() for concept fan-out (series→5, author→2, description→2, tags→3 keys) - Replace _normalize_metadata_tags in webui/service.py with normalize_metadata_map from domain - Replace _normalize_metadata in epub3/exporter.py with normalize_metadata_map from domain - Replace manual fan-out in form.py with expand_metadata_aliases() - Rewrite metadata_overrides.py to use expand_metadata_aliases() - Remove unused normalize_metadata_casefold import from exporters.py - 19 new tests for expand_metadata_aliases (1528 total)
325 lines
11 KiB
Python
325 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import tempfile
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional, Mapping, Sequence
|
|
|
|
import static_ffmpeg
|
|
|
|
from abogen.domain.metadata_helpers import (
|
|
split_people_field,
|
|
split_simple_list,
|
|
first_nonempty,
|
|
extract_year,
|
|
normalize_series_sequence,
|
|
_SERIES_SEQUENCE_TAG_KEYS,
|
|
)
|
|
from abogen.epub3.exporter import build_epub3_package
|
|
from abogen.utils import create_process
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class ExportConfig:
|
|
"""Configuration for export operations."""
|
|
ffmpeg_path: str = "ffmpeg"
|
|
verify_ssl: bool = True
|
|
|
|
|
|
class ExportService:
|
|
"""Unified service for audiobook exports (M4B, FFMETADATA, EPUB3, Audiobookshelf)."""
|
|
|
|
def __init__(self, config: Optional[ExportConfig] = None):
|
|
self.config = config or ExportConfig()
|
|
static_ffmpeg.add_paths()
|
|
|
|
# ----------------------------------------------------------------------
|
|
# FFMETADATA
|
|
# ----------------------------------------------------------------------
|
|
|
|
def render_ffmetadata(
|
|
self,
|
|
metadata: Dict[str, Any],
|
|
chapters: List[Dict[str, Any]],
|
|
) -> str:
|
|
"""Render FFMETADATA content."""
|
|
lines = [";FFMETADATA1"]
|
|
|
|
for key, value in (metadata or {}).items():
|
|
if value is None:
|
|
continue
|
|
key_str = str(key).strip()
|
|
if not key_str:
|
|
continue
|
|
lines.append(f"{key_str}={self._escape_ffmetadata_value(value)}")
|
|
|
|
for chapter in chapters or []:
|
|
start = chapter.get("start")
|
|
end = chapter.get("end")
|
|
if start is None or end is None:
|
|
continue
|
|
try:
|
|
start_ms = max(0, int(round(float(start) * 1000)))
|
|
end_ms = int(round(float(end) * 1000))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if end_ms <= start_ms:
|
|
end_ms = start_ms + 1
|
|
lines.append("[CHAPTER]")
|
|
lines.append("TIMEBASE=1/1000")
|
|
lines.append(f"START={start_ms}")
|
|
lines.append(f"END={end_ms}")
|
|
title = chapter.get("title")
|
|
if title:
|
|
lines.append(f"title={self._escape_ffmetadata_value(title)}")
|
|
voices = chapter.get("voices")
|
|
if voices and isinstance(voices, list):
|
|
voice_str = ", ".join(
|
|
f"{v.get('voice', '')}@{v.get('provider', '')}"
|
|
for v in voices if v.get("voice")
|
|
)
|
|
if voice_str:
|
|
lines.append(f"voice={self._escape_ffmetadata_value(voice_str)}")
|
|
|
|
return "\n".join(lines) + "\n"
|
|
|
|
@staticmethod
|
|
def _escape_ffmetadata_value(value: Any) -> str:
|
|
escaped = str(value).replace("\\", "\\\\").replace("\n", "\\n")
|
|
escaped = escaped.replace("=", "\\=").replace(";", "\\;").replace("#", "\\#")
|
|
return escaped
|
|
|
|
def write_ffmetadata_file(
|
|
self,
|
|
audio_path: Path,
|
|
metadata: Dict[str, Any],
|
|
chapters: List[Dict[str, Any]],
|
|
) -> Optional[Path]:
|
|
"""Write FFMETADATA file to temp location."""
|
|
content = self.render_ffmetadata(metadata, chapters)
|
|
if content.strip() == ";FFMETADATA1":
|
|
return None
|
|
|
|
directory = audio_path.parent if audio_path.parent.exists() else Path(tempfile.gettempdir())
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w",
|
|
encoding="utf-8",
|
|
suffix=".ffmeta",
|
|
delete=False,
|
|
dir=str(directory),
|
|
) as handle:
|
|
handle.write(content)
|
|
return Path(handle.name)
|
|
|
|
# ----------------------------------------------------------------------
|
|
# M4B Export
|
|
# ----------------------------------------------------------------------
|
|
|
|
def embed_m4b_metadata(
|
|
self,
|
|
audio_path: Path,
|
|
metadata: Dict[str, Any],
|
|
chapters: List[Dict[str, Any]],
|
|
cover: "CoverConfig | None" = None,
|
|
cover_path: Optional[Path] = None,
|
|
cover_mime: Optional[str] = None,
|
|
log_callback: Optional[callable] = None,
|
|
) -> None:
|
|
"""Embed metadata and chapters into M4B file using FFmpeg + Mutagen."""
|
|
from abogen.domain.config_types import CoverConfig
|
|
if isinstance(cover, CoverConfig):
|
|
cover_path = cover.path
|
|
cover_mime = cover.mime
|
|
ffmetadata_path = self.write_ffmetadata_file(audio_path, metadata, chapters)
|
|
|
|
metadata_args = self._metadata_to_ffmpeg_args(metadata)
|
|
|
|
cmd = ["ffmpeg", "-y", "-i", str(audio_path)]
|
|
|
|
if ffmetadata_path:
|
|
cmd.extend(["-f", "ffmetadata", "-i", str(ffmetadata_path)])
|
|
|
|
if cover_path and cover_path.exists():
|
|
cmd.extend(["-i", str(cover_path)])
|
|
cmd.extend(["-map", "0:a"])
|
|
cmd.extend(["-map", "1:v:0", "-c:v:0", "mjpeg", "-disposition:v:0", "attached_pic"])
|
|
if cover_mime:
|
|
cmd.extend(["-metadata:s:v:0", f"mimetype={cover_mime}"])
|
|
cmd.extend(["-metadata:s:v:0", "title=Cover Art"])
|
|
else:
|
|
cmd.extend(["-map", "0:a"])
|
|
|
|
cmd.extend(["-c:a", "copy"])
|
|
|
|
if ffmetadata_path:
|
|
cmd.extend(["-map_metadata", "1", "-map_chapters", "1"])
|
|
else:
|
|
cmd.extend(["-map_metadata", "0"])
|
|
|
|
if metadata_args:
|
|
cmd.extend(metadata_args)
|
|
|
|
cmd.extend(["-movflags", "+faststart+use_metadata_tags"])
|
|
|
|
temp_output = audio_path.with_suffix(audio_path.suffix + ".tmp")
|
|
if audio_path.suffix.lower() in {".m4b", ".mp4", ".m4a"}:
|
|
cmd.extend(["-f", "mp4"])
|
|
cmd.append(str(temp_output))
|
|
|
|
if log_callback:
|
|
log_callback("Embedding metadata into M4B output")
|
|
|
|
process = create_process(cmd, text=True)
|
|
return_code = process.wait()
|
|
|
|
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)
|
|
|
|
if log_callback:
|
|
log_callback("Embedded metadata and chapters into M4B output", "info")
|
|
|
|
# Apply chapters via Mutagen for better compatibility
|
|
self._apply_m4b_chapters_mutagen(audio_path, chapters, log_callback)
|
|
|
|
@staticmethod
|
|
def _metadata_to_ffmpeg_args(metadata: Dict[str, Any]) -> List[str]:
|
|
args = []
|
|
for key, value in (metadata or {}).items():
|
|
if value in (None, ""):
|
|
continue
|
|
key_str = str(key).strip()
|
|
if not key_str:
|
|
continue
|
|
normalized_key = key_str.lower()
|
|
if normalized_key == "year":
|
|
ffmpeg_key = "date"
|
|
else:
|
|
ffmpeg_key = key_str
|
|
args.extend(["-metadata", f"{ffmpeg_key}={value}"])
|
|
return args
|
|
|
|
def _apply_m4b_chapters_mutagen(
|
|
self,
|
|
audio_path: Path,
|
|
chapters: List[Dict[str, Any]],
|
|
log_callback: Optional[callable] = None,
|
|
) -> bool:
|
|
"""Apply chapter atoms using Mutagen."""
|
|
if not chapters:
|
|
return False
|
|
|
|
try:
|
|
from fractions import Fraction
|
|
from mutagen.mp4 import MP4, MP4Chapter
|
|
except ImportError:
|
|
if log_callback:
|
|
log_callback("Unable to write MP4 chapter atoms because mutagen is not installed.", "warning")
|
|
return False
|
|
|
|
try:
|
|
mp4 = MP4(str(audio_path))
|
|
except Exception as exc:
|
|
if log_callback:
|
|
log_callback(f"Failed to open m4b for chapter embedding: {exc}", "warning")
|
|
return False
|
|
|
|
chapter_objects = []
|
|
for index, entry in enumerate(sorted(chapters, key=lambda item: float(item.get("start") or 0.0))):
|
|
start_raw = entry.get("start")
|
|
if start_raw is None:
|
|
continue
|
|
try:
|
|
start_seconds = max(0.0, float(start_raw))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
|
|
title_value = entry.get("title")
|
|
title_text = str(title_value) if title_value else f"Chapter {index + 1}"
|
|
|
|
start_fraction = Fraction(int(round(start_seconds * 1000)), 1000)
|
|
chapter_atom = MP4Chapter(start_fraction, title_text)
|
|
|
|
end_raw = entry.get("end")
|
|
if end_raw is not None:
|
|
try:
|
|
end_seconds = float(end_raw)
|
|
except (TypeError, ValueError):
|
|
end_seconds = None
|
|
if end_seconds is not None and end_seconds > start_seconds:
|
|
chapter_atom.end = Fraction(int(round(end_seconds * 1000)), 1000)
|
|
|
|
chapter_objects.append(chapter_atom)
|
|
|
|
if not chapter_objects:
|
|
return False
|
|
|
|
try:
|
|
mp4.chapters = chapter_objects
|
|
mp4.save()
|
|
except Exception as exc:
|
|
if log_callback:
|
|
log_callback(f"Failed to persist MP4 chapter atoms: {exc}", "warning")
|
|
return False
|
|
|
|
if log_callback:
|
|
log_callback(f"Applied {len(chapter_objects)} chapter markers via mutagen", "info")
|
|
return True
|
|
|
|
# ----------------------------------------------------------------------
|
|
# EPUB3 Export
|
|
# ----------------------------------------------------------------------
|
|
|
|
def export_epub3(
|
|
self,
|
|
output_path: Path,
|
|
book_id: str,
|
|
extraction: Any, # ExtractionResult
|
|
metadata_tags: Dict[str, Any],
|
|
chapter_markers: Sequence[Dict[str, Any]],
|
|
chunk_markers: Sequence[Dict[str, Any]],
|
|
chunks: Iterable[Dict[str, Any]],
|
|
audio_path: Path,
|
|
speaker_mode: str = "single",
|
|
cover_path: Optional[Path] = None,
|
|
cover_mime: Optional[str] = None,
|
|
) -> Path:
|
|
"""Export EPUB3 with media overlays."""
|
|
return build_epub3_package(
|
|
output_path=output_path,
|
|
book_id=book_id,
|
|
extraction=extraction,
|
|
metadata_tags=metadata_tags,
|
|
chapter_markers=chapter_markers,
|
|
chunk_markers=chunk_markers,
|
|
chunks=chunks,
|
|
audio_path=audio_path,
|
|
speaker_mode=speaker_mode,
|
|
cover_image_path=cover_path,
|
|
cover_image_mime=cover_mime,
|
|
)
|
|
|
|
|
|
if value is None:
|
|
return default
|
|
return bool(value)
|
|
|
|
|
|
__all__ = [
|
|
"ExportConfig",
|
|
"ExportService",
|
|
]
|