feat: add metadata to audio files including generated cover art

1 files changed, 109 insertions, 32 deletions

diff --git a/converter/audio.py b/converter/audio.py
index 98864f6..ebcac21 100644
--- a/converter/audio.py
+++ b/converter/audio.py
@@ -6,7 +6,7 @@ import shutil
 import subprocess
 import traceback
 from pathlib import Path
-from typing import Dict, List, Optional, Tuple
+from typing import Dict, List, NamedTuple, Optional, Tuple
 
 from . import config
 
@@ -63,6 +63,54 @@ def _encode_args(output_format: str) -> List[str]:
     return ["-b:a", config.AUDIO_BITRATE]
 
 
+class TrackMeta(NamedTuple):
+    """Tags embedded into a finished audiobook file."""
+
+    title: str
+    artist: str = ""
+    album: str = ""
+    track: Optional[int] = None
+    total_tracks: Optional[int] = None
+
+
+def _tag_args(meta: TrackMeta, output_format: str) -> List[str]:
+    """Return -metadata args plus format-specific tagging flags."""
+    args = ["-metadata", f"title={meta.title}"]
+    if meta.artist:
+        args += ["-metadata", f"artist={meta.artist}"]
+    if meta.album:
+        args += ["-metadata", f"album={meta.album}"]
+    if meta.track and meta.total_tracks:
+        args += ["-metadata", f"track={meta.track}/{meta.total_tracks}"]
+    if output_format == "mp3":
+        # ID3v2.3 is what essentially every player reads; ffmpeg's default
+        # (v2.4) still confuses some of them.
+        args += ["-id3v2_version", "3"]
+    return args
+
+
+# Formats whose container has no reliable embedded-picture support.
+_NO_COVER_FORMATS = ("ogg", "wav")
+
+
+def _cover_args(output_format: str, cover_input_index: int) -> List[str]:
+    """Return per-output args attaching a cover image as an embedded picture.
+
+    The cover must already be added as an ffmpeg input; ``cover_input_index``
+    is that input's position on the command line. mp3/flac keep the PNG
+    stream as-is; m4b re-encodes to JPEG, which audiobook players expect.
+    """
+    if output_format in _NO_COVER_FORMATS:
+        return []
+    codec = "mjpeg" if output_format == "m4b" else "copy"
+    args = ["-map", f"{cover_input_index}:v", "-c:v", codec,
+            "-disposition:v", "attached_pic",
+            "-metadata:s:v", "title=Album cover"]
+    if output_format == "m4b":
+        args += ["-q:v", "3"]
+    return args
+
+
 _brand_supported: Optional[bool] = None
 
 
@@ -100,30 +148,42 @@ def _m4b_container_args() -> List[str]:
 
 
 def build_concat_command(concat_list: Path, output_path: Path, output_format: str,
-                         speed: float = 1.0, speed_path: Optional[Path] = None) -> List[str]:
+                         speed: float = 1.0, speed_path: Optional[Path] = None,
+                         meta: Optional[TrackMeta] = None,
+                         cover: Optional[Path] = None) -> List[str]:
     """Build the ffmpeg command that concatenates chunk audio into a book file."""
     encode = _encode_args(output_format)
     container = _m4b_container_args() if output_format == "m4b" else []
+    inputs = ["-f", "concat", "-safe", "0", "-i", str(concat_list)]
+    cover_index = None
+    if cover is not None and output_format not in _NO_COVER_FORMATS:
+        inputs += ["-i", str(cover)]
+        cover_index = 1
+    cover_block = (_cover_args(output_format, cover_index)
+                   if cover_index is not None else [])
+    tags = _tag_args(meta, output_format) if meta else []
     filters = atempo_filters(speed)
     if filters:
         if speed_path is None:
             raise ValueError("speed_path is required when speed is not 1.0")
         return [
-            "ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", str(concat_list),
-            "-filter_complex",
+            "ffmpeg", "-y", *inputs, "-filter_complex",
             f"[0:a]split=2[base][spd];[spd]{filters}[spdout]",
-            "-map", "[base]", *encode, *container, str(output_path),
-            "-map", "[spdout]", *encode, *container, str(speed_path),
+            "-map", "[base]", *encode, *tags, *cover_block, *container, str(output_path),
+            "-map", "[spdout]", *encode, *tags, *cover_block, *container, str(speed_path),
         ]
+    output_maps = ["-map", "0:a"] if cover_block else []
     return [
-        "ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", str(concat_list),
-        *encode, *container, str(output_path),
+        "ffmpeg", "-y", *inputs, *output_maps,
+        *encode, *tags, *cover_block, *container, str(output_path),
     ]
 
 
 def build_m4b_chapters_command(concat_list: Path, metadata_file: Path, output_path: Path,
-                               speed: float = 1.0, speed_path: Optional[Path] = None,
-                               speed_metadata_file: Optional[Path] = None) -> List[str]:
+                                speed: float = 1.0, speed_path: Optional[Path] = None,
+                                speed_metadata_file: Optional[Path] = None,
+                                meta: Optional[TrackMeta] = None,
+                                cover: Optional[Path] = None) -> List[str]:
     """Build the ffmpeg command that assembles chapter audio into one m4b.
 
     Chapter metadata inputs are bound to their outputs with explicit
@@ -132,22 +192,30 @@ def build_m4b_chapters_command(concat_list: Path, metadata_file: Path, output_pa
     """
     encode = _encode_args("m4b")
     container = _m4b_container_args()
