367 lines
13 KiB
Python
367 lines
13 KiB
Python
"""Turn user options + probe results into an ffmpeg command line.
|
|
|
|
Pure Python (no Qt) so it can be unit tested.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import shlex
|
|
import subprocess
|
|
from dataclasses import asdict, dataclass, field, fields
|
|
from pathlib import Path
|
|
|
|
from .formats import (
|
|
AUDIO_ENCODERS_BY_NAME,
|
|
CONTAINERS,
|
|
CUSTOM_RESOLUTION,
|
|
VIDEO_ENCODERS_BY_NAME,
|
|
container_accepts_audio,
|
|
container_accepts_video,
|
|
fps_expr,
|
|
subtitle_plan,
|
|
)
|
|
from .probe import MediaInfo, Stream
|
|
from .tools import IS_WINDOWS
|
|
|
|
VIDEO_MODES = {
|
|
"copy": "Copy (keep original, no re-encode)",
|
|
"auto": "Auto (copy if compatible, else re-encode)",
|
|
"encode": "Re-encode",
|
|
}
|
|
AUDIO_MODES = {
|
|
"copy": "Copy (keep original, no re-encode)",
|
|
"auto": "Auto (copy if compatible, else re-encode)",
|
|
"encode": "Re-encode",
|
|
"strip": "Strip audio (no audio in output)",
|
|
}
|
|
EXISTS_POLICIES = {
|
|
"rename": "Add a number (keep both)",
|
|
"overwrite": "Overwrite",
|
|
"skip": "Skip file",
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class Options:
|
|
container: str = "mp4"
|
|
output_dir: str = "" # empty = next to the source file
|
|
suffix: str = ""
|
|
if_exists: str = "rename"
|
|
|
|
video_mode: str = "copy"
|
|
video_encoder: str = "libx264"
|
|
resolution: int = 0 # target height, 0 = original, -1 = custom
|
|
custom_width: int = 0
|
|
custom_height: int = 0
|
|
no_upscale: bool = False
|
|
fps: str = "Original"
|
|
quality: int = 20
|
|
speed: int = 2
|
|
|
|
audio_mode: str = "auto"
|
|
audio_encoder: str = "aac"
|
|
audio_bitrate: int = 0 # kbps, 0 = auto
|
|
audio_channels: int = 0 # 0 = original
|
|
audio_sample_rate: int = 0 # Hz, 0 = original
|
|
|
|
subtitles: bool = True
|
|
|
|
def to_dict(self) -> dict:
|
|
return asdict(self)
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict) -> "Options":
|
|
opts = cls()
|
|
for f in fields(cls):
|
|
if f.name in data:
|
|
try:
|
|
setattr(opts, f.name, type(getattr(opts, f.name))(data[f.name]))
|
|
except (TypeError, ValueError):
|
|
pass
|
|
return opts
|
|
|
|
|
|
@dataclass
|
|
class BuiltCommand:
|
|
args: list[str]
|
|
notes: list[str] = field(default_factory=list) # human readable decisions / warnings
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Video scaling
|
|
# --------------------------------------------------------------------------
|
|
|
|
def scale_filter(video: Stream, opts: Options) -> str | None:
|
|
"""Return an ffmpeg ``scale`` filter for the requested resolution, or None."""
|
|
size = video.display_size
|
|
sw, sh = size if size else (0, 0)
|
|
|
|
if opts.resolution == CUSTOM_RESOLUTION:
|
|
w, h = max(0, opts.custom_width), max(0, opts.custom_height)
|
|
if not w and not h:
|
|
return None
|
|
if w and h:
|
|
factor = min(w / sw, h / sh) if size else 0
|
|
expr = f"scale={w}:{h}:force_original_aspect_ratio=decrease:force_divisible_by=2"
|
|
elif w:
|
|
factor = w / sw if size else 0
|
|
expr = f"scale={w - w % 2}:-2"
|
|
else:
|
|
factor = h / sh if size else 0
|
|
expr = f"scale=-2:{h - h % 2}"
|
|
elif opts.resolution > 0:
|
|
target = opts.resolution
|
|
if size and sh > sw:
|
|
# Portrait video: "1080p" means 1080 pixels wide.
|
|
factor = target / sw
|
|
expr = f"scale={target}:-2"
|
|
else:
|
|
factor = target / sh if size else 0
|
|
expr = f"scale=-2:{target}"
|
|
else:
|
|
return None
|
|
|
|
if size:
|
|
if abs(factor - 1) < 1e-6:
|
|
return None
|
|
if opts.no_upscale and factor > 1:
|
|
return None
|
|
return expr + ":flags=lanczos"
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Command builder
|
|
# --------------------------------------------------------------------------
|
|
|
|
def build_command(
|
|
ffmpeg: str,
|
|
info: MediaInfo,
|
|
opts: Options,
|
|
audio_indices: list[int],
|
|
output: Path,
|
|
) -> BuiltCommand:
|
|
container = CONTAINERS[opts.container]
|
|
notes: list[str] = []
|
|
args = [
|
|
ffmpeg, "-hide_banner", "-nostdin", "-y",
|
|
"-loglevel", "warning",
|
|
"-progress", "pipe:1", "-nostats",
|
|
"-i", str(info.path),
|
|
]
|
|
maps: list[str] = []
|
|
codec_args: list[str] = []
|
|
|
|
# ---- video -----------------------------------------------------------
|
|
video = info.video
|
|
if video is not None:
|
|
maps += ["-map", f"0:{video.index}"]
|
|
filters = []
|
|
scale = scale_filter(video, opts)
|
|
if scale:
|
|
filters.append(scale)
|
|
fps = fps_expr(opts.fps)
|
|
if fps:
|
|
filters.append(f"fps={fps}")
|
|
|
|
if opts.video_mode == "copy":
|
|
encode = False
|
|
if not container_accepts_video(container, video.codec):
|
|
notes.append(f"Warning: {video.codec} video may not be supported in {container.ext} when copying")
|
|
elif opts.video_mode == "auto":
|
|
encode = bool(filters) or not container_accepts_video(container, video.codec)
|
|
else:
|
|
encode = True
|
|
|
|
if encode:
|
|
enc = VIDEO_ENCODERS_BY_NAME[opts.video_encoder]
|
|
if filters:
|
|
codec_args += ["-vf", ",".join(filters)]
|
|
codec_args += enc.args(opts.quality, opts.speed)
|
|
out_codec = enc.codec
|
|
notes.append(f"Video: re-encode {video.codec} -> {enc.label}" + (f" [{', '.join(filters)}]" if filters else ""))
|
|
else:
|
|
codec_args += ["-c:v", "copy"]
|
|
out_codec = video.codec
|
|
notes.append(f"Video: copy {video.codec}")
|
|
if out_codec == "hevc" and opts.container in ("mp4", "mov", "m4v"):
|
|
# Apple devices (and some smart TVs) only play HEVC tagged as hvc1.
|
|
codec_args += ["-tag:v", "hvc1"]
|
|
|
|
# ---- audio -----------------------------------------------------------
|
|
if opts.audio_mode == "strip":
|
|
codec_args.append("-an")
|
|
notes.append("Audio: stripped")
|
|
else:
|
|
selected = [s for s in (info.stream(i) for i in audio_indices) if s is not None and s.kind == "audio"]
|
|
if not selected and info.audio:
|
|
notes.append("Audio: no tracks selected - output will have no audio")
|
|
enc = AUDIO_ENCODERS_BY_NAME[opts.audio_encoder]
|
|
for i, s in enumerate(selected):
|
|
maps += ["-map", f"0:{s.index}"]
|
|
codec_args += _audio_track_args(i, s, opts, container, enc, notes)
|
|
codec_args += [f"-disposition:a:{i}", "default" if i == 0 else "0"]
|
|
|
|
# ---- subtitles -------------------------------------------------------
|
|
if opts.subtitles:
|
|
j = 0
|
|
for s in info.subtitles:
|
|
plan = subtitle_plan(container, s.codec)
|
|
if plan is None:
|
|
notes.append(f"Subtitles: dropped track #{s.index} ({s.codec}) - not supported in {container.ext}")
|
|
continue
|
|
maps += ["-map", f"0:{s.index}"]
|
|
codec_args += [f"-c:s:{j}", plan]
|
|
j += 1
|
|
if opts.container == "mkv" and any(s.kind == "attachment" for s in info.streams):
|
|
# Fonts used by ASS subtitles.
|
|
maps += ["-map", "0:t?"]
|
|
codec_args += ["-c:t", "copy"]
|
|
|
|
args += maps + codec_args
|
|
args += ["-max_muxing_queue_size", "4096"]
|
|
args += list(container.extra_args)
|
|
args.append(str(output))
|
|
return BuiltCommand(args=args, notes=notes)
|
|
|
|
|
|
def _audio_track_args(i, s: Stream, opts: Options, container, enc, notes: list[str]) -> list[str]:
|
|
want_ch, want_sr = opts.audio_channels, opts.audio_sample_rate
|
|
label = f"Audio #{s.index}" + (f" [{s.language}]" if s.language else "")
|
|
|
|
if opts.audio_mode == "copy":
|
|
if not container_accepts_audio(container, s.codec):
|
|
notes.append(f"Warning: {label}: {s.codec} may not be supported in {container.ext} when copying")
|
|
notes.append(f"{label}: copy {s.codec}")
|
|
return [f"-c:a:{i}", "copy"]
|
|
|
|
if opts.audio_mode == "auto":
|
|
changes = (want_ch and want_ch != s.channels) or (want_sr and want_sr != s.sample_rate)
|
|
if container_accepts_audio(container, s.codec) and not changes:
|
|
notes.append(f"{label}: copy {s.codec}")
|
|
return [f"-c:a:{i}", "copy"]
|
|
|
|
out = [f"-c:a:{i}", enc.name]
|
|
channels = want_ch or s.channels or 2
|
|
if channels > enc.max_channels:
|
|
notes.append(f"{label}: {enc.label} supports at most {enc.max_channels} channels - downmixing")
|
|
channels = enc.max_channels
|
|
out += [f"-ac:a:{i}", str(channels)]
|
|
elif want_ch:
|
|
out += [f"-ac:a:{i}", str(channels)]
|
|
|
|
if want_sr:
|
|
rate = enc.nearest_sample_rate(want_sr)
|
|
if rate != want_sr:
|
|
notes.append(f"{label}: {enc.label} does not support {want_sr} Hz - using {rate} Hz")
|
|
out += [f"-ar:a:{i}", str(rate)]
|
|
|
|
if not enc.lossless:
|
|
kbps = min(opts.audio_bitrate or enc.auto_bitrate(channels), enc.max_kbps)
|
|
out += [f"-b:a:{i}", f"{kbps}k"]
|
|
if enc.name == "libopus" and channels > 2:
|
|
out += [f"-mapping_family:a:{i}", "1"]
|
|
notes.append(f"{label}: re-encode {s.codec} -> {enc.label}")
|
|
return out
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Output paths
|
|
# --------------------------------------------------------------------------
|
|
|
|
def _norm(p: Path) -> str:
|
|
return os.path.normcase(os.path.abspath(p))
|
|
|
|
|
|
def plan_output_path(src: Path, opts: Options, taken: set[str], exists=os.path.exists) -> Path | None:
|
|
"""Pick the output file for ``src``. Returns None if the file should be skipped.
|
|
|
|
``taken`` holds normalised paths already produced in this batch; they are
|
|
never overwritten even with the "overwrite" policy.
|
|
"""
|
|
ext = CONTAINERS[opts.container].ext
|
|
out_dir = Path(opts.output_dir) if opts.output_dir else src.parent
|
|
stem = src.stem + opts.suffix
|
|
cand = out_dir / f"{stem}{ext}"
|
|
if _norm(cand) == _norm(src):
|
|
# Never overwrite the input (e.g. mp4 -> mp4 in the same folder).
|
|
stem += "_converted"
|
|
cand = out_dir / f"{stem}{ext}"
|
|
|
|
in_batch = _norm(cand) in taken
|
|
if not in_batch and not exists(cand):
|
|
return cand
|
|
if not in_batch and opts.if_exists == "skip":
|
|
return None
|
|
if not in_batch and opts.if_exists == "overwrite" and _norm(cand) != _norm(src):
|
|
return cand
|
|
n = 1
|
|
while True:
|
|
p = out_dir / f"{stem} ({n}){ext}"
|
|
if _norm(p) not in taken and not exists(p) and _norm(p) != _norm(src):
|
|
return p
|
|
n += 1
|
|
|
|
|
|
def temp_output_path(final: Path) -> Path:
|
|
"""ffmpeg writes here first; renamed to ``final`` only on success."""
|
|
return final.with_name(f"{final.stem}.part{final.suffix}")
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Audio track selection helpers
|
|
# --------------------------------------------------------------------------
|
|
|
|
def default_audio_selection(info: MediaInfo) -> list[int]:
|
|
return [s.index for s in info.audio]
|
|
|
|
|
|
def match_audio_selection(ref_all: list[Stream], ref_selected: list[int], target_all: list[Stream]) -> list[int]:
|
|
"""Map a track selection made on one file onto another file's audio tracks.
|
|
|
|
Identical track layouts map by position; otherwise each selected track is
|
|
matched by language, then codec / channels / title / position. Falls back to
|
|
the target's default (or first) track so audio is never silently lost.
|
|
"""
|
|
if not target_all:
|
|
return []
|
|
if not ref_selected:
|
|
return []
|
|
sig = lambda s: (s.language, s.codec, s.channels) # noqa: E731
|
|
ref_pos = {s.index: pos for pos, s in enumerate(ref_all)}
|
|
by_index = {s.index: s for s in ref_all}
|
|
|
|
if [sig(s) for s in ref_all] == [sig(s) for s in target_all]:
|
|
return [target_all[ref_pos[i]].index for i in ref_selected if i in ref_pos]
|
|
|
|
result: list[int] = []
|
|
for i in ref_selected:
|
|
r = by_index.get(i)
|
|
if r is None:
|
|
continue
|
|
best, best_score = None, 0
|
|
for pos, t in enumerate(target_all):
|
|
if t.index in result:
|
|
continue
|
|
if r.language and t.language and r.language != t.language:
|
|
continue
|
|
score = 1
|
|
score += 8 if r.language and r.language == t.language else 0
|
|
score += 2 if r.codec == t.codec else 0
|
|
score += 2 if r.channels == t.channels else 0
|
|
score += 2 if r.title and r.title == t.title else 0
|
|
score += 1 if pos == ref_pos[i] else 0
|
|
if score > best_score:
|
|
best, best_score = t, score
|
|
if best is not None:
|
|
result.append(best.index)
|
|
|
|
if not result:
|
|
default = next((t for t in target_all if t.default), target_all[0])
|
|
result = [default.index]
|
|
return result
|
|
|
|
|
|
def format_command(args: list[str]) -> str:
|
|
return subprocess.list2cmdline(args) if IS_WINDOWS else shlex.join(args)
|