413 lines
13 KiB
Python
413 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from datetime import timezone
|
|
from pathlib import Path
|
|
|
|
from tools.avisynth_runner import (
|
|
AvisynthRunnerSet,
|
|
runner_set_from_env_or_bundled,
|
|
validate_runner_path,
|
|
)
|
|
from tools.console import prompt_input, prompt_yes_no
|
|
from tools.timezones import local_timezone, parse_timezone_offset, timezone_to_string
|
|
from tools.video_encode_output import (
|
|
ENCODER_PRESETS,
|
|
AudioEncodeOptions,
|
|
VideoCodecOptions,
|
|
default_audio_bitrate,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class OutputNamingOptions:
|
|
use_timestamps: bool
|
|
timezone_value: timezone
|
|
|
|
|
|
def ask_ffms2_plugin_path() -> Path | None:
|
|
env_path = os.environ.get("MBT_FFMS2_PLUGIN", "").strip()
|
|
if env_path:
|
|
return _validate_plugin_path(env_path)
|
|
|
|
if not sys.stdin.isatty() or not _env_truthy("MBT_ASK_FFMS2_PLUGIN"):
|
|
return None
|
|
|
|
answer = prompt_input(
|
|
"\nFFMS2 plugin path for Avisynth LoadPlugin "
|
|
"(blank = use Avisynth autoload): "
|
|
).strip()
|
|
if not answer:
|
|
return None
|
|
return _validate_plugin_path(answer)
|
|
|
|
|
|
def _env_truthy(name: str) -> bool:
|
|
return os.environ.get(name, "").strip().lower() in {"1", "true", "yes", "y", "on"}
|
|
|
|
|
|
def _validate_plugin_path(raw_path: str) -> Path:
|
|
plugin_path = Path(raw_path.strip().strip('"')).expanduser()
|
|
if not plugin_path.is_file():
|
|
raise RuntimeError(f"FFMS2 plugin file not found: {plugin_path}")
|
|
return plugin_path.resolve()
|
|
|
|
|
|
def ask_avisynth_runners() -> AvisynthRunnerSet | None:
|
|
runners = runner_set_from_env_or_bundled()
|
|
if runners is not None:
|
|
return runners
|
|
|
|
if not sys.stdin.isatty():
|
|
return None
|
|
|
|
print(
|
|
"\nAviSynth runner is needed for validation and encoding.\n"
|
|
"Use media-batch-tools' mbt_avs_runner executable. The script first "
|
|
"looks for the bundled runner automatically. On Windows, the default is "
|
|
"64-bit mbt_avs_runner.exe; 32-bit scripts can use "
|
|
"mbt_avs_runner32.exe via a trailing #32bit marker. If you leave this "
|
|
"blank, the workspace and editable .avs files are created, but "
|
|
"validation and encoding stop."
|
|
)
|
|
answer = prompt_input(
|
|
"AviSynth runner executable path (blank = create workspace only): "
|
|
).strip()
|
|
if not answer:
|
|
return None
|
|
runner = validate_runner_path(answer)
|
|
return AvisynthRunnerSet(default=runner, runner64=runner)
|
|
|
|
|
|
def require_avisynth_runners_for_encode() -> AvisynthRunnerSet:
|
|
runners = runner_set_from_env_or_bundled()
|
|
if runners is not None:
|
|
return runners
|
|
|
|
if not sys.stdin.isatty():
|
|
raise RuntimeError(
|
|
"Build tools/avisynth_runner/mbt_avs_runner on Linux, provide "
|
|
"tools/avisynth_runner/mbt_avs_runner.exe on Windows, or set "
|
|
"MBT_AVS_RUNNER before enabling video encode. For 32-bit Windows "
|
|
"AviSynth, set MBT_AVS_RUNNER32."
|
|
)
|
|
|
|
print(
|
|
"\nEncoding needs media-batch-tools' AviSynth runner executable "
|
|
"(mbt_avs_runner or mbt_avs_runner.exe)."
|
|
)
|
|
answer = prompt_input("AviSynth runner executable path: ").strip()
|
|
if not answer:
|
|
raise RuntimeError("Avisynth runner path is required for encoding.")
|
|
runner = validate_runner_path(answer)
|
|
return AvisynthRunnerSet(default=runner, runner64=runner)
|
|
|
|
|
|
def ask_encode_video_only() -> bool:
|
|
env_value = os.environ.get("MBT_ENCODE_VIDEO")
|
|
if env_value is not None:
|
|
return env_value.strip().lower() in {"1", "true", "yes", "y", "on"}
|
|
if not sys.stdin.isatty():
|
|
return False
|
|
return prompt_yes_no("Encode validated outputs now?", default=False)
|
|
|
|
|
|
def ask_output_naming_options(
|
|
*,
|
|
default_timezone: timezone | None = None,
|
|
) -> OutputNamingOptions:
|
|
env_value = os.environ.get("MBT_VIDEO_TIMESTAMP_NAMES")
|
|
if env_value is not None:
|
|
use_timestamps = env_value.strip().lower() in {"1", "true", "yes", "y", "on"}
|
|
elif sys.stdin.isatty():
|
|
use_timestamps = prompt_yes_no(
|
|
"Name encoded files from their edited beginning timestamp?",
|
|
default=True,
|
|
)
|
|
else:
|
|
use_timestamps = False
|
|
|
|
default_timezone = default_timezone or local_timezone()
|
|
env_timezone = os.environ.get("MBT_VIDEO_TIMEZONE", "").strip()
|
|
if not use_timestamps:
|
|
timezone_value = default_timezone
|
|
elif env_timezone:
|
|
try:
|
|
timezone_value = parse_timezone_offset(env_timezone)
|
|
except ValueError as exc:
|
|
raise RuntimeError(f"Invalid MBT_VIDEO_TIMEZONE: {exc}") from exc
|
|
elif sys.stdin.isatty():
|
|
while True:
|
|
answer = prompt_input(
|
|
"Output filename timezone "
|
|
f"[{timezone_to_string(default_timezone)}]: "
|
|
).strip()
|
|
if not answer:
|
|
timezone_value = default_timezone
|
|
break
|
|
try:
|
|
timezone_value = parse_timezone_offset(answer)
|
|
break
|
|
except ValueError as exc:
|
|
print(exc)
|
|
else:
|
|
timezone_value = default_timezone
|
|
|
|
return OutputNamingOptions(
|
|
use_timestamps=use_timestamps,
|
|
timezone_value=timezone_value,
|
|
)
|
|
|
|
|
|
def ask_video_codec_options() -> VideoCodecOptions:
|
|
codec = _codec_from_env()
|
|
extension = _container_extension_from_env()
|
|
crf = _int_from_env("MBT_VIDEO_CRF")
|
|
preset = os.environ.get("MBT_VIDEO_PRESET", "").strip()
|
|
profile = _optional_env("MBT_VIDEO_PROFILE")
|
|
level = _optional_env("MBT_VIDEO_LEVEL")
|
|
threads = _threads_from_env()
|
|
|
|
if sys.stdin.isatty():
|
|
if extension is None:
|
|
extension = _prompt_container_extension()
|
|
if codec is None:
|
|
codec = _prompt_codec()
|
|
if crf is None:
|
|
default_crf = 16 if codec == "libx264" else 21
|
|
crf = _prompt_int(f"CRF [{default_crf}]: ", default=default_crf)
|
|
if not preset:
|
|
preset = _prompt_preset(default="slow")
|
|
if "MBT_VIDEO_THREADS" not in os.environ:
|
|
threads = _prompt_threads(default=None)
|
|
else:
|
|
extension = extension or ".mp4"
|
|
codec = codec or "libx265"
|
|
crf = crf if crf is not None else (16 if codec == "libx264" else 21)
|
|
preset = preset or "slow"
|
|
|
|
return VideoCodecOptions(
|
|
codec=codec,
|
|
crf=crf,
|
|
preset=preset,
|
|
extension=extension,
|
|
profile=profile,
|
|
level=level,
|
|
threads=threads,
|
|
)
|
|
|
|
|
|
def _prompt_preset(*, default: str) -> str:
|
|
options = "/".join(ENCODER_PRESETS)
|
|
while True:
|
|
value = prompt_input(f"Encoder preset [{default}] ({options}): ").strip().lower()
|
|
if not value:
|
|
return default
|
|
if value in ENCODER_PRESETS:
|
|
return value
|
|
print(f"Preset must be one of: {options}.")
|
|
|
|
|
|
def _threads_from_env() -> int | None:
|
|
raw = os.environ.get("MBT_VIDEO_THREADS", "").strip().lower()
|
|
if not raw or raw == "auto":
|
|
return None
|
|
try:
|
|
value = int(raw)
|
|
except ValueError as exc:
|
|
raise RuntimeError("MBT_VIDEO_THREADS must be auto or a positive integer") from exc
|
|
if value < 1:
|
|
raise RuntimeError("MBT_VIDEO_THREADS must be auto or a positive integer")
|
|
return value
|
|
|
|
|
|
def _prompt_threads(*, default: int | None) -> int | None:
|
|
default_text = str(default) if default is not None else "auto"
|
|
while True:
|
|
value = prompt_input(f"Encoder threads [{default_text}] (auto or positive integer): ").strip().lower()
|
|
if not value or value == "auto":
|
|
return default
|
|
try:
|
|
threads = int(value)
|
|
except ValueError:
|
|
print("Threads must be auto or a positive integer.")
|
|
continue
|
|
if threads >= 1:
|
|
return threads
|
|
print("Threads must be auto or a positive integer.")
|
|
|
|
|
|
def ask_audio_options(
|
|
*,
|
|
has_speed_changes: bool,
|
|
container_extension: str,
|
|
) -> AudioEncodeOptions:
|
|
env_mode = os.environ.get("MBT_VIDEO_AUDIO", "").strip().lower()
|
|
env_bitrate = os.environ.get("MBT_VIDEO_AUDIO_BITRATE", "").strip()
|
|
preserve_pitch = ask_audio_pitch_option(has_speed_changes=has_speed_changes)
|
|
if env_mode:
|
|
mode = _normalize_audio_mode(env_mode)
|
|
_validate_audio_container(mode, container_extension)
|
|
return AudioEncodeOptions(
|
|
mode=mode,
|
|
bitrate=env_bitrate or default_audio_bitrate(mode),
|
|
preserve_pitch=preserve_pitch,
|
|
)
|
|
|
|
if not sys.stdin.isatty():
|
|
return AudioEncodeOptions(mode="none", preserve_pitch=preserve_pitch)
|
|
|
|
while True:
|
|
if container_extension == ".mkv":
|
|
prompt = "Audio: Enter=Opus, a=AAC, f=FLAC, c=copy/trim audio, n=no audio: "
|
|
else:
|
|
prompt = "Audio: Enter=AAC re-encode, c=copy/trim audio, n=no audio: "
|
|
answer = prompt_input(prompt).strip().lower()
|
|
if not answer:
|
|
mode = "opus" if container_extension == ".mkv" else "aac"
|
|
break
|
|
if answer in {"a", "aac"}:
|
|
mode = "aac"
|
|
break
|
|
if container_extension == ".mkv" and answer in {"o", "opus"}:
|
|
mode = "opus"
|
|
break
|
|
if container_extension == ".mkv" and answer in {"f", "flac"}:
|
|
mode = "flac"
|
|
break
|
|
if answer in {"c", "copy"}:
|
|
mode = "copy"
|
|
break
|
|
if answer in {"n", "no", "none"}:
|
|
mode = "none"
|
|
break
|
|
print("Please choose Enter, c, or n.")
|
|
|
|
bitrate = default_audio_bitrate(mode)
|
|
if mode in {"aac", "opus"}:
|
|
label = "AAC" if mode == "aac" else "Opus"
|
|
bitrate = prompt_input(f"{label} audio bitrate [{bitrate}]: ").strip() or bitrate
|
|
return AudioEncodeOptions(mode=mode, bitrate=bitrate, preserve_pitch=preserve_pitch)
|
|
|
|
|
|
def ask_audio_pitch_option(*, has_speed_changes: bool) -> bool:
|
|
env_value = os.environ.get("MBT_VIDEO_AUDIO_PITCH", "").strip().lower()
|
|
if env_value:
|
|
if env_value in {"preserve", "preserved", "keep", "same"}:
|
|
return True
|
|
if env_value in {"shift", "change", "changed", "speed"}:
|
|
return False
|
|
raise RuntimeError("MBT_VIDEO_AUDIO_PITCH must be preserve or shift")
|
|
|
|
if not has_speed_changes or not sys.stdin.isatty():
|
|
return True
|
|
|
|
while True:
|
|
answer = prompt_input(
|
|
"Speed-changed audio pitch: Enter=preserve pitch, s=shift with speed: "
|
|
).strip().lower()
|
|
if not answer or answer in {"p", "preserve", "keep", "same"}:
|
|
return True
|
|
if answer in {"s", "shift", "change", "speed"}:
|
|
return False
|
|
print("Please choose Enter or s.")
|
|
|
|
|
|
def _normalize_audio_mode(value: str) -> str:
|
|
if value in {"aac", "reencode", "re-encode"}:
|
|
return "aac"
|
|
if value in {"opus", "libopus"}:
|
|
return "opus"
|
|
if value in {"flac", "lossless"}:
|
|
return "flac"
|
|
if value in {"copy", "streamcopy", "stream-copy"}:
|
|
return "copy"
|
|
if value in {"none", "no", "off", "0"}:
|
|
return "none"
|
|
raise RuntimeError("MBT_VIDEO_AUDIO must be aac, opus, flac, copy, or none")
|
|
|
|
|
|
def _validate_audio_container(mode: str, container_extension: str) -> None:
|
|
if container_extension == ".mp4" and mode in {"opus", "flac"}:
|
|
raise RuntimeError("MP4 output supports aac, copy, or none audio in this tool")
|
|
|
|
|
|
def _container_extension_from_env() -> str | None:
|
|
raw = os.environ.get("MBT_VIDEO_CONTAINER", "").strip().lower()
|
|
if not raw:
|
|
return None
|
|
return _normalize_container_extension(raw)
|
|
|
|
|
|
def _prompt_container_extension() -> str:
|
|
while True:
|
|
answer = prompt_input("Output container: Enter=MP4, m=MKV: ").strip().lower()
|
|
if not answer:
|
|
return ".mp4"
|
|
try:
|
|
return _normalize_container_extension(answer)
|
|
except ValueError as exc:
|
|
print(exc)
|
|
|
|
|
|
def _normalize_container_extension(value: str) -> str:
|
|
if value in {"mp4", ".mp4"}:
|
|
return ".mp4"
|
|
if value in {"m", "mkv", ".mkv", "matroska"}:
|
|
return ".mkv"
|
|
raise ValueError("container must be mp4 or mkv")
|
|
|
|
|
|
def _codec_from_env() -> str | None:
|
|
raw = os.environ.get("MBT_VIDEO_CODEC", "").strip().lower()
|
|
if not raw:
|
|
return None
|
|
return _normalize_codec(raw)
|
|
|
|
|
|
def _prompt_codec() -> str:
|
|
while True:
|
|
answer = prompt_input("Video codec: x264, Enter=x265: ").strip().lower()
|
|
if not answer:
|
|
return "libx265"
|
|
try:
|
|
return _normalize_codec(answer)
|
|
except ValueError as exc:
|
|
print(exc)
|
|
|
|
|
|
def _normalize_codec(value: str) -> str:
|
|
if value in {"x264", "h264", "libx264"}:
|
|
return "libx264"
|
|
if value in {"x265", "h265", "hevc", "libx265"}:
|
|
return "libx265"
|
|
raise ValueError("codec must be x264 or x265")
|
|
|
|
|
|
def _int_from_env(name: str) -> int | None:
|
|
raw = os.environ.get(name, "").strip()
|
|
if not raw:
|
|
return None
|
|
try:
|
|
return int(raw)
|
|
except ValueError as exc:
|
|
raise RuntimeError(f"{name} must be an integer") from exc
|
|
|
|
|
|
def _optional_env(name: str) -> str | None:
|
|
value = os.environ.get(name, "").strip()
|
|
return value or None
|
|
|
|
|
|
def _prompt_int(prompt: str, *, default: int) -> int:
|
|
while True:
|
|
answer = prompt_input(prompt).strip()
|
|
if not answer:
|
|
return default
|
|
try:
|
|
return int(answer)
|
|
except ValueError:
|
|
print("Please enter an integer.")
|