diff --git a/README.md b/README.md index cac0afc..1cc05d5 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,28 @@ python3 media_shrinker.py .. \ Outputs are written under `../under_2gb/`. Existing generated output directories and `split_over*` directories should be excluded from scans to avoid reconverting generated media. Files under 2GB are included by default; use `--over-limit-only` only when intentionally processing oversized sources exclusively. +## Config file for repeat workflows + +Instead of re-typing long flag sets, store them once in a `.codec-carver.json` file in the scan root (checked first) or the current working directory: + +```json +{ + "flac_all": true, + "exclude_dir_prefix": ["split_over"], + "max_duration_seconds": 14400, + "workers": 2, + "output_dir": "under_2gb" +} +``` + +Then repeat runs collapse to `python3 media_shrinker.py .. --execute --download-icloud`. + +- Keys map 1:1 to CLI options with dashes replaced by underscores (`--target-bytes` becomes `target_bytes`). +- Explicit CLI flags always override config values; without a config file, behavior is identical to a plain invocation. +- `root` and `--execute` are intentionally not configurable: the config file is discovered via the scan root, and a config file must never silently turn a dry run into a real conversion. +- Unknown keys, wrong value types, and malformed JSON abort with a clear error listing the valid keys. +- JSON is used instead of TOML because the stdlib TOML parser requires Python 3.11+, while this project also supports Python 3.10. + ## Duration splitting - `--max-duration-seconds 14400` keeps every generated file below four hours. diff --git a/config_file.py b/config_file.py new file mode 100644 index 0000000..920daaf --- /dev/null +++ b/config_file.py @@ -0,0 +1,190 @@ +#!/usr/bin/env python3 +"""Optional per-project config file support for media_shrinker. + +Users who run the same shrink workflow repeatedly can store their flag set in +a ``.codec-carver.json`` file at the scan root (or the current working +directory) instead of re-typing long command lines. Values from the config +file are applied only where the user did not pass an explicit CLI flag, so +the command line always wins and a missing config file leaves behavior +byte-identical to a plain invocation. + +Format choice: JSON instead of TOML. The stdlib TOML parser (``tomllib``) +only exists on Python 3.11+, while this project also supports Python 3.10. +``json`` is available on every supported interpreter, needs no third-party +dependency, and covers the flat key/value shape this file requires. + +Example ``.codec-carver.json``:: + + { + "target_bytes": 500000000, + "flac_all": true, + "workers": 2, + "exclude_dir_prefix": ["split_", "tmp_"] + } + +Keys map 1:1 to ``media_shrinker`` CLI options with dashes replaced by +underscores (``--target-bytes`` becomes ``target_bytes``). The ``root`` +positional and the ``--execute`` flag are deliberately not configurable: +``root`` is where the config file itself is discovered, and ``--execute`` +must stay an explicit per-run decision so a config file can never silently +turn a dry run into a real conversion. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + + +CONFIG_FILENAME = ".codec-carver.json" + + +class ConfigFileError(ValueError): + """Raised when a config file exists but cannot be used safely.""" + + +def _bool_value(key: str, value: Any) -> bool: + """Return value if it is a JSON boolean, else raise ConfigFileError.""" + + if isinstance(value, bool): + return value + raise ConfigFileError(_type_error_message(key, "a boolean (true/false)", value)) + + +def _int_value(key: str, value: Any) -> int: + """Return value if it is a JSON integer, else raise ConfigFileError.""" + + if isinstance(value, int) and not isinstance(value, bool): + return value + raise ConfigFileError(_type_error_message(key, "an integer", value)) + + +def _float_value(key: str, value: Any) -> float: + """Return value as float if it is a JSON number, else raise ConfigFileError.""" + + if isinstance(value, (int, float)) and not isinstance(value, bool): + return float(value) + raise ConfigFileError(_type_error_message(key, "a number", value)) + + +def _str_value(key: str, value: Any) -> str: + """Return value if it is a JSON string, else raise ConfigFileError.""" + + if isinstance(value, str): + return value + raise ConfigFileError(_type_error_message(key, "a string", value)) + + +def _path_value(key: str, value: Any) -> Path: + """Return value converted to a Path if it is a JSON string.""" + + return Path(_str_value(key, value)) + + +def _str_list_value(key: str, value: Any) -> list[str]: + """Return value if it is a JSON array of strings, else raise ConfigFileError.""" + + if isinstance(value, list) and all(isinstance(item, str) for item in value): + return list(value) + raise ConfigFileError(_type_error_message(key, "an array of strings", value)) + + +def _type_error_message(key: str, expected: str, value: Any) -> str: + """Build a consistent type-mismatch error message for one config key.""" + + return ( + f"Config key '{key}' expects {expected}, " + f"got {type(value).__name__}: {value!r}" + ) + + +#: Allowlisted config keys mapped to their validating converter. Each key is a +#: real ``media_shrinker.parse_args`` argparse dest so validated values can be +#: assigned onto the parsed namespace unchanged. +CONFIG_SCHEMA = { + "size_limit_bytes": _int_value, + "target_bytes": _int_value, + "max_duration_seconds": _float_value, + "output_dir": _path_value, + "report": _path_value, + "ffmpeg": _str_value, + "ffprobe": _str_value, + "brctl": _str_value, + "download_icloud": _bool_value, + "include_under_limit": _bool_value, + "exclude_dir_prefix": _str_list_value, + "flac_all": _bool_value, + "workers": _int_value, + "ffmpeg_threads": _int_value, + "silence_noise": _str_value, + "silence_min_duration_seconds": _float_value, + "overwrite": _bool_value, +} + + +def find_config_path(root: Path) -> Path | None: + """Return the config file path for a run, or None when no file exists. + + The scan root is checked first so per-library configs travel with the + media they describe; the current working directory is the fallback for + users who keep one config next to where they launch the tool. + """ + + for base in (Path(root), Path.cwd()): + candidate = base / CONFIG_FILENAME + if candidate.is_file(): + return candidate + return None + + +def validate_config(raw: dict[str, Any], *, source: Path) -> dict[str, Any]: + """Validate raw config data against CONFIG_SCHEMA and convert values. + + Raises: + ConfigFileError: If any key is unknown or any value has the wrong type. + """ + + unknown = sorted(set(raw) - set(CONFIG_SCHEMA)) + if unknown: + raise ConfigFileError( + f"Unknown config key(s) in {source}: {', '.join(unknown)}. " + f"Valid keys: {', '.join(sorted(CONFIG_SCHEMA))}" + ) + return {key: CONFIG_SCHEMA[key](key, value) for key, value in raw.items()} + + +def load_config(root: Path) -> dict[str, Any]: + """Load and validate the ``.codec-carver.json`` config for a scan root. + + Looks for the file in ``root`` first, then in the current working + directory. Returns an empty dict when no config file exists, so callers + can treat "no config" and "empty config" identically. + + Raises: + ConfigFileError: If the file exists but is unreadable, is not valid + JSON, is not a JSON object, contains unknown keys, or contains + values of the wrong type. + """ + + path = find_config_path(root) + if path is None: + return {} + + try: + text = path.read_text(encoding="utf-8") + except OSError as exc: + raise ConfigFileError(f"Could not read config file {path}: {exc}") from exc + + try: + raw = json.loads(text) + except json.JSONDecodeError as exc: + raise ConfigFileError(f"Malformed JSON in config file {path}: {exc}") from exc + + if not isinstance(raw, dict): + raise ConfigFileError( + f"Config file {path} must contain a JSON object at the top level, " + f"got {type(raw).__name__}" + ) + + return validate_config(raw, source=path) diff --git a/media_shrinker.py b/media_shrinker.py index d72a8d5..d50b7e5 100644 --- a/media_shrinker.py +++ b/media_shrinker.py @@ -23,6 +23,8 @@ from pathlib import Path from typing import Any, Iterable +import config_file + SUPPORTED_EXTENSIONS = { ".3gp", @@ -1142,8 +1144,8 @@ def _normalize_argv(argv: list[str] | None) -> list[str] | None: return normalized -def parse_args(argv: list[str] | None = None) -> argparse.Namespace: - """Parse CLI arguments.""" +def _build_parser() -> argparse.ArgumentParser: + """Build the CLI argument parser.""" parser = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.ArgumentDefaultsHelpFormatter @@ -1247,7 +1249,68 @@ 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)) + return parser + + +def _explicitly_set_dests( + parser: argparse.ArgumentParser, + argv: list[str] | None, + dests: Iterable[str], +) -> set[str]: + """Return the subset of dests the user explicitly set on the command line. + + Uses the documented argparse behavior that pre-set namespace attributes + are kept unless an option is actually provided: each dest is seeded with + a unique sentinel, argv is re-parsed onto that namespace, and any dest + that no longer holds the sentinel was explicitly given by the user. + """ + + sentinel = object() + dests = tuple(dests) + namespace = argparse.Namespace(**{dest: sentinel for dest in dests}) + parser.parse_args(argv, namespace=namespace) + return {dest for dest in dests if getattr(namespace, dest) is not sentinel} + + +def _apply_config_file( + parser: argparse.ArgumentParser, + args: argparse.Namespace, + argv: list[str] | None, +) -> None: + """Overlay `.codec-carver.json` values onto args for unset CLI options. + + Explicit CLI flags always win over config-file values; with no config + file present, args is returned untouched. Config problems (unknown keys, + type mismatches, malformed JSON) abort with a normal argparse error + message instead of a traceback. + """ + + try: + config = config_file.load_config(Path(args.root)) + except config_file.ConfigFileError as exc: + parser.error(str(exc)) + if not config: + return + explicit = _explicitly_set_dests(parser, argv, config) + for dest, value in config.items(): + if dest not in explicit: + setattr(args, dest, value) + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + """Parse CLI arguments, overlaying `.codec-carver.json` config defaults. + + A `.codec-carver.json` file in the scan root (or the current working + directory) supplies defaults for repeat workflows; see the config_file + module for the format and the allowed keys. Values from the config file + apply only where the user did not pass the corresponding CLI flag. + """ + + parser = _build_parser() + normalized = _normalize_argv(argv) + args = parser.parse_args(normalized) + _apply_config_file(parser, args, normalized) + return args def _execute_conversions( diff --git a/tests/test_config_file.py b/tests/test_config_file.py new file mode 100644 index 0000000..4a79540 --- /dev/null +++ b/tests/test_config_file.py @@ -0,0 +1,307 @@ +import contextlib +import io +import json +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +import config_file +import media_shrinker + + +def _write_config(directory: Path, payload) -> Path: + path = directory / config_file.CONFIG_FILENAME + path.write_text(json.dumps(payload), encoding="utf-8") + return path + + +@contextlib.contextmanager +def _chdir(path: Path): + previous = Path.cwd() + os.chdir(path) + try: + yield + finally: + os.chdir(previous) + + +class LoadConfigTests(unittest.TestCase): + def test_absent_file_returns_empty_dict(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + with _chdir(root): + self.assertEqual(config_file.load_config(root), {}) + + def test_valid_config_is_loaded_and_converted(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + _write_config( + root, + { + "target_bytes": 123, + "max_duration_seconds": 60, + "output_dir": "converted", + "flac_all": True, + "exclude_dir_prefix": ["split_", "tmp_"], + }, + ) + config = config_file.load_config(root) + + self.assertEqual(config["target_bytes"], 123) + self.assertEqual(config["max_duration_seconds"], 60.0) + self.assertIsInstance(config["max_duration_seconds"], float) + self.assertEqual(config["output_dir"], Path("converted")) + self.assertIs(config["flac_all"], True) + self.assertEqual(config["exclude_dir_prefix"], ["split_", "tmp_"]) + + def test_root_config_wins_over_cwd_config(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + base = Path(temp_dir) + root = base / "root" + cwd = base / "cwd" + root.mkdir() + cwd.mkdir() + _write_config(root, {"workers": 3}) + _write_config(cwd, {"workers": 7}) + with _chdir(cwd): + config = config_file.load_config(root) + + self.assertEqual(config["workers"], 3) + + def test_cwd_config_used_when_root_has_none(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + base = Path(temp_dir) + root = base / "root" + cwd = base / "cwd" + root.mkdir() + cwd.mkdir() + _write_config(cwd, {"workers": 7}) + with _chdir(cwd): + config = config_file.load_config(root) + + self.assertEqual(config["workers"], 7) + + def test_unknown_key_error_lists_valid_keys(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + _write_config(root, {"bogus_key": 1, "target_bytes": 5}) + with self.assertRaises(config_file.ConfigFileError) as ctx: + config_file.load_config(root) + + message = str(ctx.exception) + self.assertIn("bogus_key", message) + self.assertIn("Valid keys:", message) + for valid_key in sorted(config_file.CONFIG_SCHEMA): + self.assertIn(valid_key, message) + + def test_execute_is_not_a_valid_config_key(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + _write_config(root, {"execute": True}) + with self.assertRaises(config_file.ConfigFileError) as ctx: + config_file.load_config(root) + + self.assertIn("execute", str(ctx.exception)) + + def test_malformed_json_raises_clear_error(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + (root / config_file.CONFIG_FILENAME).write_text( + "{not json", encoding="utf-8" + ) + with self.assertRaises(config_file.ConfigFileError) as ctx: + config_file.load_config(root) + + self.assertIn("Malformed JSON", str(ctx.exception)) + + def test_non_object_top_level_raises_clear_error(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + (root / config_file.CONFIG_FILENAME).write_text( + "[1, 2]", encoding="utf-8" + ) + with self.assertRaises(config_file.ConfigFileError) as ctx: + config_file.load_config(root) + + self.assertIn("JSON object", str(ctx.exception)) + + def test_unreadable_file_raises_clear_error(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + _write_config(root, {"workers": 1}) + with patch.object(Path, "read_text", side_effect=OSError("boom")): + with self.assertRaises(config_file.ConfigFileError) as ctx: + config_file.load_config(root) + + self.assertIn("Could not read config file", str(ctx.exception)) + + +class TypeValidationTests(unittest.TestCase): + def _assert_type_error(self, payload, expected_fragment: str) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + _write_config(root, payload) + with self.assertRaises(config_file.ConfigFileError) as ctx: + config_file.load_config(root) + self.assertIn(expected_fragment, str(ctx.exception)) + + def test_int_key_rejects_string(self) -> None: + self._assert_type_error({"target_bytes": "big"}, "expects an integer") + + def test_int_key_rejects_bool(self) -> None: + self._assert_type_error({"workers": True}, "expects an integer") + + def test_float_key_rejects_string(self) -> None: + self._assert_type_error({"max_duration_seconds": "60"}, "expects a number") + + def test_float_key_rejects_bool(self) -> None: + self._assert_type_error( + {"silence_min_duration_seconds": True}, "expects a number" + ) + + def test_bool_key_rejects_int(self) -> None: + self._assert_type_error({"flac_all": 1}, "expects a boolean") + + def test_str_key_rejects_number(self) -> None: + self._assert_type_error({"silence_noise": -35}, "expects a string") + + def test_path_key_rejects_number(self) -> None: + self._assert_type_error({"output_dir": 5}, "expects a string") + + def test_str_list_key_rejects_scalar(self) -> None: + self._assert_type_error( + {"exclude_dir_prefix": "split_"}, "expects an array of strings" + ) + + def test_str_list_key_rejects_mixed_items(self) -> None: + self._assert_type_error( + {"exclude_dir_prefix": ["split_", 3]}, "expects an array of strings" + ) + + +class ParseArgsConfigIntegrationTests(unittest.TestCase): + def test_config_in_root_supplies_defaults(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + _write_config( + root, + { + "target_bytes": 123, + "flac_all": True, + "workers": 2, + "output_dir": "converted", + "include_under_limit": False, + }, + ) + args = media_shrinker.parse_args([str(root)]) + + self.assertEqual(args.target_bytes, 123) + self.assertTrue(args.flac_all) + self.assertEqual(args.workers, 2) + self.assertEqual(args.output_dir, Path("converted")) + self.assertFalse(args.include_under_limit) + + def test_cli_flag_overrides_config_value(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + _write_config(root, {"target_bytes": 123, "workers": 2}) + args = media_shrinker.parse_args( + [str(root), "--target-bytes", "9"] + ) + + self.assertEqual(args.target_bytes, 9) + self.assertEqual(args.workers, 2) + + def test_cli_flag_overrides_config_even_when_equal_to_default(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + _write_config( + root, {"target_bytes": 123, "include_under_limit": False} + ) + args = media_shrinker.parse_args( + [ + str(root), + "--target-bytes", + str(media_shrinker.DEFAULT_TARGET_BYTES), + "--include-under-limit", + ] + ) + + self.assertEqual(args.target_bytes, media_shrinker.DEFAULT_TARGET_BYTES) + self.assertTrue(args.include_under_limit) + + def test_absent_config_keeps_defaults(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + with _chdir(root): + args = media_shrinker.parse_args([str(root)]) + + self.assertEqual(args.target_bytes, media_shrinker.DEFAULT_TARGET_BYTES) + self.assertEqual( + args.size_limit_bytes, media_shrinker.DEFAULT_SIZE_LIMIT_BYTES + ) + self.assertEqual(args.output_dir, Path("under_2gb")) + self.assertFalse(args.flac_all) + self.assertEqual(args.workers, 0) + + def test_unknown_config_key_exits_with_clear_error(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + _write_config(root, {"bogus_key": 1}) + stderr = io.StringIO() + with contextlib.redirect_stderr(stderr): + with self.assertRaises(SystemExit) as ctx: + media_shrinker.parse_args([str(root)]) + + self.assertEqual(ctx.exception.code, 2) + self.assertIn("bogus_key", stderr.getvalue()) + self.assertIn("Valid keys:", stderr.getvalue()) + + def test_malformed_config_exits_without_traceback(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + (root / config_file.CONFIG_FILENAME).write_text( + "{oops", encoding="utf-8" + ) + stderr = io.StringIO() + with contextlib.redirect_stderr(stderr): + with self.assertRaises(SystemExit) as ctx: + media_shrinker.parse_args([str(root)]) + + self.assertEqual(ctx.exception.code, 2) + self.assertIn("Malformed JSON", stderr.getvalue()) + + def test_type_mismatch_exits_with_clear_error(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + _write_config(root, {"target_bytes": "big"}) + stderr = io.StringIO() + with contextlib.redirect_stderr(stderr): + with self.assertRaises(SystemExit) as ctx: + media_shrinker.parse_args([str(root)]) + + self.assertEqual(ctx.exception.code, 2) + self.assertIn("expects an integer", stderr.getvalue()) + + def test_main_dry_run_uses_config_size_limit(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + source = root / "source.wav" + source.write_bytes(b"1234") + _write_config( + root, {"size_limit_bytes": 1, "include_under_limit": False} + ) + + with patch("builtins.print") as mock_print: + rc = media_shrinker.main([str(root)]) + + self.assertEqual(rc, 0) + printed = ["\t".join(map(str, call.args)) for call in mock_print.call_args_list] + self.assertIn(f"DRY-RUN\t4\t{source.name}", printed) + self.assertIn("TOTAL_SELECTED=1", printed) + + +if __name__ == "__main__": + unittest.main()