"""Pure logic: decide what to do with a file and build the ffmpeg command."""
from dataclasses import dataclass
from remote_compression.settings import (
INTERLACED_FIELD_ORDERS,
MKV_COPY_SUBS,
TEXT_CONVERT_SUBS,
)
#: Sentinel tokens substituted at execution time.
SOURCE = "{source}"
TARGET = "{target}"
#: Characters banned from remote command lines (cmd.exe expansion, redirections,
#: separators, sh substitutions). Remote paths are hex names generated by us, and
#: option tokens are fixed ASCII, so hitting this guard means a programming error.
FORBIDDEN_CHARS = set('%!^"<>|&;$`\n\r')
#: Tokens containing these characters get double-quoted in the remote command to
#: prevent sh pathname expansion (harmless for cmd.exe).
_GLOB_CHARS = set("?*[")
[docs]
@dataclass(frozen=True, slots=True)
class Plan:
"""
What has to be done for a given file.
Attributes
----------
todo: :class:`bool`
Whether any work is needed. A compliant file is never remuxed.
transcode: :class:`bool`
Whether the video codec must change.
resize: :class:`bool`
Whether the video must be downscaled.
warnings: :class:`tuple` of :class:`str`
Human-readable warnings (e.g. a subtitle stream that will be dropped).
"""
todo: bool
transcode: bool
resize: bool
warnings: tuple = ()
[docs]
@dataclass(frozen=True, slots=True)
class FfmpegPlan:
"""
A ready-to-run ffmpeg command with :data:`SOURCE`/:data:`TARGET` placeholders.
Attributes
----------
tokens: :class:`tuple` of :class:`str`
Full argv, with :data:`SOURCE` and :data:`TARGET` each appearing exactly once.
"""
tokens: tuple
def __post_init__(self):
if self.tokens.count(SOURCE) != 1 or self.tokens.count(TARGET) != 1:
msg = "FfmpegPlan tokens must contain SOURCE and TARGET exactly once each."
raise ValueError(msg)
[docs]
def argv(self, source, target):
"""
Substitute placeholders and return the local execution form.
Parameters
----------
source: :class:`~pathlib.Path` or :class:`str`
Input file.
target: :class:`~pathlib.Path` or :class:`str`
Output file.
Returns
-------
:class:`list` of :class:`str`
Argument list for :func:`subprocess.run` (never a shell).
"""
subst = {SOURCE: str(source), TARGET: str(target)}
return [subst.get(t, t) for t in self.tokens]
[docs]
def command(self, r_source, r_target, binary=None):
"""
Substitute placeholders and return the remote execution form.
Parameters
----------
r_source: :class:`str`
Remote input path, relative to the SFTP home (e.g. ``.rcomp/<hex>.mp4``).
r_target: :class:`str`
Remote output path, same convention.
binary: :class:`str`, optional
Explicit ffmpeg path replacing the leading ``ffmpeg`` token
(``remote_ffmpeg`` setting). Subject to the same safety guards.
Returns
-------
:class:`str`
A single command line safe for both POSIX shells and cmd.exe.
Raises
------
ValueError
If any token or path is non-ASCII or contains a forbidden character.
The original file name never travels through a remote shell (only
through SFTP), so this cannot be triggered by user data.
"""
tokens = self.tokens if binary is None else (binary, *self.tokens[1:])
rendered = []
for token in tokens:
if token == SOURCE:
token = r_source
elif token == TARGET:
token = r_target
if not token.isascii() or FORBIDDEN_CHARS & set(token):
msg = f"Unsafe token for a remote command: {token!r}"
raise ValueError(msg)
if token in (r_source, r_target) or _GLOB_CHARS & set(token) or " " in token:
token = f'"{token}"'
rendered.append(token)
command = " ".join(rendered)
if command.startswith('"'):
# cmd.exe /c strips quotes differently when the line starts with one.
msg = f"Remote command may not start with a quote: {command!r}"
raise ValueError(msg)
return command
[docs]
def evaluate(info, settings):
"""
Decide what has to be done for a file.
Parameters
----------
info: :class:`~remote_compression.probe.MediaInfo`
Probe result for the file.
settings: :class:`~remote_compression.settings.Settings`
Target settings.
Returns
-------
:class:`Plan`
Work description. ``todo`` is True iff a transcode or a resize is
needed; a compliant file is never remuxed to the target container.
"""
transcode = info.codec_name != settings.effective_codec_name
# `height` caps the *definition* (short side): a 1080x1920 portrait is as
# much "1080p" as a 1920x1080 landscape.
resize = settings.height is not None and min(info.width, info.height) > settings.height
warnings = ()
if (transcode or resize) and settings.map_streams and settings.container == "mkv":
_, warnings = subtitle_args(info.subtitles)
return Plan(todo=transcode or resize, transcode=transcode, resize=resize, warnings=warnings)
[docs]
def subtitle_args(subtitles):
"""
Build the subtitle handling arguments for an mkv target.
Parameters
----------
subtitles: :class:`tuple` of :class:`~remote_compression.probe.SubtitleStream`
Subtitle streams of the source.
Returns
-------
:class:`list` of :class:`str`
ffmpeg tokens: per stream, ``copy`` when mkv accepts the codec as-is,
``srt`` conversion for pure-text codecs (``mov_text``), and an unmap for
anything else. Collapsed to a single ``-c:s`` when uniform.
:class:`tuple` of :class:`str`
One warning per dropped stream.
"""
kept, dropped = [], []
for stream in subtitles:
if stream.codec_name in MKV_COPY_SUBS:
kept.append((stream, "copy"))
elif stream.codec_name in TEXT_CONVERT_SUBS:
kept.append((stream, "srt"))
else:
dropped.append(stream)
args = [t for stream in dropped for t in ("-map", f"-0:s:{stream.index}")]
actions = {action for _, action in kept}
if len(actions) == 1:
args += ["-c:s", actions.pop()]
else:
# Output subtitle indices are renumbered once drops are applied.
args += [t for j, (_, action) in enumerate(kept) for t in (f"-c:s:{j}", action)]
warnings = tuple(
f"subtitle stream {s.index} ({s.codec_name or 'unknown codec'}) "
"cannot be stored in mkv and will be dropped"
for s in dropped
)
return args, warnings
[docs]
def build_ffmpeg_command(info, settings):
"""
Build the full ffmpeg command for a file.
Parameters
----------
info: :class:`~remote_compression.probe.MediaInfo`
Probe result for the file.
settings: :class:`~remote_compression.settings.Settings`
Target settings.
Returns
-------
:class:`FfmpegPlan`
Command with placeholders, usable locally (:meth:`FfmpegPlan.argv`) or
remotely (:meth:`FfmpegPlan.command`).
Notes
-----
``-nostats -progress pipe:1`` replaces the stderr chatter with a
machine-readable ``key=value`` stream on stdout: it feeds the progress bar
and doubles as the liveness heartbeat for remote execution.
"""
tokens = [
"ffmpeg",
"-y",
"-nostdin",
"-hide_banner",
"-loglevel", "error",
"-nostats",
"-progress", "pipe:1",
"-i", SOURCE,
]
subtitle_tokens = []
if settings.map_streams:
# 0:V excludes cover art; '?' tolerates missing audio/subtitles.
tokens += ["-map", "0:V", "-map", "0:a?", "-map", "0:s?"]
if settings.container == "mkv":
subtitle_tokens, _ = subtitle_args(info.subtitles)
plan = evaluate(info, settings)
filters = []
if info.width % 2 or info.height % 2:
# x265 refuses odd dimensions in 4:2:0 (ancient sources: msmpeg4...);
# drop at most one row/column rather than resampling.
filters.append("crop=trunc(iw/2)*2:trunc(ih/2)*2")
if info.field_order in INTERLACED_FIELD_ORDERS:
# deinterlace before any downscale (DVD, AVCHD 1080i...)
filters.append("bwdif")
if plan.resize:
# ffmpeg auto-rotates before filters, so portrait frames need the cap
# on the width slot to bound their short side.
if info.width >= info.height:
filters.append(f"scale=-2:{settings.height}")
else:
filters.append(f"scale={settings.height}:-2")
if filters:
tokens += ["-vf", ",".join(filters)]
# A resize implies re-encoding too: filters are incompatible with -c:v copy.
tokens += ["-c:v", settings.codec]
if settings.crf is not None:
tokens += ["-crf", str(settings.crf)]
if settings.ffmpeg_preset is not None:
tokens += ["-preset", settings.ffmpeg_preset]
tokens += ["-c:a", "copy"]
tokens += subtitle_tokens
tokens += ["-max_muxing_queue_size", "9999", TARGET]
return FfmpegPlan(tokens=tuple(tokens))