+    tags = _tag_args(meta, "m4b") if meta else []
+    inputs = ["-f", "concat", "-safe", "0", "-i", str(concat_list),
+              "-i", str(metadata_file)]
+    input_count = 2  # concat audio + ffmetadata
+    if speed_path is not None:
+        inputs += ["-i", str(speed_metadata_file)]
+        input_count += 1
+    cover_block: List[str] = []
+    if cover is not None:
+        inputs += ["-i", str(cover)]
+        cover_block = _cover_args("m4b", input_count)
     if speed_path is not None:
         return [
-            "ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", str(concat_list),
-            "-i", str(metadata_file), "-i", str(speed_metadata_file),
-            "-filter_complex",
+            "ffmpeg", "-y", *inputs, "-filter_complex",
             f"[0:a]split=2[base][spd];[spd]{atempo_filters(speed)}[spdout]",
-            "-map", "[base]", "-map_metadata", "1", "-map_chapters", "1",
-            *encode, *container, str(output_path),
-            "-map", "[spdout]", "-map_metadata", "2", "-map_chapters", "2",
-            *encode, *container, str(speed_path),
+            "-map", "[base]", *cover_block, "-map_metadata", "1", "-map_chapters", "1",
+            *encode, *tags, *container, str(output_path),
+            "-map", "[spdout]", *cover_block, "-map_metadata", "2", "-map_chapters", "2",
+            *encode, *tags, *container, str(speed_path),
         ]
     return [
-        "ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", str(concat_list),
-        "-i", str(metadata_file),
-        "-map", "0:a", "-map_metadata", "1", "-map_chapters", "1",
-        *encode, *container, str(output_path),
+        "ffmpeg", "-y", *inputs,
+        "-map", "0:a", *cover_block, "-map_metadata", "1", "-map_chapters", "1",
+        *encode, *tags, *container, str(output_path),
     ]
 
 
@@ -215,16 +283,20 @@ def _collect_chunk_files(total_chunks: int,
 def combine_chunks(total_chunks: int, output_path: Path,
                    chunk_results: Optional[Dict[int, Optional[Path]]] = None,
                    speed: float = 1.0, output_format: str = "mp3",
-                   intermediate: bool = False) -> bool:
+                   intermediate: bool = False,
+                   meta: Optional[TrackMeta] = None,
+                   cover: Optional[Path] = None) -> bool:
     """Combine audio chunks into the final audiobook using ffmpeg's concat demuxer.
 
     ``chunk_results`` maps chunk numbers to the audio file each chunk produced
     (None for failed chunks); failed and missing chunks are skipped. When
     ``speed`` differs from 1.0, an additional speed-adjusted copy is written
-    next to the normal-speed file. Chunks are streamed by ffmpeg, so the whole
-    book is never held in memory. Set ``intermediate`` for scratch chapter
-    audio on the way to a larger output (e.g. a chaptered m4b) so save
-    messages don't present it as the final audiobook.
+    next to the normal-speed file. ``meta``/``cover`` embed tags and cover
+    art into the output (skipped for intermediate chapter scratch audio).
+    Chunks are streamed by ffmpeg, so the whole book is never held in
+    memory. Set ``intermediate`` for scratch chapter audio on the way to a
+    larger output (e.g. a chaptered m4b) so save messages don't present it
+    as the final audiobook.
     """
     if shutil.which("ffmpeg") is None or shutil.which("ffprobe") is None:
         logger.error("ffmpeg and ffprobe are required to combine audio chunks (install ffmpeg)")
@@ -249,7 +321,9 @@ def combine_chunks(total_chunks: int, output_path: Path,
         if atempo_filters(speed):
             speed_path = output_path.with_name(f"{output_path.stem}_{speed:g}{output_path.suffix}")
         cmd = build_concat_command(concat_list, output_path, output_format,
-                                   speed=speed, speed_path=speed_path)
+                                   speed=speed, speed_path=speed_path,
+                                   meta=None if intermediate else meta,
+                                   cover=None if intermediate else cover)
 
         proc = subprocess.run(cmd, capture_output=True, text=True)
         if proc.returncode != 0:
@@ -373,13 +447,15 @@ def build_ffmetadata(chapters: List[tuple], path: Path) -> None:
 
 
 def combine_chapters_to_m4b(chapter_files: List[Path], titles: List[str],
-                            output_path: Path, speed: float = 1.0) -> bool:
+                            output_path: Path, speed: float = 1.0,
+                            meta: Optional[TrackMeta] = None,
+                            cover: Optional[Path] = None) -> bool:
     """Concatenate per-chapter audio into a single m4b with embedded chapter markers.
 
     Chapter start/end times are derived from each chapter file's duration and
-    written as ffmpeg chapter metadata. When ``speed`` differs from 1.0, a
-    speed-adjusted copy (with rescaled chapter markers) is written alongside
-    the normal-speed file.
+    written as ffmpeg chapter metadata. ``meta``/``cover`` embed tags and
+    cover art. When ``speed`` differs from 1.0, a speed-adjusted copy (with
+    rescaled chapter markers) is written alongside the normal-speed file.
     """
     if shutil.which("ffmpeg") is None or shutil.which("ffprobe") is None:
         logger.error("ffmpeg and ffprobe are required to build an m4b with chapters")
@@ -413,7 +489,8 @@ def combine_chapters_to_m4b(chapter_files: List[Path], titles: List[str],
         cmd = build_m4b_chapters_command(concat_list, metadata_file, output_path,
                                          speed=speed, speed_path=speed_path,
                                          speed_metadata_file=speed_metadata_file
-                                         if speed_path is not None else None)
+                                         if speed_path is not None else None,
+                                         meta=meta, cover=cover)
 
         proc = subprocess.run(cmd, capture_output=True, text=True)
         if proc.returncode != 0: