mirror of
https://github.com/denizsafak/abogen.git
synced 2026-07-18 05:40:26 +02:00
refactor: migrate FFmpeg metadata functions to infrastructure/exporters.py
- Extract FFmpeg metadata functions to infrastructure/exporters.py as ExportService - _escape_ffmetadata_value → _escape_ffmetadata_value - _render_ffmetadata → render_ffmetadata - _write_ffmetadata_file → write_ffmetadata_file - _metadata_to_ffmpeg_args → _metadata_to_ffmpeg_args - _apply_m4b_chapters_with_mutagen → _apply_m4b_chapters_mutagen - _embed_m4b_metadata → embed_m4b_metadata - Add tests/test_exporters.py with 28 tests for ExportService - Update tests/test_ffmetadata.py to use ExportService - Update conversion_runner.py to use ExportService - All tests pass with new implementation matching old behavior
This commit is contained in:
Artem Akymenko
@@ -0,0 +1,673 @@
|
||||
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.epub3.exporter import build_epub3_package
|
||||
from abogen.integrations.audiobookshelf import (
|
||||
AudiobookshelfClient,
|
||||
AudiobookshelfConfig,
|
||||
AudiobookshelfUploadError,
|
||||
)
|
||||
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)}")
|
||||
voice = chapter.get("voice")
|
||||
if voice:
|
||||
lines.append(f"voice={self._escape_ffmetadata_value(voice)}")
|
||||
|
||||
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_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."""
|
||||
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,
|
||||
)
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Audiobookshelf Integration
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
def build_audiobookshelf_metadata(self, job: Any) -> Dict[str, Any]:
|
||||
"""Build Audiobookshelf metadata from job."""
|
||||
tags = self._normalize_metadata_casefold(getattr(job, "metadata_tags", {}))
|
||||
filename = Path(getattr(job, "original_filename", "") or "").stem or "Audiobook"
|
||||
|
||||
title = self._first_nonempty(
|
||||
tags.get("title"),
|
||||
tags.get("book_title"),
|
||||
tags.get("name"),
|
||||
tags.get("album"),
|
||||
filename,
|
||||
)
|
||||
authors = self._split_people_field(
|
||||
tags.get("authors")
|
||||
or tags.get("author")
|
||||
or tags.get("album_artist")
|
||||
or tags.get("artist")
|
||||
)
|
||||
narrators = self._split_people_field(tags.get("narrators") or tags.get("narrator"))
|
||||
description = self._first_nonempty(
|
||||
tags.get("description"), tags.get("summary"), tags.get("comment")
|
||||
)
|
||||
genres = self._split_simple_list(tags.get("genre"))
|
||||
keywords = self._split_simple_list(tags.get("tags") or tags.get("keywords"))
|
||||
language = self._first_nonempty(tags.get("language"), tags.get("lang")) or getattr(job, "language", "") or ""
|
||||
series_name = self._first_nonempty(
|
||||
tags.get("series"),
|
||||
tags.get("series_name"),
|
||||
tags.get("seriesname"),
|
||||
tags.get("series_title"),
|
||||
tags.get("seriestitle"),
|
||||
)
|
||||
|
||||
series_sequence = None
|
||||
for key in ("series_index", "series_position", "series_sequence", "series_number", "seriesnumber", "book_number", "booknumber"):
|
||||
raw = tags.get(key)
|
||||
normalized = self._normalize_series_sequence(raw)
|
||||
if normalized:
|
||||
series_sequence = normalized
|
||||
break
|
||||
if not series_name:
|
||||
series_sequence = None
|
||||
|
||||
data = {
|
||||
"title": title,
|
||||
"subtitle": tags.get("subtitle"),
|
||||
"authors": authors,
|
||||
"narrators": narrators,
|
||||
"description": description,
|
||||
"publisher": tags.get("publisher"),
|
||||
"genres": genres,
|
||||
"tags": keywords,
|
||||
"language": language,
|
||||
"publishedYear": self._extract_year(
|
||||
tags.get("published") or tags.get("publication_year") or tags.get("date") or tags.get("year")
|
||||
),
|
||||
"seriesName": series_name,
|
||||
"seriesSequence": series_sequence,
|
||||
"isbn": self._first_nonempty(tags.get("isbn"), tags.get("asin")),
|
||||
}
|
||||
|
||||
published_date = self._first_nonempty(
|
||||
tags.get("published"), tags.get("publication_date"), tags.get("date")
|
||||
)
|
||||
if published_date:
|
||||
data["publishedDate"] = published_date
|
||||
|
||||
rating_text = self._first_nonempty(tags.get("rating"), tags.get("my_rating"))
|
||||
if rating_text:
|
||||
try:
|
||||
data["rating"] = float(str(rating_text).strip())
|
||||
except ValueError:
|
||||
pass
|
||||
rating_max_text = self._first_nonempty(tags.get("rating_max"), tags.get("rating_scale"))
|
||||
if rating_max_text:
|
||||
try:
|
||||
data["ratingMax"] = float(str(rating_max_text).strip())
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# Remove empty values
|
||||
cleaned = {}
|
||||
for key, value in data.items():
|
||||
if value is None:
|
||||
continue
|
||||
if isinstance(value, str) and not value.strip():
|
||||
continue
|
||||
if isinstance(value, (list, tuple)) and not value:
|
||||
continue
|
||||
cleaned[key] = value
|
||||
return cleaned
|
||||
|
||||
def load_audiobookshelf_chapters(self, job: Any) -> Optional[List[Dict[str, Any]]]:
|
||||
"""Load chapters from job artifacts for Audiobookshelf."""
|
||||
metadata_ref = job.result.artifacts.get("metadata") if getattr(job, "result", None) else None
|
||||
if not metadata_ref:
|
||||
return None
|
||||
metadata_path = metadata_ref if isinstance(metadata_ref, Path) else Path(str(metadata_ref))
|
||||
if not metadata_path.exists():
|
||||
return None
|
||||
try:
|
||||
payload = json.loads(metadata_path.read_text(encoding="utf-8"))
|
||||
except (OSError, json.JSONDecodeError):
|
||||
return None
|
||||
chapters = payload.get("chapters")
|
||||
if not isinstance(chapters, list):
|
||||
return None
|
||||
cleaned = []
|
||||
for entry in chapters:
|
||||
if not isinstance(entry, Mapping):
|
||||
continue
|
||||
title = self._first_nonempty(entry.get("title"), entry.get("original_title"))
|
||||
start = entry.get("start")
|
||||
end = entry.get("end")
|
||||
if title is None or not isinstance(start, (int, float)):
|
||||
continue
|
||||
chapter_payload = {"title": title, "start": float(start)}
|
||||
if isinstance(end, (int, float)):
|
||||
chapter_payload["end"] = float(end)
|
||||
cleaned.append(chapter_payload)
|
||||
return cleaned or None
|
||||
|
||||
def upload_audiobookshelf(
|
||||
self,
|
||||
job: Any,
|
||||
audio_path: Path,
|
||||
subtitle_paths: List[Path],
|
||||
chapters: List[Dict[str, Any]],
|
||||
metadata: Dict[str, Any],
|
||||
cover_path: Optional[Path] = None,
|
||||
config: Optional[AudiobookshelfConfig] = None,
|
||||
log_callback: Optional[callable] = None,
|
||||
) -> None:
|
||||
"""Upload to Audiobookshelf."""
|
||||
if config is None:
|
||||
# Load from job or global config
|
||||
cfg = getattr(job, "_abs_config", None)
|
||||
if cfg is None:
|
||||
from abogen.utils import load_config
|
||||
global_cfg = load_config() or {}
|
||||
abs_cfg = global_cfg.get("audiobookshelf")
|
||||
if isinstance(abs_cfg, Mapping):
|
||||
config = AudiobookshelfConfig(
|
||||
base_url=str(abs_cfg.get("base_url") or "").strip(),
|
||||
api_token=str(abs_cfg.get("api_token") or "").strip(),
|
||||
library_id=str(abs_cfg.get("library_id") or "").strip(),
|
||||
collection_id=(str(abs_cfg.get("collection_id") or "").strip() or None),
|
||||
folder_id=str(abs_cfg.get("folder_id") or "").strip(),
|
||||
verify_ssl=self._coerce_bool(abs_cfg.get("verify_ssl"), True),
|
||||
send_cover=self._coerce_bool(abs_cfg.get("send_cover"), True),
|
||||
send_chapters=self._coerce_bool(abs_cfg.get("send_chapters"), True),
|
||||
send_subtitles=self._coerce_bool(abs_cfg.get("send_subtitles"), False),
|
||||
timeout=float(abs_cfg.get("timeout", 3600.0)),
|
||||
)
|
||||
else:
|
||||
if log_callback:
|
||||
log_callback("Audiobookshelf upload skipped: not configured", "warning")
|
||||
return
|
||||
|
||||
if not config.base_url or not config.api_token or not config.library_id:
|
||||
if log_callback:
|
||||
log_callback("Audiobookshelf upload skipped: configure base URL, API token, and library ID first", "warning")
|
||||
return
|
||||
if not config.folder_id:
|
||||
if log_callback:
|
||||
log_callback("Audiobookshelf upload skipped: enter folder name or ID in settings", "warning")
|
||||
return
|
||||
|
||||
if not audio_path.exists():
|
||||
if log_callback:
|
||||
log_callback("Audiobookshelf upload skipped: audio output not found", "warning")
|
||||
return
|
||||
|
||||
existing_subtitles = [p for p in subtitle_paths if p.exists()] if config.send_subtitles else None
|
||||
chapters_to_send = chapters if config.send_chapters else None
|
||||
|
||||
client = AudiobookshelfClient(config)
|
||||
|
||||
display_title = metadata.get("title") or audio_path.stem
|
||||
try:
|
||||
existing_items = client.find_existing_items(display_title, folder_id=config.folder_id)
|
||||
except AudiobookshelfUploadError as exc:
|
||||
if log_callback:
|
||||
log_callback(f"Audiobookshelf lookup failed: {exc}", "error")
|
||||
return
|
||||
|
||||
if existing_items:
|
||||
if log_callback:
|
||||
log_callback(f"Removing existing Audiobookshelf item(s) for '{display_title}' before upload.", "info")
|
||||
try:
|
||||
client.delete_items(existing_items)
|
||||
except Exception as exc:
|
||||
if log_callback:
|
||||
log_callback(f"Failed to remove existing item(s): {exc}", "warning")
|
||||
|
||||
cover_to_send = cover_path
|
||||
if config.send_cover and cover_to_send:
|
||||
if isinstance(cover_to_send, str):
|
||||
cover_to_send = Path(cover_to_send)
|
||||
if not cover_to_send.exists():
|
||||
cover_to_send = None
|
||||
|
||||
client.upload_audiobook(
|
||||
audio_path,
|
||||
metadata=metadata,
|
||||
cover_path=cover_to_send,
|
||||
chapters=chapters_to_send,
|
||||
subtitles=existing_subtitles,
|
||||
)
|
||||
|
||||
if log_callback:
|
||||
log_callback("Audiobookshelf upload queued.", "info")
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
@staticmethod
|
||||
def _normalize_metadata_casefold(values: Optional[Mapping[str, Any]]) -> Dict[str, Any]:
|
||||
normalized = {}
|
||||
if not values:
|
||||
return normalized
|
||||
for key, value in values.items():
|
||||
if value is None:
|
||||
continue
|
||||
key_text = str(key).strip().lower()
|
||||
if not key_text:
|
||||
continue
|
||||
if isinstance(value, (list, tuple, set)):
|
||||
normalized[key_text] = value
|
||||
else:
|
||||
text = str(value).strip()
|
||||
if text:
|
||||
normalized[key_text] = text
|
||||
return normalized
|
||||
|
||||
@staticmethod
|
||||
def _split_people_field(raw: Any) -> List[str]:
|
||||
if raw is None:
|
||||
return []
|
||||
if isinstance(raw, (list, tuple, set)):
|
||||
results = []
|
||||
for item in raw:
|
||||
results.extend(ExportService._split_people_field(item))
|
||||
return results
|
||||
text = str(raw or "").strip()
|
||||
if not text:
|
||||
return []
|
||||
import re
|
||||
tokens = [token.strip() for token in re.split(r"[;,/&]|\band\b", text, flags=re.IGNORECASE) if token.strip()]
|
||||
seen = set()
|
||||
ordered = []
|
||||
for token in tokens:
|
||||
key = token.casefold()
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
ordered.append(token)
|
||||
return ordered
|
||||
|
||||
@staticmethod
|
||||
def _split_simple_list(raw: Any) -> List[str]:
|
||||
if raw is None:
|
||||
return []
|
||||
if isinstance(raw, (list, tuple, set)):
|
||||
results = []
|
||||
for item in raw:
|
||||
results.extend(ExportService._split_simple_list(item))
|
||||
return results
|
||||
text = str(raw or "").strip()
|
||||
if not text:
|
||||
return []
|
||||
import re
|
||||
tokens = [token.strip() for token in re.split(r"[;,\n]", text) if token.strip()]
|
||||
seen = set()
|
||||
ordered = []
|
||||
for token in tokens:
|
||||
key = token.casefold()
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
ordered.append(token)
|
||||
return ordered
|
||||
|
||||
@staticmethod
|
||||
def _first_nonempty(*values: Any) -> Optional[str]:
|
||||
for value in values:
|
||||
if value is None:
|
||||
continue
|
||||
if isinstance(value, (list, tuple, set)):
|
||||
items = list(value)
|
||||
if not items:
|
||||
continue
|
||||
value = items[0]
|
||||
text = str(value).strip()
|
||||
if text:
|
||||
return text
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _extract_year(raw: Optional[str]) -> Optional[int]:
|
||||
if not raw:
|
||||
return None
|
||||
text = str(raw).strip()
|
||||
if not text:
|
||||
return None
|
||||
import re
|
||||
match = re.search(r"(19|20)\d{2}", text)
|
||||
if match:
|
||||
try:
|
||||
return int(match.group(0))
|
||||
except ValueError:
|
||||
return None
|
||||
try:
|
||||
parsed = int(text)
|
||||
except ValueError:
|
||||
return None
|
||||
if 0 < parsed < 3000:
|
||||
return parsed
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _normalize_series_sequence(raw: Any) -> Optional[str]:
|
||||
if raw is None:
|
||||
return None
|
||||
if isinstance(raw, (int, float)):
|
||||
if isinstance(raw, float) and (raw != raw or raw == float("inf") or raw == float("-inf")):
|
||||
return None
|
||||
text = str(raw)
|
||||
else:
|
||||
text = str(raw).strip()
|
||||
if not text:
|
||||
return None
|
||||
candidate = text.replace(",", ".")
|
||||
import re
|
||||
match = re.search(r"\d+(?:\.\d+)?", candidate)
|
||||
if not match:
|
||||
return None
|
||||
normalized = match.group(0)
|
||||
if "." in normalized:
|
||||
normalized = normalized.rstrip("0").rstrip(".")
|
||||
return normalized or "0"
|
||||
try:
|
||||
return str(int(normalized))
|
||||
except ValueError:
|
||||
cleaned = normalized.lstrip("0")
|
||||
return cleaned or "0"
|
||||
|
||||
@staticmethod
|
||||
def _coerce_bool(value: Any, default: bool = True) -> bool:
|
||||
if isinstance(value, bool):
|
||||
return value
|
||||
if isinstance(value, str):
|
||||
lowered = value.strip().lower()
|
||||
if lowered in {"true", "1", "yes", "on"}:
|
||||
return True
|
||||
if lowered in {"false", "0", "no", "off"}:
|
||||
return False
|
||||
return default
|
||||
if value is None:
|
||||
return default
|
||||
return bool(value)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"ExportConfig",
|
||||
"ExportService",
|
||||
]
|
||||
@@ -1,12 +1,10 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import traceback
|
||||
import gc
|
||||
from datetime import datetime
|
||||
@@ -21,6 +19,7 @@ import soundfile as sf
|
||||
import static_ffmpeg
|
||||
|
||||
from abogen.tts_plugin.utils import get_voices, is_plugin_registered, resolve_voice_to_plugin
|
||||
from abogen.infrastructure.exporters import ExportService
|
||||
from abogen.epub3.exporter import build_epub3_package
|
||||
from abogen.kokoro_text_normalization import ApostropheConfig, normalize_for_pipeline, HAS_NUM2WORDS
|
||||
from abogen.normalization_settings import (
|
||||
@@ -52,6 +51,8 @@ from abogen.infrastructure.subtitle_writer import create_subtitle_writer, Subtit
|
||||
from .service import Job, JobStatus
|
||||
|
||||
|
||||
_export_svc = ExportService()
|
||||
|
||||
SPLIT_PATTERN = r"\n+"
|
||||
SAMPLE_RATE = 24000
|
||||
|
||||
@@ -1277,85 +1278,6 @@ def _chunk_text_for_tts(entry: Mapping[str, Any]) -> str:
|
||||
).strip()
|
||||
|
||||
|
||||
def _escape_ffmetadata_value(value: str) -> str:
|
||||
escaped = str(value).replace("\\", "\\\\").replace("\n", "\\n")
|
||||
escaped = escaped.replace("=", "\\=").replace(";", "\\;").replace("#", "\\#")
|
||||
return escaped
|
||||
|
||||
|
||||
def _metadata_to_ffmpeg_args(metadata: Dict[str, Any]) -> List[str]:
|
||||
args: List[str] = []
|
||||
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 _render_ffmetadata(metadata: Dict[str, Any], chapters: List[Dict[str, Any]]) -> str:
|
||||
lines: List[str] = [";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}={_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={_escape_ffmetadata_value(title)}")
|
||||
voice = chapter.get("voice")
|
||||
if voice:
|
||||
lines.append(f"voice={_escape_ffmetadata_value(voice)}")
|
||||
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
|
||||
def _write_ffmetadata_file(
|
||||
audio_path: Path,
|
||||
metadata: Dict[str, Any],
|
||||
chapters: List[Dict[str, Any]],
|
||||
) -> Optional[Path]:
|
||||
content = _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)
|
||||
|
||||
|
||||
def _apply_m4b_chapters_with_mutagen(
|
||||
audio_path: Path,
|
||||
chapters: List[Dict[str, Any]],
|
||||
@@ -1427,14 +1349,14 @@ def _embed_m4b_metadata(
|
||||
) -> None:
|
||||
metadata_map = dict(metadata_payload.get("metadata") or {})
|
||||
chapter_entries = list(metadata_payload.get("chapters") or [])
|
||||
ffmetadata_path = _write_ffmetadata_file(audio_path, metadata_map, chapter_entries)
|
||||
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 = _metadata_to_ffmpeg_args(metadata_map)
|
||||
metadata_args = _export_svc._metadata_to_ffmpeg_args(metadata_map)
|
||||
|
||||
if not ffmetadata_path and not cover_path and not metadata_args:
|
||||
return
|
||||
@@ -2595,7 +2517,7 @@ def _build_ffmpeg_command(path: Path, fmt: str, metadata: Optional[Dict[str, str
|
||||
base += ["-c:a", "copy"]
|
||||
|
||||
if metadata:
|
||||
base.extend(_metadata_to_ffmpeg_args(metadata))
|
||||
base.extend(_export_svc._metadata_to_ffmpeg_args(metadata))
|
||||
base.append(str(path))
|
||||
return base
|
||||
|
||||
|
||||
@@ -0,0 +1,163 @@
|
||||
"""Tests for ExportService FFmpeg metadata methods."""
|
||||
import os
|
||||
import sys
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
|
||||
|
||||
from pathlib import Path
|
||||
from abogen.infrastructure.exporters import ExportService
|
||||
|
||||
|
||||
class TestEscapeFfmetadataValue:
|
||||
def setup_method(self):
|
||||
self.svc = ExportService()
|
||||
|
||||
def test_simple_string(self):
|
||||
assert self.svc._escape_ffmetadata_value("hello") == "hello"
|
||||
|
||||
def test_escapes_backslash(self):
|
||||
assert self.svc._escape_ffmetadata_value("a\\b") == "a\\\\b"
|
||||
|
||||
def test_escapes_newline(self):
|
||||
assert self.svc._escape_ffmetadata_value("line1\nline2") == "line1\\nline2"
|
||||
|
||||
def test_escapes_equals(self):
|
||||
assert self.svc._escape_ffmetadata_value("key=value") == "key\\=value"
|
||||
|
||||
def test_escapes_semicolon(self):
|
||||
assert self.svc._escape_ffmetadata_value("a;b") == "a\\;b"
|
||||
|
||||
def test_escapes_hash(self):
|
||||
assert self.svc._escape_ffmetadata_value("#comment") == "\\#comment"
|
||||
|
||||
def test_escapes_all_special(self):
|
||||
result = self.svc._escape_ffmetadata_value("a\\b\nc=d;e#f")
|
||||
assert "\\\\" in result
|
||||
assert "\\n" in result
|
||||
assert "\\=" in result
|
||||
assert "\\;" in result
|
||||
assert "\\#" in result
|
||||
|
||||
def test_empty_string(self):
|
||||
assert self.svc._escape_ffmetadata_value("") == ""
|
||||
|
||||
|
||||
class TestRenderFfmetadata:
|
||||
def setup_method(self):
|
||||
self.svc = ExportService()
|
||||
|
||||
def test_renders_header(self):
|
||||
result = self.svc.render_ffmetadata({"title": "My Book"}, [])
|
||||
assert result.startswith(";FFMETADATA1\n")
|
||||
assert "title=My Book\n" in result
|
||||
|
||||
def test_renders_multiple_keys(self):
|
||||
result = self.svc.render_ffmetadata({"title": "T", "artist": "A"}, [])
|
||||
assert "title=T\n" in result
|
||||
assert "artist=A\n" in result
|
||||
|
||||
def test_skips_none_values(self):
|
||||
result = self.svc.render_ffmetadata({"title": None}, [])
|
||||
assert "title=" not in result
|
||||
|
||||
def test_renders_chapters(self):
|
||||
chapters = [{"start": 0.0, "end": 10.0, "title": "Ch 1"}]
|
||||
result = self.svc.render_ffmetadata({}, chapters)
|
||||
assert "[CHAPTER]" in result
|
||||
assert "TIMEBASE=1/1000" in result
|
||||
assert "START=0" in result
|
||||
assert "END=10000" in result
|
||||
assert "title=Ch 1" in result
|
||||
|
||||
def test_renders_voice_in_chapter(self):
|
||||
chapters = [{"start": 0.0, "end": 5.0, "voice": "af_heart"}]
|
||||
result = self.svc.render_ffmetadata({}, chapters)
|
||||
assert "voice=af_heart" in result
|
||||
|
||||
def test_skips_chapters_without_times(self):
|
||||
chapters = [{"title": "No times"}]
|
||||
result = self.svc.render_ffmetadata({}, chapters)
|
||||
assert "[CHAPTER]" not in result
|
||||
|
||||
def test_end_equals_start_gets_minimum_duration(self):
|
||||
chapters = [{"start": 5.0, "end": 5.0, "title": "Zero"}]
|
||||
result = self.svc.render_ffmetadata({}, chapters)
|
||||
assert "START=5000" in result
|
||||
assert "END=5001" in result
|
||||
|
||||
def test_empty_metadata_and_chapters(self):
|
||||
result = self.svc.render_ffmetadata({}, [])
|
||||
assert result.strip() == ";FFMETADATA1"
|
||||
|
||||
def test_escapes_special_chars_in_title(self):
|
||||
chapters = [{"start": 0.0, "end": 1.0, "title": "A=B;C#D"}]
|
||||
result = self.svc.render_ffmetadata({}, chapters)
|
||||
assert "\\=" in result
|
||||
assert "\\;" in result
|
||||
assert "\\#" in result
|
||||
|
||||
def test_negative_start_clamped_to_zero(self):
|
||||
chapters = [{"start": -1.0, "end": 5.0, "title": "Neg"}]
|
||||
result = self.svc.render_ffmetadata({}, chapters)
|
||||
assert "START=0" in result
|
||||
|
||||
|
||||
class TestMetadataToFfmpegArgs:
|
||||
def setup_method(self):
|
||||
self.svc = ExportService()
|
||||
|
||||
def test_simple_metadata(self):
|
||||
args = self.svc._metadata_to_ffmpeg_args({"title": "My Book"})
|
||||
assert args == ["-metadata", "title=My Book"]
|
||||
|
||||
def test_year_becomes_date(self):
|
||||
args = self.svc._metadata_to_ffmpeg_args({"year": "2024"})
|
||||
assert args == ["-metadata", "date=2024"]
|
||||
|
||||
def test_skips_none_and_empty(self):
|
||||
args = self.svc._metadata_to_ffmpeg_args({"title": None, "artist": ""})
|
||||
assert args == []
|
||||
|
||||
def test_skips_empty_key(self):
|
||||
args = self.svc._metadata_to_ffmpeg_args({"": "value"})
|
||||
assert args == []
|
||||
|
||||
def test_multiple_keys(self):
|
||||
args = self.svc._metadata_to_ffmpeg_args({"title": "T", "artist": "A"})
|
||||
assert "-metadata" in args
|
||||
assert "title=T" in args
|
||||
assert "artist=A" in args
|
||||
|
||||
def test_empty_metadata(self):
|
||||
assert self.svc._metadata_to_ffmpeg_args({}) == []
|
||||
|
||||
def test_none_metadata(self):
|
||||
assert self.svc._metadata_to_ffmpeg_args(None) == []
|
||||
|
||||
|
||||
class TestWriteFfmetadataFile:
|
||||
def setup_method(self):
|
||||
self.svc = ExportService()
|
||||
|
||||
def test_writes_file(self, tmp_path):
|
||||
audio = tmp_path / "test.mp3"
|
||||
audio.touch()
|
||||
meta = {"title": "My Book"}
|
||||
chapters = [{"start": 0.0, "end": 5.0, "title": "Ch 1"}]
|
||||
result = self.svc.write_ffmetadata_file(audio, meta, chapters)
|
||||
assert result is not None
|
||||
assert result.exists()
|
||||
content = result.read_text()
|
||||
assert ";FFMETADATA1" in content
|
||||
assert "title=My Book" in content
|
||||
|
||||
def test_returns_none_for_empty(self, tmp_path):
|
||||
audio = tmp_path / "test.mp3"
|
||||
audio.touch()
|
||||
result = self.svc.write_ffmetadata_file(audio, {}, [])
|
||||
assert result is None
|
||||
|
||||
def test_returns_none_for_only_header(self, tmp_path):
|
||||
audio = tmp_path / "test.mp3"
|
||||
audio.touch()
|
||||
result = self.svc.write_ffmetadata_file(audio, None, None)
|
||||
assert result is None
|
||||
@@ -1,8 +1,9 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from abogen.infrastructure.exporters import ExportService
|
||||
|
||||
from abogen.webui.conversion_runner import _render_ffmetadata, _write_ffmetadata_file
|
||||
|
||||
svc = ExportService()
|
||||
|
||||
|
||||
def test_render_ffmetadata_includes_chapters(tmp_path):
|
||||
@@ -17,7 +18,7 @@ def test_render_ffmetadata_includes_chapters(tmp_path):
|
||||
{"start": 5.0, "end": 12.345, "title": "Chapter 2"},
|
||||
]
|
||||
|
||||
rendered = _render_ffmetadata(metadata, chapters)
|
||||
rendered = svc.render_ffmetadata(metadata, chapters)
|
||||
|
||||
assert ";FFMETADATA1" in rendered
|
||||
assert "title=Sample Book" in rendered
|
||||
@@ -30,7 +31,7 @@ def test_render_ffmetadata_includes_chapters(tmp_path):
|
||||
assert "voice=voice_a" in rendered
|
||||
|
||||
audio_path = tmp_path / "book.m4b"
|
||||
metadata_path = _write_ffmetadata_file(audio_path, metadata, chapters)
|
||||
metadata_path = svc.write_ffmetadata_file(audio_path, metadata, chapters)
|
||||
assert metadata_path is not None
|
||||
assert metadata_path.exists()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user