diff --git a/abogen/infrastructure/exporters.py b/abogen/infrastructure/exporters.py new file mode 100644 index 0000000..cc2f599 --- /dev/null +++ b/abogen/infrastructure/exporters.py @@ -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", +] diff --git a/abogen/webui/conversion_runner.py b/abogen/webui/conversion_runner.py index f0b52a2..4508f32 100644 --- a/abogen/webui/conversion_runner.py +++ b/abogen/webui/conversion_runner.py @@ -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 diff --git a/tests/test_exporters.py b/tests/test_exporters.py new file mode 100644 index 0000000..398cc73 --- /dev/null +++ b/tests/test_exporters.py @@ -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 diff --git a/tests/test_ffmetadata.py b/tests/test_ffmetadata.py index cc58cc8..c96e9af 100644 --- a/tests/test_ffmetadata.py +++ b/tests/test_ffmetadata.py @@ -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()