"""Directory scan and the batch loop (error isolation, circuit breaker, summary)."""
import contextlib
import logging
import os
import time
from dataclasses import dataclass
from pathlib import Path
from remote_compression.compression import FileResult, compress
from remote_compression.host import BootstrapError, open_host
from remote_compression.progress import BatchBar, FileBar, human_size
from remote_compression.ssh import SSHError
from remote_compression.tracking import Tracker
logger = logging.getLogger("rcomp")
#: Consecutive failures beyond which the batch aborts (something is systemic).
MAX_CONSECUTIVE_FAILURES = 5
#: Legacy per-directory tracking files from rcomp < 0.2 (dill format, unused).
LEGACY_KEEP = ".keep"
[docs]
class BatchAborted(RuntimeError):
"""Too many consecutive failures: the batch stops instead of burning CPU."""
[docs]
@dataclass(slots=True)
class BatchSummary:
"""
Counters of a batch run.
Attributes
----------
done, conform, tracked, failed, not_profitable, dry_run: :class:`int`
Files per outcome (``tracked`` = skipped through the failure log).
bytes_before, bytes_after: :class:`int`
Cumulated sizes of successfully compressed files.
keep_files_seen: :class:`int`
Legacy ``.keep`` files crossed during the scan.
"""
done: int = 0
conform: int = 0
tracked: int = 0
failed: int = 0
not_profitable: int = 0
dry_run: int = 0
bytes_before: int = 0
bytes_after: int = 0
keep_files_seen: int = 0
[docs]
def bump(self, result):
"""Account for one :class:`~remote_compression.compression.FileResult`."""
setattr(self, result.status, getattr(self, result.status) + 1)
if result.status == "done":
self.bytes_before += result.old_size
self.bytes_after += result.new_size
[docs]
def scan(target, extensions):
"""
List the videos to consider.
Parameters
----------
target: :class:`~pathlib.Path` or :class:`str`
A video file, or a directory walked recursively (deterministic order).
extensions: :class:`frozenset` of :class:`str`
Extensions to keep (compared lower-case, so ``.MP4`` files count).
Returns
-------
:class:`list` of :class:`~pathlib.Path`
Video files; ``ori_*`` backups and ``comp_*`` leftovers are skipped.
:class:`int`
Number of legacy ``.keep`` files crossed (see ``rcomp cleanup``).
"""
target = Path(target)
if target.is_file():
return [target], 0
files, keeps = [], 0
for dirpath, dirnames, filenames in os.walk(target):
dirnames.sort()
for name in sorted(filenames):
if name == LEGACY_KEEP:
keeps += 1
continue
if name.startswith(("ori_", "comp_")):
continue
path = Path(dirpath) / name
if path.suffix.lower() in extensions:
files.append(path)
return files, keeps
[docs]
def find_keep_files(target):
"""
Returns
-------
:class:`list` of :class:`~pathlib.Path`
All legacy ``.keep`` files under `target`.
"""
target = Path(target)
return sorted(target.rglob(LEGACY_KEEP)) if target.is_dir() else []
[docs]
def run_batch(target, settings, dry_run=False, retry_failed=False, no_progress=False,
tracker=None, host=None):
"""
Compress every relevant video under `target`.
Parameters
----------
target: :class:`~pathlib.Path` or :class:`str`
Video file or directory.
settings: :class:`~remote_compression.settings.Settings`
Resolved settings.
dry_run: :class:`bool`
Show the work without doing it: no ffmpeg, no SSH connection, no write.
retry_failed: :class:`bool`
Ignore the failure log and retry those files.
no_progress: :class:`bool`
Disable the tqdm bars.
tracker: :class:`~remote_compression.tracking.Tracker`, optional
Injected failure log (defaults to the user-wide one).
host: optional
Injected execution host (defaults to :func:`~remote_compression.host.open_host`
on the settings). Entered as a context manager either way.
Returns
-------
:class:`BatchSummary`
Also logged at the end, including on interruption.
Raises
------
BatchAborted
After :data:`MAX_CONSECUTIVE_FAILURES` consecutive failures.
"""
from remote_compression.progress import bars_enabled
files, keeps = scan(target, settings.extensions)
summary = BatchSummary(keep_files_seen=keeps)
if not files:
logger.info("No video file found under %s.", target)
return summary
tracker = tracker if tracker is not None else Tracker()
enabled = bars_enabled(no_progress) and not dry_run
if dry_run:
host_context = contextlib.nullcontext()
else:
host_context = host if host is not None else open_host(settings)
consecutive_failures = 0
interrupted = False
with host_context as live_host:
# bar and summary only exist once the host is up: a bootstrap failure
# must yield its own clear message, not a misleading empty summary
start = time.monotonic()
batch_bar = BatchBar(total=len(files), enabled=enabled)
try:
for file in files:
if not retry_failed:
reason = tracker.should_skip(file)
if reason is not None:
logger.debug("%s: skipped, in the failure log (%s).", file.name, reason)
summary.tracked += 1
batch_bar.advance(summary)
continue
bar_holder = []
def progress_factory(source, info, _holder=bar_holder):
bar = FileBar(source.name, duration=info.duration, enabled=enabled)
_holder.append(bar)
return bar
try:
result = compress(
file, settings,
tracker=tracker, host=live_host,
dry_run=dry_run, progress_factory=progress_factory,
)
except (KeyboardInterrupt, BatchAborted):
raise
except (SSHError, BootstrapError):
raise # connection lost for good: the batch cannot continue
except Exception as e: # noqa: BLE001 - isolation: one bad file must not kill the batch
logger.error("%s: unexpected error: %r", file.name, e)
if not dry_run:
tracker.record(file, "exception", repr(e))
result = FileResult(file, "failed", "exception")
finally:
for bar in bar_holder:
bar.close()
summary.bump(result)
if result.status == "failed":
consecutive_failures += 1
if consecutive_failures > MAX_CONSECUTIVE_FAILURES:
msg = (
f"{consecutive_failures} consecutive failures - something looks "
"systemic (full disk? broken encoder settings?). Aborting."
)
raise BatchAborted(msg)
else:
consecutive_failures = 0
batch_bar.advance(summary)
except KeyboardInterrupt:
interrupted = True
raise
finally:
batch_bar.close()
_log_summary(summary, time.monotonic() - start, interrupted)
return summary
def _log_summary(summary, elapsed, interrupted):
parts = []
for label in ("done", "conform", "tracked", "not_profitable", "failed", "dry_run"):
value = getattr(summary, label)
if value:
parts.append(f"{value} {label}")
headline = ", ".join(parts) if parts else "nothing to do"
prefix = "Interrupted - partial summary: " if interrupted else "Summary: "
logger.info("%s%s (%.0f s).", prefix, headline, elapsed)
if summary.bytes_before:
saved = summary.bytes_before - summary.bytes_after
logger.info(
"Compressed %s down to %s (saved %s).",
human_size(summary.bytes_before), human_size(summary.bytes_after), human_size(saved),
)
if summary.keep_files_seen:
logger.info(
"%d legacy .keep file(s) found - run 'rcomp cleanup' to remove them.",
summary.keep_files_seen,
)