Support additional attachments in melt

README.md

@@ -72,6 +72,14 @@ python -m twotone concatenate --help
 
 The melt tool scans for duplicate video files and creates a single output using the best quality segments from each copy. Duplicates can be provided manually or taken from a Jellyfin server. By default all input files used to create the output are removed after a successful run. Use `--keep-input-files` to preserve them.
 
+In addition to thumbnails, melt now preserves all kinds of attachments (such as
+fonts or images) found in the source files. If multiple files provide
+incomparable attachments of the same content type, melt chooses one from the
+best-quality video file and logs a warning. Chapter data is treated in the same
+way and copied from the selected source. Attached fonts are especially important
+for styled ASS/SSA subtitles as players rely on them to display the subtitles
+with the intended typography.
+
 ```bash
 python -m twotone melt --help
 ```

tests/chapters/simple_chapters.txt

@@ -0,0 +1,2 @@
+CHAPTER01=00:00:00.000
+CHAPTER01NAME=Intro

tests/common.py

@@ -180,6 +180,14 @@ def get_subtitle(name: str) -> str:
     return os.path.join(current_path, "subtitles", name)
 
 
+def get_font(name: str) -> str:
+    return os.path.join(current_path, "fonts", name)
+
+
+def get_chapter(name: str) -> str:
+    return os.path.join(current_path, "chapters", name)
+
+
 def build_test_video(
     output_path: str,
     wd: str,
@@ -188,23 +196,29 @@ def build_test_video(
     audio_name: Union[str, None] = None,
     subtitle: Union[str, bool, None] = None,
     thumbnail_name: Union[str, None] = None,
+    chapter_name: Union[str, None] = None,
+    attachments: List[str] | None = None,
 ) -> str:
     with tempfile.TemporaryDirectory(dir = wd) as tmp_dir:
         video_path = get_video(video_name)
         audio_path = None if audio_name is None else get_audio(audio_name)
         thumbnail_path = None if thumbnail_name is None else get_image(thumbnail_name)
+        chapters_path = None if chapter_name is None else get_chapter(chapter_name)
 
         subtitle_path = get_subtitle(subtitle) if isinstance(subtitle, str) else None
         if subtitle_path is None and isinstance(subtitle, bool) and subtitle:
             video_length = video_utils.get_video_duration(video_path)
             subtitle_path = os.path.join(tmp_dir, "temporary_subtitle_file.srt")
             generate_subrip_subtitles(subtitle_path, length = video_length)
 
-        video_utils.generate_mkv(output_path,
-                                video_path,
-                                [subtitles_utils.build_subtitle_from_path(subtitle_path)] if subtitle_path else None,
-                                [subtitles_utils.build_audio_from_path(audio_path)] if audio_path else None,
-                                thumbnail_path,
+        video_utils.generate_mkv(
+            output_path,
+            video_path,
+            [subtitles_utils.build_subtitle_from_path(subtitle_path)] if subtitle_path else None,
+            [subtitles_utils.build_audio_from_path(audio_path)] if audio_path else None,
+            thumbnail_path,
+            chapters_path,
+            attachments,
         )
 
         return output_path

tests/fonts/dummy.ttf

@@ -0,0 +1 @@
+DUMMYFONT

tests/test_melt.py

@@ -13,7 +13,20 @@
 from twotone.tools.melt import Melter
 from twotone.tools.melt.melt import StaticSource, StreamsPicker
 from twotone.tools.utils.files_utils import ScopedDirectory
-from common import TwoToneTestCase, FileCache, add_test_media, add_to_test_dir, build_test_video, current_path, get_audio, get_video, hashes, list_files
+from common import (
+    TwoToneTestCase,
+    FileCache,
+    add_test_media,
+    add_to_test_dir,
+    build_test_video,
+    current_path,
+    get_audio,
+    get_video,
+    get_font,
+    get_chapter,
+    hashes,
+    list_files,
+)
 
 
 def normalize(obj):
@@ -417,6 +430,48 @@ def test_additional_attachements(self):
         self.assertEqual(len(output_file_data["tracks"]["video"]), 1)
         self.assertEqual(len(output_file_data["attachments"]), 1)
 
+    def test_chapters(self):
+        video1 = build_test_video(
+            os.path.join(self.wd.path, "c1.mkv"),
+            self.wd.path,
+            "fog-over-mountainside-13008647.mp4",
+            subtitle=True,
+            chapter_name="simple_chapters.txt",
+        )
+        video2 = build_test_video(
+            os.path.join(self.wd.path, "c2.mkv"),
+            self.wd.path,
+            "fog-over-mountainside-13008647.mp4",
+            subtitle=True,
+        )
+
+        interruption = generic_utils.InterruptibleProcess()
+        duplicates = StaticSource(interruption)
+        duplicates.add_entry("Fog", video1)
+        duplicates.add_entry("Fog", video2)
+        duplicates.add_metadata(video1, "audio_lang", "eng")
+        duplicates.add_metadata(video2, "audio_lang", "eng")
+
+        output_dir = os.path.join(self.wd.path, "output")
+        os.makedirs(output_dir)
+
+        melter = Melter(
+            logging.getLogger("Melter"),
+            interruption,
+            duplicates,
+            live_run=True,
+            wd=self.wd.path,
+            output=output_dir,
+        )
+        melter.melt()
+
+        output_file_hash = hashes(output_dir)
+        self.assertEqual(len(output_file_hash), 1)
+        output_file = list(output_file_hash)[0]
+
+        output_file_data = video_utils.get_video_data_mkvmerge(output_file)
+        self.assertTrue(output_file_data["chapters"])
+
 
     def test_attachement_in_file_with_useless_streams(self):
         # video #1 comes with all interesting data. the only thing video #2 can offer is an attachment.
@@ -446,6 +501,50 @@ def test_attachement_in_file_with_useless_streams(self):
         self.assertEqual(len(output_file_data["tracks"]["video"]), 1)
         self.assertEqual(len(output_file_data["attachments"]), 1)
 
+    def test_font_attachment(self):
+        font = get_font("dummy.ttf")
+        video1 = build_test_video(
+            os.path.join(self.wd.path, "o1.mkv"),
+            self.wd.path,
+            "fog-over-mountainside-13008647.mp4",
+            subtitle=True,
+            attachments=[font],
+        )
+        video2 = build_test_video(
+            os.path.join(self.wd.path, "o2.mkv"),
+            self.wd.path,
+            "fog-over-mountainside-13008647.mp4",
+            subtitle=True,
+        )
+
+        interruption = generic_utils.InterruptibleProcess()
+        duplicates = StaticSource(interruption)
+        duplicates.add_entry("Fog", video1)
+        duplicates.add_entry("Fog", video2)
+        duplicates.add_metadata(video1, "audio_lang", "eng")
+        duplicates.add_metadata(video2, "audio_lang", "eng")
+
+        output_dir = os.path.join(self.wd.path, "output")
+        os.makedirs(output_dir)
+
+        melter = Melter(
+            logging.getLogger("Melter"),
+            interruption,
+            duplicates,
+            live_run=True,
+            wd=self.wd.path,
+            output=output_dir,
+        )
+        melter.melt()
+
+        output_file_hash = hashes(output_dir)
+        self.assertEqual(len(output_file_hash), 1)
+        output_file = list(output_file_hash)[0]
+
+        output_file_data = video_utils.get_video_data_mkvmerge(output_file)
+        self.assertEqual(len(output_file_data["tracks"]["video"]), 1)
+        self.assertEqual(len(output_file_data["attachments"]), 1)
+
 
     sample_streams = [
         # case: merge all audio tracks

twotone/tools/melt/attachments_picker.py

@@ -8,13 +8,54 @@ class AttachmentsPicker:
     def __init__(self, logger: logging.Logger):
         self.logger = logger
 
-    def pick_attachments(self, files_details: Dict):
-        picked_attachments = []
+    def pick_attachments(self, files_details: Dict, best_video_file: str):
+        """Pick attachments for output.
+
+        If multiple attachments of the same content type are available they are
+        considered incomparable. In that case one attachment from
+        ``best_video_file`` is chosen and a warning is logged.
+        """
+
+        attachments_by_type = {}
         for file, attachments in files_details.items():
             for attachment in attachments:
-                picked_attachments.append((file, attachment["tid"]))
+                ctype = attachment.get("content_type")
+                attachments_by_type.setdefault(ctype, []).append((file, attachment))
+
+        picked = []
+        for ctype, items in attachments_by_type.items():
+            if len(items) == 1:
+                file, att = items[0]
+                picked.append((file, att["tid"]))
+                continue
+
+            # prefer attachment from best_video_file if available
+            chosen = None
+            for file, att in items:
+                if file == best_video_file:
+                    chosen = (file, att)
+                    break
+            if not chosen:
+                chosen = items[0]
+
+            self.logger.warning(
+                f"Multiple attachments of type {ctype} found; using {chosen[1]['file_name']} from {chosen[0]}"
+            )
+            picked.append((chosen[0], chosen[1]["tid"]))
+
+        return picked
 
-        if picked_attachments:
-            return [picked_attachments[0]]
+    def pick_chapter(self, chapters: Dict[str, bool], best_video_file: str):
+        """Pick chapter source file."""
+        chapter_files = [f for f, has in chapters.items() if has]
+        if not chapter_files:
+            return None
+        chosen = None
+        if len(chapter_files) > 1:
+            chosen = best_video_file if best_video_file in chapter_files else chapter_files[0]
+            self.logger.warning(
+                f"Multiple chapter sources found; using chapters from {chosen}"
+            )
         else:
-            return []
+            chosen = chapter_files[0]
+        return chosen

twotone/tools/melt/melt.py

@@ -197,6 +197,7 @@ def show(key: str) -> bool:
         self.logger.info(f"File {files_utils.get_printable_path(file, common_prefix)} details:")
         tracks = details["tracks"]
         attachments = details["attachments"]
+        chapters = details.get("chapters")
 
         for stream_type, streams in tracks.items():
             info = f"{stream_type}: {len(streams)} track(s), languages: "
@@ -208,6 +209,8 @@ def show(key: str) -> bool:
             file_name = attachment["file_name"]
             self.logger.info(f"attachment: {file_name}")
 
+        self.logger.info(f"chapters: {'yes' if chapters else 'no'}")
+
         # more details for debug
         for stream_type, streams in tracks.items():
             self.logger.debug(f"\t{stream_type}:")
@@ -232,14 +235,19 @@ def _print_streams_details(self, common_prefix, all_streams: List):
                 self.logger.info(f"{stype} track ID #{tid} with language {language} from {printable_path}")
 
     def _print_attachements_details(self, common_prefix, all_attachments: List):
-         for stream in all_attachments:
+        for stream in all_attachments:
             path = stream[0]
             tid = stream[1]
 
             printable_path = files_utils.get_printable_path(path, common_prefix)
             self.logger.info(f"Attachment ID #{tid} from {printable_path}")
 
-    def _process_duplicates(self, duplicates: List[str]) -> List[Dict] | None:
+    def _print_chapter_details(self, common_prefix, chapter: str | None):
+        if chapter:
+            printable_path = files_utils.get_printable_path(chapter, common_prefix)
+            self.logger.info(f"Chapters from {printable_path}")
+
+    def _process_duplicates(self, duplicates: List[str]) -> Tuple[Dict, List, str | None] | None:
         self.logger.info("------------------------------------")
         self.logger.info("Processing group of duplicated files")
         self.logger.info("------------------------------------")
@@ -248,6 +256,7 @@ def _process_duplicates(self, duplicates: List[str]) -> List[Dict] | None:
         # use mkvmerge-based probing enriched with ffprobe data
         details_full = {file: video_utils.get_video_data_mkvmerge(file, enrich=True) for file in duplicates}
         attachments = {file: info["attachments"] for file, info in details_full.items()}
+        chapters_info = {file: info.get("chapters") for file, info in details_full.items()}
         tracks = {file: info["tracks"] for file, info in details_full.items()}
 
         common_prefix = files_utils.get_common_prefix(duplicates)
@@ -272,12 +281,16 @@ def _process_duplicates(self, duplicates: List[str]) -> List[Dict] | None:
             self.logger.error(re)
             return None
 
-        picked_attachments = AttachmentsPicker(self.logger).pick_attachments(attachments)
+        best_video_file = video_streams[0][0]
+        picker = AttachmentsPicker(self.logger)
+        picked_attachments = picker.pick_attachments(attachments, best_video_file)
+        picked_chapter = picker.pick_chapter(chapters_info, best_video_file)
 
         # print proposed output file
         self.logger.info("Streams used to create output video file:")
         self._print_streams_details(common_prefix, [(stype, streams) for stype, streams in zip(["video", "audio", "subtitle"], [video_streams, audio_streams, subtitle_streams])])
         self._print_attachements_details(common_prefix, picked_attachments)
+        self._print_chapter_details(common_prefix, picked_chapter)
 
         # build streams mapping
         streams = defaultdict(list)
@@ -349,7 +362,7 @@ def _process_duplicates(self, duplicates: List[str]) -> List[Dict] | None:
                 "type": "subtitle",
             })
 
-        return streams, picked_attachments
+        return streams, picked_attachments, picked_chapter
 
     def _process_duplicates_set(self, duplicates: Dict[str, List[str]]):
         def process_entries(entries: List[str]) -> List[Tuple[List[str], str]]:
@@ -417,13 +430,15 @@ def file_without_ext(path: str) -> str:
                     self.logger.info("Skipping output generation")
                     continue
 
-                streams, attachments = result
+                streams, attachments, chapter = result
                 if not self.live_run:
                     self.logger.info("Dry run. Skipping output generation")
                     continue
 
                 required_input_files = { file_path for file_path in streams }
                 required_input_files |= { info[0] for info in attachments }
+                if chapter:
+                    required_input_files.add(chapter)
 
                 output = os.path.join(self.output, title, output_name + ".mkv")
                 if os.path.exists(output):
@@ -528,6 +543,11 @@ def find_preferred(stype: str):
                     if track_order:
                         generation_args.extend(["--track-order", ",".join(track_order)])
 
+                    if chapter:
+                        generation_args.extend(["--chapters", chapter])
+                    else:
+                        generation_args.append("--no-chapters")
+
                     process_utils.raise_on_error(process_utils.start_process("mkvmerge", generation_args, show_progress = True))
 
                     self.logger.info(f"{output} saved.")