"""Per-file pipeline: probe, decide, compress, verify, swap."""
import logging
import subprocess
from dataclasses import dataclass
from pathlib import Path
from remote_compression.command import build_ffmpeg_command, evaluate
from remote_compression.probe import probe
from remote_compression.progress import human_size
logger = logging.getLogger("rcomp")
#: Host failure reasons that are deterministic (recorded in the tracker).
#: Transient ones (upload/download/exec-timeout) are retried on the next run.
_TRACKED_HOST_REASONS = {"ffmpeg-failed", ""}
[docs]
@dataclass(slots=True)
class FileResult:
"""
Outcome of :func:`compress` for one file.
Attributes
----------
path: :class:`~pathlib.Path`
The file concerned (the produced file when status is ``done``).
status: :class:`str`
``done``, ``conform``, ``failed``, ``not_profitable`` or ``dry_run``.
message: :class:`str`
Failure reason or extra context.
old_size: :class:`int`
Source size in bytes (when relevant).
new_size: :class:`int`
Output size in bytes (when produced).
"""
path: Path
status: str
message: str = ""
old_size: int = 0
new_size: int = 0
[docs]
def compress(source, settings, tracker=None, host=None, dry_run=False, progress_factory=None):
"""
Compress one file according to `settings`.
Parameters
----------
source: :class:`~pathlib.Path` or :class:`str`
Video to compress.
settings: :class:`~remote_compression.settings.Settings`
Resolved settings.
tracker: :class:`~remote_compression.tracking.Tracker`, optional
Central log for deterministic failures (never written in dry-run).
host: :class:`~remote_compression.host.LocalHost` or :class:`~remote_compression.host.RemoteHost`
Execution host (already bootstrapped). Unused in dry-run.
dry_run: :class:`bool`
Show what would be done without touching anything.
progress_factory: callable, optional
``factory(source, info) -> chunk callback`` built once the file is
probed (the probe provides the duration for the bar).
Returns
-------
:class:`FileResult`
Business failures are returned, never raised; connection losses and
Ctrl-C propagate to the batch loop.
Notes
-----
Both replace modes share the same success checks (ffmpeg exit status,
output present and non-empty, then a profitability ratio): a corrupt or
bigger output can no longer overwrite a source.
"""
source = Path(source)
if dry_run:
tracker = None
info = probe(source)
if info is None:
logger.warning("%s: cannot probe (corrupt file or unsupported format).", source.name)
if tracker:
tracker.record(source, "probe_failed")
return FileResult(source, "failed", "probe_failed")
plan = evaluate(info, settings)
for warning in plan.warnings:
logger.warning("%s: %s", source.name, warning)
if not plan.todo:
logger.debug("%s: already compliant.", source.name)
return FileResult(source, "conform")
ffplan = build_ffmpeg_command(info, settings)
final = source.with_suffix("." + settings.container)
comp = source.with_name(f"comp_{source.stem}.{settings.container}")
ori = source.with_name(f"ori_{source.name}")
if dry_run:
logger.info("[dry-run] %s -> %s", source.name, final.name)
# show the actual command (comp_* target): safe to copy-paste, unlike
# a command whose output would overwrite its input
logger.info("[dry-run] %s", subprocess.list2cmdline(ffplan.argv(source, comp)))
return FileResult(source, "dry_run")
# Guards, before any ffmpeg work.
if final != source and final.exists():
logger.warning("%s: target %s already exists, skipping.", source.name, final.name)
if tracker:
tracker.record(source, "target_exists", str(final))
return FileResult(source, "failed", "target_exists")
if not settings.replace and ori.exists():
logger.warning("%s: backup %s already exists, skipping.", source.name, ori.name)
if tracker:
tracker.record(source, "ori_exists", str(ori))
return FileResult(source, "failed", "ori_exists")
old_size = source.stat().st_size
on_progress = progress_factory(source, info) if progress_factory is not None else None
try:
result = host.compress_file(source, comp, ffplan, on_progress=on_progress)
except BaseException:
# aborted mid-download (connection loss, Ctrl-C): drop the local partial
comp.unlink(missing_ok=True)
raise
if not (result.ok and comp.exists() and comp.stat().st_size > 0):
comp.unlink(missing_ok=True)
reason = result.reason or "ffmpeg_error"
full_detail = (result.stderr_tail or "").strip()
detail = full_detail[-500:]
if len(full_detail) > len(detail):
# the tracking entry only keeps the tail; the run log gets everything
logger.debug("%s: full ffmpeg output tail:\n%s", source.name, full_detail)
logger.warning("%s: compression failed (%s). %s", source.name, reason, detail)
if tracker and result.reason in _TRACKED_HOST_REASONS:
tracker.record(source, "ffmpeg_error", detail)
return FileResult(source, "failed", reason, old_size)
new_size = comp.stat().st_size
ratio = new_size / old_size
if not (0.01 < ratio < 1.0):
comp.unlink()
logger.info(
"%s: not profitable (%.0f%% of the original size), keeping the original.",
source.name, 100 * ratio,
)
if tracker:
tracker.record(source, "not_profitable", f"ratio {ratio:.2f}")
return FileResult(source, "not_profitable", f"ratio {ratio:.2f}", old_size, new_size)
if settings.replace:
comp.replace(final)
if final != source:
source.unlink()
else:
source.rename(ori) # before comp.rename: final may BE source (.mkv sources)
comp.rename(final)
if tracker:
tracker.clear(source)
logger.info(
"%s: %s -> %s (%.0f%% of original).",
source.name, human_size(old_size), human_size(new_size), 100 * ratio,
)
return FileResult(final, "done", "", old_size, new_size)