From 7d48f40fe97902eb77763e130a5eb4be4259fac8 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 6 Jul 2026 09:43:22 +0900 Subject: [PATCH] feat: add output metadata tagging via --set-title/--set-artist/--set-album/--set-comment Add four CLI options that inject -metadata key=value pairs into the generated ffmpeg commands so archived outputs carry meaningful, searchable tags in players and music libraries. - New _metadata_args() helper emits deterministic (sorted-key) -metadata pairs; values are passed through verbatim as single argv items (plans run without a shell), so special characters are safe. - Tags are threaded through the same chain as prefer_flac: parse_args -> _execute_conversions -> convert_file -> _convert_segment -> _execute_segment_conversion -> build_audio_plan/build_opus_plan, including the FLAC-too-large -> Opus fallback path. - Pairs are injected after -map_metadata 0 / -map_chapters 0 in both the FLAC and Opus plans, so each provided key overrides that specific tag copied from the source while all other source metadata is preserved (standard ffmpeg semantics). - Default behavior is byte-identical: with no --set-* options the generated argument lists are unchanged. - Tests: parse_args tag collection (all/partial/none), -metadata pair placement in both plans, byte-identical no-tags plans, special-character pass-through, and tags forwarding in _execute_conversions. - Docs: README "Metadata tagging" section. Gates: coverage 100% on media_shrinker.py, interrogate 100%, unittest baseline unchanged (5 pre-existing macOS os.listxattr errors only). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01CJRVbDrp1vGYkJgNHMGPpG --- README.md | 13 +++ media_shrinker.py | 93 ++++++++++++++- tests/test_media_shrinker.py | 214 +++++++++++++++++++++++++++++++++++ 3 files changed, 315 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index cac0afc..633f836 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,19 @@ Outputs are written under `../under_2gb/`. Existing generated output directories - Split outputs are named with part suffixes, for example `meeting.wav.part0001.flac`, `meeting.wav.part0002.flac`. - Tune silence detection with `--silence-noise` and `--silence-min-duration-seconds` when recordings need stricter or looser silence boundaries. +## Metadata tagging + +- `--set-title`, `--set-artist`, `--set-album`, and `--set-comment` stamp the corresponding tags on every generated output, so archived files stay searchable in players and music libraries. +- Generated commands already copy source metadata with `-map_metadata 0`; the `--set-*` values are injected after it, so each provided key overrides that specific source tag while all other source metadata is preserved (standard ffmpeg semantics). +- When none of the `--set-*` options are passed, generated ffmpeg commands are byte-identical to the untagged behavior. +- Values are passed to ffmpeg as single argv items without a shell, so spaces, quotes, and other special characters are safe as given. + +```bash +python3 media_shrinker.py .. --execute \ + --set-album "Board Meetings 2026" \ + --set-comment "archived by codec-carver" +``` + ## Safety notes - Source files selected by the scan are protected from deletion or overwrite; keep `--output-dir` as a generated-only directory so excluded originals are never mistaken for stale generated outputs. diff --git a/media_shrinker.py b/media_shrinker.py index d72a8d5..45e6f95 100644 --- a/media_shrinker.py +++ b/media_shrinker.py @@ -312,6 +312,28 @@ def calculate_audio_bitrate( return bitrate +def _metadata_args(tags: dict[str, str] | None) -> list[str]: + """Return ffmpeg ``-metadata key=value`` arguments for the given tags. + + Keys are emitted in sorted order so generated commands are deterministic. + The pairs are meant to be inserted after ``-map_metadata 0`` in a plan: + ffmpeg applies them on top of the metadata copied from the source, so each + provided key overrides the corresponding source tag while every other + source tag is preserved. Values are passed through verbatim as single + argv items (plans run without a shell), so special characters are safe. + + Returns an empty list when tags is None or empty, keeping default plans + byte-identical to those built before metadata tagging existed. + """ + + if not tags: + return [] + args: list[str] = [] + for key in sorted(tags): + args.extend(("-metadata", f"{key}={tags[key]}")) + return args + + def build_audio_plan( source_path: Path, probe: MediaProbe, @@ -321,8 +343,14 @@ def build_audio_plan( prefer_flac: bool = False, ffmpeg_threads: int | None = None, segment: MediaSegment | None = None, + tags: dict[str, str] | None = None, ) -> ConversionPlan: - """Build the preferred audio-only conversion plan for source_path.""" + """Build the preferred audio-only conversion plan for source_path. + + When tags is provided, ``-metadata key=value`` pairs are injected after + ``-map_metadata 0`` so they override those specific keys copied from the + source while leaving all other source metadata intact. + """ if probe.has_video: raise MediaShrinkerError( @@ -356,6 +384,7 @@ def build_audio_plan( "0", "-map_chapters", "0", + *_metadata_args(tags), "-vn", "-c:a", "flac", @@ -379,6 +408,7 @@ def build_audio_plan( output_dir=output_dir, ffmpeg_threads=ffmpeg_threads, segment=segment, + tags=tags, ) @@ -390,8 +420,14 @@ def build_opus_plan( output_dir: Path, ffmpeg_threads: int | None = None, segment: MediaSegment | None = None, + tags: dict[str, str] | None = None, ) -> ConversionPlan: - """Build a high-quality Opus plan that fits the target size.""" + """Build a high-quality Opus plan that fits the target size. + + When tags is provided, ``-metadata key=value`` pairs are injected after + ``-map_metadata 0`` so they override those specific keys copied from the + source while leaving all other source metadata intact. + """ duration_seconds = ( segment.duration_seconds if segment is not None else probe.duration_seconds @@ -420,6 +456,7 @@ def build_opus_plan( "0", "-map_chapters", "0", + *_metadata_args(tags), "-vn", "-c:a", "libopus", @@ -727,8 +764,13 @@ def convert_file( protected_sources: Iterable[Path] = (), resolved_protected_sources: frozenset[Path] | None = None, original_size: int | None = None, + tags: dict[str, str] | None = None, ) -> list[ConversionResult]: - """Convert one file and return generated segment results without deleting the source.""" + """Convert one file and return generated segment results without deleting the source. + + When tags is provided, each generated output is stamped with those + ``-metadata`` key/value pairs on top of the metadata copied from the source. + """ source = Path(source) root = Path(root) @@ -777,6 +819,7 @@ def convert_file( overwrite=overwrite, max_segment_duration_seconds=max_segment_duration_seconds, protected_sources=resolved_sources, + tags=tags, ) for segment in segments ] @@ -874,6 +917,7 @@ def _execute_segment_conversion( overwrite: bool, max_segment_duration_seconds: float, resolved_protected_sources: frozenset[Path], + tags: dict[str, str] | None = None, ) -> ConversionResult: """Execute the conversion plan(s) for a single segment.""" plan = build_audio_plan( @@ -884,6 +928,7 @@ def _execute_segment_conversion( prefer_flac=prefer_flac, ffmpeg_threads=ffmpeg_threads, segment=segment, + tags=tags, ) final_output = _resolve_collision(plan.output_path, overwrite=overwrite) @@ -910,6 +955,7 @@ def _execute_segment_conversion( output_dir=output_dir, ffmpeg_threads=ffmpeg_threads, segment=segment, + tags=tags, ) final_output = _resolve_collision(opus_plan.output_path, overwrite=overwrite) first_result = _execute_plan( @@ -994,6 +1040,7 @@ def _convert_segment( overwrite: bool, max_segment_duration_seconds: float, protected_sources: frozenset[Path] = frozenset(), + tags: dict[str, str] | None = None, ) -> ConversionResult: """Convert one media segment fitting the target size limit.""" # protected_sources is passed from convert_file where it is already fully resolved. @@ -1043,6 +1090,7 @@ def _convert_segment( overwrite=overwrite, max_segment_duration_seconds=max_segment_duration_seconds, resolved_protected_sources=resolved_protected_sources, + tags=tags, ) @@ -1143,7 +1191,13 @@ def _normalize_argv(argv: list[str] | None) -> list[str] | None: def parse_args(argv: list[str] | None = None) -> argparse.Namespace: - """Parse CLI arguments.""" + """Parse CLI arguments. + + The returned namespace also carries a ``tags`` dict collecting the + ``--set-title``/``--set-artist``/``--set-album``/``--set-comment`` values + that were actually provided; it is empty when none were passed, which + keeps generated ffmpeg commands byte-identical to the untagged behavior. + """ parser = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.ArgumentDefaultsHelpFormatter @@ -1247,7 +1301,35 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: action="store_true", help="Allow overwriting generated output paths", ) - return parser.parse_args(_normalize_argv(argv)) + parser.add_argument( + "--set-title", + default=None, + help="Set the 'title' metadata tag on every generated output", + ) + parser.add_argument( + "--set-artist", + default=None, + help="Set the 'artist' metadata tag on every generated output", + ) + parser.add_argument( + "--set-album", + default=None, + help="Set the 'album' metadata tag on every generated output", + ) + parser.add_argument( + "--set-comment", + default=None, + help="Set the 'comment' metadata tag on every generated output", + ) + args = parser.parse_args(_normalize_argv(argv)) + provided_tags = { + "title": args.set_title, + "artist": args.set_artist, + "album": args.set_album, + "comment": args.set_comment, + } + args.tags = {key: value for key, value in provided_tags.items() if value is not None} + return args def _execute_conversions( @@ -1286,6 +1368,7 @@ def process_candidate(candidate_tuple: tuple[Path, int]) -> list[ConversionResul protected_sources=protected_sources, resolved_protected_sources=resolved_candidates, original_size=size, + tags=args.tags or None, ) except Exception as exc: # noqa: BLE001 - batch processing records per-file failures. return [ diff --git a/tests/test_media_shrinker.py b/tests/test_media_shrinker.py index 0ab2397..f706883 100644 --- a/tests/test_media_shrinker.py +++ b/tests/test_media_shrinker.py @@ -15,6 +15,7 @@ SilenceInterval, build_audio_plan, build_icloud_download_command, + build_opus_plan, build_segments, calculate_audio_bitrate, choose_worker_count, @@ -916,6 +917,219 @@ def test_parse_silencedetect_intervals_pairs_long_silence_start_and_end( ) +class MetadataTaggingTests(unittest.TestCase): + """Cover --set-* CLI tags flowing into ffmpeg -metadata arguments.""" + + @staticmethod + def _lossless_probe() -> MediaProbe: + return MediaProbe( + duration_seconds=3600.0, + size_bytes=4_294_808_936, + audio_codec="pcm_s16le", + audio_bit_rate=1_411_200, + has_video=False, + format_name="wav", + ) + + @staticmethod + def _lossy_probe() -> MediaProbe: + return MediaProbe( + duration_seconds=10_000.0, + size_bytes=3_000_000_000, + audio_codec="aac", + audio_bit_rate=2_500_000, + has_video=False, + format_name="mov,mp4,m4a,3gp,3g2,mj2", + ) + + def test_metadata_args_emits_pairs_in_sorted_key_order(self) -> None: + args = media_shrinker._metadata_args({"title": "T", "artist": "A"}) + + self.assertEqual(args, ["-metadata", "artist=A", "-metadata", "title=T"]) + + def test_metadata_args_returns_empty_list_for_none_and_empty(self) -> None: + self.assertEqual(media_shrinker._metadata_args(None), []) + self.assertEqual(media_shrinker._metadata_args({}), []) + + def test_flac_plan_injects_metadata_overrides_after_map_metadata(self) -> None: + plan = build_audio_plan( + Path("meeting.wav"), + self._lossless_probe(), + target_bytes=1_900_000_000, + output_dir=Path("out"), + tags={"title": "Board Meeting", "album": "Archive 2026"}, + ) + + args = plan.ffmpeg_args + self.assertEqual(args.count("-metadata"), 2) + album_index = args.index("album=Archive 2026") + title_index = args.index("title=Board Meeting") + self.assertLess(album_index, title_index) + self.assertLess(args.index("-map_metadata"), album_index) + self.assertLess(title_index, args.index("-vn")) + self.assertEqual(args[album_index - 1], "-metadata") + self.assertEqual(args[title_index - 1], "-metadata") + + def test_opus_plan_injects_metadata_overrides_after_map_metadata(self) -> None: + plan = build_opus_plan( + Path("long.m4a"), + self._lossy_probe(), + target_bytes=1_900_000_000, + output_dir=Path("out"), + tags={"artist": "Field Recorder", "comment": "auto-archived"}, + ) + + args = plan.ffmpeg_args + self.assertEqual(args.count("-metadata"), 2) + artist_index = args.index("artist=Field Recorder") + comment_index = args.index("comment=auto-archived") + self.assertLess(artist_index, comment_index) + self.assertLess(args.index("-map_metadata"), artist_index) + self.assertLess(comment_index, args.index("-vn")) + self.assertEqual(args[artist_index - 1], "-metadata") + self.assertEqual(args[comment_index - 1], "-metadata") + + def test_no_tags_keeps_plan_args_byte_identical(self) -> None: + for probe, source in ( + (self._lossless_probe(), Path("meeting.wav")), + (self._lossy_probe(), Path("long.m4a")), + ): + with self.subTest(source=source): + baseline = build_audio_plan( + source, probe, target_bytes=1_900_000_000, output_dir=Path("out") + ) + with_none = build_audio_plan( + source, + probe, + target_bytes=1_900_000_000, + output_dir=Path("out"), + tags=None, + ) + with_empty = build_audio_plan( + source, + probe, + target_bytes=1_900_000_000, + output_dir=Path("out"), + tags={}, + ) + self.assertEqual(baseline.ffmpeg_args, with_none.ffmpeg_args) + self.assertEqual(baseline.ffmpeg_args, with_empty.ffmpeg_args) + + def test_no_tags_keeps_opus_plan_args_byte_identical(self) -> None: + baseline = build_opus_plan( + Path("long.m4a"), + self._lossy_probe(), + target_bytes=1_900_000_000, + output_dir=Path("out"), + ) + with_none = build_opus_plan( + Path("long.m4a"), + self._lossy_probe(), + target_bytes=1_900_000_000, + output_dir=Path("out"), + tags=None, + ) + + self.assertEqual(baseline.ffmpeg_args, with_none.ffmpeg_args) + + def test_special_characters_pass_through_as_single_argv_items(self) -> None: + value = 'My "Great" Album; $(rm -rf /) && echo -n \'x\' 한글 = tricky' + plan = build_opus_plan( + Path("long.m4a"), + self._lossy_probe(), + target_bytes=1_900_000_000, + output_dir=Path("out"), + tags={"album": value}, + ) + + self.assertIn(f"album={value}", plan.ffmpeg_args) + metadata_index = plan.ffmpeg_args.index("-metadata") + self.assertEqual(plan.ffmpeg_args[metadata_index + 1], f"album={value}") + + def test_parse_args_collects_all_provided_tags(self) -> None: + args = media_shrinker.parse_args( + [ + "media", + "--set-title", + "Weekly Sync", + "--set-artist", + "Recorder", + "--set-album", + "Meetings", + "--set-comment", + "shrunk by codec-carver", + ] + ) + + self.assertEqual( + args.tags, + { + "title": "Weekly Sync", + "artist": "Recorder", + "album": "Meetings", + "comment": "shrunk by codec-carver", + }, + ) + + def test_parse_args_collects_only_non_none_tags(self) -> None: + args = media_shrinker.parse_args( + ["media", "--set-title", "Weekly Sync", "--set-album", "Meetings"] + ) + + self.assertEqual(args.tags, {"title": "Weekly Sync", "album": "Meetings"}) + self.assertIsNone(args.set_artist) + self.assertIsNone(args.set_comment) + + def test_parse_args_defaults_to_empty_tags(self) -> None: + args = media_shrinker.parse_args(["media"]) + + self.assertEqual(args.tags, {}) + self.assertIsNone(args.set_title) + self.assertIsNone(args.set_artist) + self.assertIsNone(args.set_album) + self.assertIsNone(args.set_comment) + + def test_execute_conversions_forwards_tags_to_convert_file(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + source = root / "ok.wav" + source.write_bytes(b"ok") + args = media_shrinker.parse_args( + [str(root), "--workers", "1", "--set-title", "Weekly Sync"] + ) + captured: dict[str, object] = {} + + def fake_convert_file(source_path: Path, **kwargs): + captured.update(kwargs) + return [] + + with patch("media_shrinker.convert_file", side_effect=fake_convert_file): + media_shrinker._execute_conversions( + [(source, 2)], args, root, root / "out" + ) + + self.assertEqual(captured["tags"], {"title": "Weekly Sync"}) + + def test_execute_conversions_passes_none_when_no_tags(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + source = root / "ok.wav" + source.write_bytes(b"ok") + args = media_shrinker.parse_args([str(root), "--workers", "1"]) + captured: dict[str, object] = {} + + def fake_convert_file(source_path: Path, **kwargs): + captured.update(kwargs) + return [] + + with patch("media_shrinker.convert_file", side_effect=fake_convert_file): + media_shrinker._execute_conversions( + [(source, 2)], args, root, root / "out" + ) + + self.assertIsNone(captured["tags"]) + + class SilenceDetectionTests(unittest.TestCase): @patch("media_shrinker.subprocess.run") def test_detect_silence_intervals_success(self, mock_run: MagicMock) -> None: