diff --git a/abogen/web/conversion_runner.py b/abogen/web/conversion_runner.py index b794b52..9003a52 100644 --- a/abogen/web/conversion_runner.py +++ b/abogen/web/conversion_runner.py @@ -6,6 +6,7 @@ import os import re import subprocess import sys +import tempfile from contextlib import ExitStack from dataclasses import dataclass from pathlib import Path @@ -197,6 +198,142 @@ def _chapter_voice_spec(job: Job, override: Optional[Dict[str, Any]]) -> str: return job.voice or "" +def _escape_ffmetadata_value(value: str) -> str: + escaped = str(value).replace("\\", "\\\\").replace("\n", "\\n") + escaped = escaped.replace("=", "\\=").replace(";", "\\;").replace("#", "\\#") + return escaped + + +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 _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 = _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 + + if not ffmetadata_path and not cover_path: + return + + job.add_log("Embedding chapters and cover 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 += ["-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"] + + command += ["-movflags", "+faststart+use_metadata_tags"] + + temp_output = audio_path.with_suffix(audio_path.suffix + ".tmp") + 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) + + def run_conversion_job(job: Job) -> None: job.add_log("Preparing conversion pipeline") canceller = _make_canceller(job) @@ -204,6 +341,9 @@ def run_conversion_job(job: Job) -> None: sink_stack = ExitStack() subtitle_writer: Optional[SubtitleWriter] = None chapter_paths: list[Path] = [] + chapter_markers: List[Dict[str, Any]] = [] + metadata_payload: Dict[str, Any] = {"metadata": {}, "chapters": []} + audio_output_path: Optional[Path] = None try: pipeline = _load_pipeline(job) extraction = extract_from_path(job.stored_path) @@ -266,7 +406,6 @@ def run_conversion_job(job: Job) -> None: subtitle_index = 1 current_time = 0.0 total_chapters = len(extraction.chapters) - chapter_markers: List[Dict[str, Any]] = [] job.add_log(f"Detected {total_chapters} chapter{'s' if total_chapters != 1 else ''}") for idx, chapter in enumerate(extraction.chapters, start=1): @@ -392,13 +531,14 @@ def run_conversion_job(job: Job) -> None: if not audio_path and chapter_paths: job.result.audio_path = chapter_paths[0] + metadata_payload = { + "metadata": dict(job.metadata_tags or {}), + "chapters": chapter_markers, + } + if metadata_dir: metadata_dir.mkdir(parents=True, exist_ok=True) metadata_file = metadata_dir / "metadata.json" - metadata_payload = { - "metadata": job.metadata_tags, - "chapters": chapter_markers, - } metadata_file.write_text(json.dumps(metadata_payload, indent=2), encoding="utf-8") job.result.artifacts["metadata"] = metadata_file @@ -408,6 +548,8 @@ def run_conversion_job(job: Job) -> None: if job.status != JobStatus.CANCELLED: job.progress = 1.0 + audio_output_path = job.result.audio_path + except _JobCancelled: job.status = JobStatus.CANCELLED job.add_log("Job cancelled", level="warning") @@ -420,6 +562,17 @@ def run_conversion_job(job: Job) -> None: if subtitle_writer: subtitle_writer.close() + if ( + audio_output_path + and job.output_format.lower() == "m4b" + and not job.cancel_requested + and job.status not in {JobStatus.FAILED, JobStatus.CANCELLED} + ): + try: + _embed_m4b_metadata(audio_output_path, metadata_payload, job) + except Exception as exc: # pragma: no cover - best effort + job.add_log(f"Unable to embed metadata into m4b output: {exc}", level="warning") + def _load_pipeline(job: Job): cfg = load_config() diff --git a/tests/test_ffmetadata.py b/tests/test_ffmetadata.py new file mode 100644 index 0000000..82d05cc --- /dev/null +++ b/tests/test_ffmetadata.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +from pathlib import Path + +from abogen.web.conversion_runner import _render_ffmetadata, _write_ffmetadata_file + + +def test_render_ffmetadata_includes_chapters(tmp_path): + metadata = { + "title": "Sample Book", + "artist": "Author Name", + "comment": "Line one\nLine two", + "publisher": "ACME=Corp", + } + chapters = [ + {"start": 0.0, "end": 5.0, "title": "Intro", "voice": "voice_a"}, + {"start": 5.0, "end": 12.345, "title": "Chapter 2"}, + ] + + rendered = _render_ffmetadata(metadata, chapters) + + assert ";FFMETADATA1" in rendered + assert "title=Sample Book" in rendered + assert "artist=Author Name" in rendered + assert "comment=Line one\\nLine two" in rendered + assert "publisher=ACME\\=Corp" in rendered + assert rendered.count("[CHAPTER]") == 2 + assert "START=0" in rendered + assert "END=5000" in rendered + assert "voice=voice_a" in rendered + + audio_path = tmp_path / "book.m4b" + metadata_path = _write_ffmetadata_file(audio_path, metadata, chapters) + assert metadata_path is not None + assert metadata_path.exists() + + content = metadata_path.read_text(encoding="utf-8") + assert "END=12345" in content + + metadata_path.unlink()