Source code for remote_compression.config

"""User preferences: TOML file in the platform config directory."""

import logging
import tomllib
from dataclasses import fields
from pathlib import Path

from platformdirs import user_config_dir

from remote_compression.command import FORBIDDEN_CHARS
from remote_compression.settings import Settings

logger = logging.getLogger("rcomp")

_FIELDS = {f.name for f in fields(Settings)}

#: TOML template written by ``rcomp config --init`` (every key commented out).
CONFIG_TEMPLATE = """\
# rcomp configuration - command-line flags always win over this file.
# Resolution order:
#   built-in defaults < [defaults] < [preset.X] (with --preset X) < explicit CLI flags.

[defaults]
# hostname = "remote_host"      # ssh alias (~/.ssh/config), or "local"
# remote_ffmpeg = "/usr/local/bin/ffmpeg"  # explicit remote binary (space-free path); omitted = PATH
# remote_workdir = ".rcomp"            # remote workspace, as the ffmpeg shell sees it
# remote_workdir_sftp = "/home/.rcomp"  # same workspace, as SFTP sees it, when the two views
#                                       # differ (Synology chroots SFTP: /home = the login home)
# codec = "libx265"             # ffmpeg encoder (free choice: libx264, libsvtav1, hevc_nvenc, ...)
# codec_name = "hevc"           # ffprobe codec name considered compliant; derived from codec when omitted
# crf = 28                      # constant rate factor; omitted = encoder default
# ffmpeg_preset = "medium"      # ffmpeg -preset; omitted = encoder default
# height = 720                  # maximal height; 0 = never downscale
# container = "mkv"             # output container; mkv swallows about anything
# map = true                    # explicit video/audio/subtitles stream mapping
# replace = false               # true: replace originals; false: keep them as ori_*
# extensions = [".avi", ".mp4", ".flv", ".wmv", ".mkv", ".ts", ".webm", ".m4v",
#               ".mov", ".mpg", ".mpeg", ".m2ts", ".mts", ".3gp"]
# auto_local = true             # switch to local mode when the host is this machine
# auto_purge_days = 7           # opportunistic remote workspace purge; 0 disables

# Named presets use the same keys as [defaults]. Apply with --preset NAME.
# Built-in presets (soft, hard, hard4) can be overridden here.

# [preset.hard]
# map = false
# replace = true

# [preset.av1]
# codec = "libsvtav1"
# crf = 32
"""


[docs] class ConfigError(ValueError): """Invalid configuration (bad TOML, unknown preset...)."""
[docs] def config_path(): """ Returns ------- :class:`~pathlib.Path` Location of the user configuration file. """ return Path(user_config_dir(appname="rcomp", appauthor=False)) / "config.toml"
[docs] def builtin_presets(): """ Returns ------- :class:`dict` Built-in presets as plain option dicts (overridable from the config file). """ return { "soft": {}, "hard": {"map_streams": False, "replace": True}, "hard4": {"map_streams": False, "replace": True, "codec": "libx264"}, }
[docs] def normalize_options(options): """ Normalize one layer of options to :class:`~remote_compression.settings.Settings` fields. Parameters ---------- options: :class:`dict` Raw options (TOML table or CLI overrides). Returns ------- :class:`dict` Normalized options: ``map`` renamed to ``map_streams``, sentinel ``0`` turned into None for `height` and `auto_purge_days`, extensions lower-cased as a frozenset. Unknown keys are dropped with a warning. """ normalized = {} for key, value in options.items(): if key == "map": key = "map_streams" if key not in _FIELDS: logger.warning("Ignoring unknown configuration key '%s'.", key) continue if key in {"height", "auto_purge_days"} and value == 0: value = None elif key == "extensions": value = frozenset("." + str(e).lstrip(".").lower() for e in value) elif key == "auto_purge_days" and value is not None: value = float(value) elif key in {"remote_ffmpeg", "remote_workdir", "remote_workdir_sftp"} and value is not None: value = str(value) if not value.isascii() or FORBIDDEN_CHARS & set(value) or " " in value: msg = ( f"{key} must be a space-free ASCII path (got {value!r}), " "e.g. /usr/local/bin/ffmpeg or /volume1/homes/me/rcomp" ) raise ConfigError(msg) normalized[key] = value return normalized
[docs] def load_config(path=None): """ Load the user configuration. Parameters ---------- path: :class:`~pathlib.Path` or :class:`str`, optional Alternative file location (defaults to :func:`config_path`). Returns ------- :class:`dict` ``{'defaults': {...}, 'presets': {name: {...}}}``, normalized. Empty tables when the file does not exist. Raises ------ ConfigError If the file exists but is not valid TOML. """ path = Path(path) if path is not None else config_path() if not path.exists(): return {"defaults": {}, "presets": {}} try: with path.open("rb") as f: data = tomllib.load(f) except tomllib.TOMLDecodeError as e: msg = f"Invalid TOML in {path}: {e}" raise ConfigError(msg) from e for section in data: if section not in {"defaults", "preset"}: logger.warning("Ignoring unknown configuration section '%s'.", section) defaults = normalize_options(data.get("defaults", {})) presets = {name: normalize_options(opts) for name, opts in data.get("preset", {}).items()} return {"defaults": defaults, "presets": presets}
[docs] def resolve_settings(config, preset=None, cli_overrides=None): """ Merge all option layers into a :class:`~remote_compression.settings.Settings`. Parameters ---------- config: :class:`dict` Output of :func:`load_config`. preset: :class:`str`, optional Name of a preset (built-in or from the config file). cli_overrides: :class:`dict`, optional Options *explicitly typed* on the command line (Settings field names). Returns ------- :class:`~remote_compression.settings.Settings` Resolved settings: dataclass defaults < config ``[defaults]`` < ``[preset.X]`` < explicit CLI flags. Raises ------ ConfigError If `preset` is not a known preset. """ merged = dict(config.get("defaults", {})) if preset is not None: presets = {**builtin_presets(), **config.get("presets", {})} if preset not in presets: msg = f"Unknown preset '{preset}'. Available presets: {', '.join(sorted(presets))}." raise ConfigError(msg) merged.update(presets[preset]) if cli_overrides: merged.update(normalize_options(cli_overrides)) return Settings(**merged)
[docs] def write_template(path=None, force=False): """ Write the commented configuration template. Parameters ---------- path: :class:`~pathlib.Path` or :class:`str`, optional Alternative file location (defaults to :func:`config_path`). force: :class:`bool` Overwrite an existing file. Returns ------- :class:`~pathlib.Path` The written file. Raises ------ ConfigError If the file exists and `force` is False. """ path = Path(path) if path is not None else config_path() if path.exists() and not force: msg = f"{path} already exists. Use --force to overwrite it." raise ConfigError(msg) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(CONFIG_TEMPLATE, encoding="utf-8") return path