"""Console script for remote_compression."""
import dataclasses
import os
import shutil
from datetime import timedelta
from pathlib import Path
import click
from click.core import ParameterSource
from remote_compression.batch import BatchAborted, find_keep_files, run_batch
from remote_compression.config import (
ConfigError,
builtin_presets,
config_path,
load_config,
resolve_settings,
write_template,
)
from remote_compression.host import BootstrapError, RemoteHost
from remote_compression.progress import human_size, logs_dir, setup_logging
from remote_compression.settings import Settings
from remote_compression.ssh import SSHError, is_localhost
from remote_compression.tracking import Tracker, tracking_key
_DEFAULTS = Settings()
[docs]
class DefaultGroup(click.Group):
"""
Group that routes to the ``run`` subcommand when the first token is not a
known subcommand: ``rcomp movies/`` and ``rcomp -P hard .`` keep working.
Side effect: a target literally named like a subcommand needs ``rcomp run
<target>``.
"""
def parse_args(self, ctx, args):
if not args:
args = ["run"]
elif args[0] not in self.commands and args[0] not in ("--help", "--version"):
args = ["run", *args]
return super().parse_args(ctx, args)
[docs]
class PathArg(click.Path):
"""
``click.Path`` that repairs the PowerShell trailing-backslash accident.
``rcomp '.\\dir\\'`` reaches the program as ``.\\dir"`` because Windows
argument parsing reads the final ``\\"`` as an escaped quote. A double
quote being illegal in Windows file names, a trailing one is always that
accident: strip it before validating.
"""
[docs]
def convert(self, value, param, ctx):
if os.name == "nt" and isinstance(value, str) and value.endswith('"'):
value = value.rstrip('"')
return super().convert(value, param, ctx)
[docs]
def verbosity_options(f):
"""Add ``--verbose``/``--quiet`` to a command."""
f = click.option("--verbose", "-v", is_flag=True, help="Show debug details (full commands, ffmpeg errors).")(f)
f = click.option("--quiet", "-q", is_flag=True, help="Only warnings, errors and the final summary.")(f)
return f
def init_logging(verbose, quiet):
setup_logging(1 if verbose else -1 if quiet else 0)
[docs]
def parse_age(text):
"""
Parse a human age into a :class:`~datetime.timedelta`.
Parameters
----------
text: :class:`str`
``90m``, ``36h``, ``7d``; a bare number means hours.
Examples
--------
>>> parse_age('90m')
datetime.timedelta(seconds=5400)
>>> parse_age('7d')
datetime.timedelta(days=7)
>>> parse_age('24')
datetime.timedelta(days=1)
"""
text = text.strip().lower()
units = {"m": 60, "h": 3600, "d": 86400}
factor = units.get(text[-1]) if text else None
number = text[:-1] if factor else text
try:
return timedelta(seconds=float(number) * (factor or 3600))
except ValueError:
msg = f"Cannot parse age '{text}' (use forms like 90m, 36h or 7d)."
raise click.BadParameter(msg) from None
@click.group(cls=DefaultGroup)
@click.version_option(package_name="remote-compression")
def main():
"""rcomp - batch video compression through ffmpeg, locally or over SSH."""
# ---------------------------------------------------------------------------
# run
@main.command()
@click.option("--preset", "-P", default=None,
help="Named preset: builtin (soft, hard, hard4) or from the config file.")
@click.option("--codec", "-C", default=_DEFAULTS.codec, show_default=True,
help="ffmpeg encoder (free choice: libx264, libsvtav1, hevc_nvenc, ...).")
@click.option("--codec-name", default=None,
help="ffprobe codec name considered compliant (derived from --codec when omitted).")
@click.option("--crf", type=click.IntRange(0, 63), default=None,
help="Constant rate factor (encoder default when omitted).")
@click.option("--ffmpeg-preset", default=None,
help="ffmpeg -preset, e.g. medium or slow (encoder default when omitted).")
@click.option("--height", "-H", type=click.IntRange(min=0), default=_DEFAULTS.height,
show_default=True, help="Maximal height; 0 disables downscaling.")
@click.option("--map/--no-map", "map_streams", default=_DEFAULTS.map_streams, show_default=True,
help="Explicit mapping of video/audio/subtitle streams.")
@click.option("--replace/--no-replace", "-R", "replace", default=_DEFAULTS.replace,
show_default=True, help="Replace originals instead of keeping ori_* backups.")
@click.option("--hostname", "-D", default=_DEFAULTS.hostname, show_default=True,
help="SSH alias of the compression server; 'local' for this machine.")
@click.option("--container", default=_DEFAULTS.container, show_default=True,
help="Output container.")
@click.option("--dry-run", "-n", is_flag=True, help="Show what would be done without doing it.")
@click.option("--retry-failed", is_flag=True, help="Retry the files present in the failure log.")
@click.option("--no-progress", is_flag=True, help="Disable the progress bars.")
@verbosity_options
@click.argument("target", default=".", type=PathArg(exists=True, path_type=Path))
@click.pass_context
def run(ctx, preset, dry_run, retry_failed, no_progress, verbose, quiet, target, **params):
"""Compress TARGET: a video file, or a directory scanned recursively."""
init_logging(verbose, quiet)
if shutil.which("ffprobe") is None:
msg = "ffprobe not found on this machine (needed locally, even for remote compression)."
raise click.ClickException(msg)
explicit = {
name: value
for name, value in params.items()
if ctx.get_parameter_source(name) is ParameterSource.COMMANDLINE
}
try:
settings = resolve_settings(load_config(), preset=preset, cli_overrides=explicit)
_ = settings.effective_codec_name # fail fast when underivable
except (ConfigError, ValueError) as e:
raise click.ClickException(str(e)) from e
try:
summary = run_batch(
target, settings,
dry_run=dry_run, retry_failed=retry_failed, no_progress=no_progress,
)
except (BootstrapError, SSHError, BatchAborted) as e:
raise click.ClickException(str(e)) from e
if summary.failed:
ctx.exit(1)
# ---------------------------------------------------------------------------
# config
@main.command(name="config")
@click.option("--init", "init_", is_flag=True, help="Write a commented configuration template.")
@click.option("--force", is_flag=True, help="Overwrite an existing file (with --init).")
@verbosity_options
def config_cmd(init_, force, verbose, quiet):
"""Show the configuration path and the effective defaults."""
init_logging(verbose, quiet)
path = config_path()
if init_:
try:
written = write_template(path, force=force)
except ConfigError as e:
raise click.ClickException(str(e)) from e
click.echo(f"Template written to {written}")
return
state = "" if path.exists() else " (not created yet - run 'rcomp config --init')"
click.echo(f"Configuration file: {path}{state}")
click.echo(f"Logs: {logs_dir()}")
try:
config_data = load_config()
settings = resolve_settings(config_data)
except ConfigError as e:
raise click.ClickException(str(e)) from e
click.echo("Effective defaults:")
for field in dataclasses.fields(settings):
value = getattr(settings, field.name)
if field.name == "extensions":
value = " ".join(sorted(value))
click.echo(f" {field.name} = {value}")
presets = sorted({**builtin_presets(), **config_data["presets"]})
click.echo(f"Available presets: {', '.join(presets)}")
# ---------------------------------------------------------------------------
# cleanup
@main.command()
@click.option("--dry-run", "-n", is_flag=True, help="List without removing.")
@verbosity_options
@click.argument("target", default=".", type=PathArg(exists=True, path_type=Path))
def cleanup(dry_run, verbose, quiet, target):
"""Remove legacy .keep files under TARGET and prune the failure log."""
init_logging(verbose, quiet)
keeps = find_keep_files(target)
for keep in keeps:
click.echo(f"{'Would remove' if dry_run else 'Removing'} {keep}")
if not dry_run:
keep.unlink()
action = "found" if dry_run else "removed"
click.echo(f"{len(keeps)} legacy .keep file(s) {action}.")
if not dry_run:
pruned = Tracker().prune()
click.echo(f"{pruned} stale failure-log entr{'y' if pruned == 1 else 'ies'} pruned.")
# ---------------------------------------------------------------------------
# failures
@main.command()
@click.option("--clear", "clear_", is_flag=True, help="Remove the listed entries.")
@verbosity_options
@click.argument("target", required=False, type=PathArg(path_type=Path))
def failures(clear_, verbose, quiet, target):
"""List (or clear) the failure log that makes files skipped.
With TARGET (a file or directory), only the matching entries are shown or
cleared. Cleared files are retried on the next run (like --retry-failed).
"""
init_logging(verbose, quiet)
tracker = Tracker()
entries = tracker.entries
if target is not None:
prefix = tracking_key(target)
entries = {
key: entry
for key, entry in entries.items()
if key == prefix or key.startswith(prefix + os.sep)
}
rows = sorted(entries.items(), key=lambda item: item[1].get("date", ""))
for key, entry in rows:
date = (entry.get("date") or "")[:10]
reason = entry.get("reason", "?")
click.echo(f"{date} {reason:<15} {key}")
detail = (entry.get("detail") or "").replace("\n", " ").strip()
if detail:
if not verbose and len(detail) > 60:
detail = detail[:60] + "..."
click.echo(f"{'':28}{detail}")
if clear_:
for key in list(entries): # entries may BE tracker.entries: copy before mutating
tracker.discard_key(key)
if rows:
tracker.save()
click.echo(f"{len(rows)} entr{'y' if len(rows) == 1 else 'ies'} cleared.")
else:
scope = "" if target is None else f" under {target}"
click.echo(f"{len(rows)} entr{'y' if len(rows) == 1 else 'ies'}{scope}.")
# ---------------------------------------------------------------------------
# remote
@main.group()
def remote():
"""Inspect or purge the remote workspace (.rcomp)."""
def _admin_host(hostname):
"""Resolve the target host for the admin commands (no bootstrap, no auto-purge)."""
overrides = {"hostname": hostname} if hostname is not None else None
try:
settings = resolve_settings(load_config(), cli_overrides=overrides)
except ConfigError as e:
raise click.ClickException(str(e)) from e
name = settings.hostname
if name.lower() == "local" or (settings.auto_local and is_localhost(name)):
msg = f"'{name}' is this machine: local mode has no remote workspace."
raise click.ClickException(msg)
return RemoteHost(name, auto_purge_days=None, workdir=settings.remote_workdir,
workdir_sftp=settings.remote_workdir_sftp)
def _human_age(delta):
seconds = int(delta.total_seconds())
days, seconds = divmod(seconds, 86400)
hours, seconds = divmod(seconds, 3600)
minutes = seconds // 60
if days:
return f"{days}d{hours:02d}h"
if hours:
return f"{hours}h{minutes:02d}m"
return f"{minutes}m"
@remote.command()
@click.option("--hostname", "-D", default=None, help="SSH alias (defaults to the configured hostname).")
@verbosity_options
def status(hostname, verbose, quiet):
"""Show the content of the remote workspace."""
init_logging(verbose, quiet)
host = _admin_host(hostname)
try:
host.session.open()
workspace = host.workspace_status()
except (SSHError, BootstrapError) as e:
raise click.ClickException(str(e)) from e
finally:
host.session.close()
if workspace is None:
click.echo(f"No workspace ({host.workdir_sftp}) on {host.hostname}: nothing was ever compressed there.")
return
if not workspace.entries:
click.echo(f"Workspace {host.workdir_sftp} on {host.hostname} is empty.")
return
click.echo(f"Workspace {host.workdir_sftp} on {host.hostname}: {workspace.count} file(s), {human_size(workspace.total_bytes)}.")
for entry in workspace.entries:
click.echo(f" {entry.name:<44} {human_size(entry.size):>10} {_human_age(entry.age())}")
click.echo("Use 'rcomp remote purge' to clean leftovers.")
@remote.command()
@click.option("--hostname", "-D", default=None, help="SSH alias (defaults to the configured hostname).")
@click.option("--older-than", default="24h", show_default=True,
help="Minimal age of the files to remove: 90m, 36h, 7d (bare number = hours).")
@click.option("--all", "purge_all", is_flag=True, help="Purge everything, regardless of age.")
@click.option("--dry-run", "-n", is_flag=True, help="List without removing.")
@verbosity_options
def purge(hostname, older_than, purge_all, dry_run, verbose, quiet):
"""Remove leftovers from the remote workspace."""
init_logging(verbose, quiet)
age = timedelta(0) if purge_all else parse_age(older_than)
host = _admin_host(hostname)
try:
host.session.open()
report = host.purge_workspace(older_than=age, dry_run=dry_run)
except (SSHError, BootstrapError) as e:
raise click.ClickException(str(e)) from e
finally:
host.session.close()
verb = "Would remove" if dry_run else "Removed"
for entry in report.removed:
click.echo(f"{verb} {entry.name} ({human_size(entry.size)}, {_human_age(entry.age())} old)")
for name, error in report.failed:
click.echo(f"Could not remove {name}: {error}", err=True)
click.echo(f"{verb.split()[0] if dry_run else 'Freed'}: {len(report.removed)} file(s), {human_size(report.freed_bytes)}.")
if __name__ == "__main__":
main() # pragma: no cover