1769 lines
61 KiB
Python
Executable File
1769 lines
61 KiB
Python
Executable File
#!/home/lucas/scripts/.venv/bin/python3
|
||
"""
|
||
FFmpeg Python CLI Tool
|
||
|
||
A wrapper for ffmpeg-python to perform common media tasks.
|
||
Parameters are documented with their equivalent FFmpeg command-line flags.
|
||
"""
|
||
|
||
import argparse
|
||
import os
|
||
import sys
|
||
import json
|
||
import ffmpeg
|
||
import re
|
||
import logging
|
||
import shutil
|
||
import subprocess
|
||
import tempfile
|
||
from pathlib import Path
|
||
from mutagen.easyid3 import EasyID3
|
||
from serverlibrary import safe_rename,get_audio_tracks,get_subtitle_tracks
|
||
|
||
|
||
TRACE = 5
|
||
class CustomLogger(logging.Logger):
|
||
def trace(self, msg: str, *args, **kwargs) -> None:
|
||
if self.isEnabledFor(TRACE):
|
||
self._log(TRACE, msg, args, **kwargs)
|
||
|
||
|
||
logging.setLoggerClass(CustomLogger)
|
||
logger: CustomLogger = logging.getLogger(__name__) # type: ignore[assignment]
|
||
LOG_LEVELS = {
|
||
"TRACE": TRACE,
|
||
"DEBUG": logging.DEBUG,
|
||
"INFO": logging.INFO,
|
||
"WARNING": logging.WARNING,
|
||
"ERROR": logging.ERROR,
|
||
"CRITICAL": logging.CRITICAL,
|
||
}
|
||
|
||
|
||
VIDEO_EXTS = {".mp4", ".mkv", ".avi", ".mov", ".webm"}
|
||
AUDIO_EXTS = {".mp3", ".m4a", ".m4b", ".opus"}
|
||
|
||
# ----------------------------
|
||
# OUTPUT ENFORCEMENT (clean + centralized)
|
||
# ----------------------------
|
||
commands = {
|
||
"test":{"name":"test", "output": False},
|
||
"info":{"name":"info", "output": False},
|
||
"convert-av":{"name":"convert-av", "output": True},
|
||
"convert-audio":{"name":"convert-audio", "output": True},
|
||
"convert-video":{"name":"convert-video", "output": True},
|
||
"remove-subs":{"name":"remove-subs", "output": True},
|
||
"remove-audio":{"name":"remove-audio", "output": True},
|
||
"extract-audio":{"name":"extract-audio", "output": True},
|
||
"downsize":{"name":"downsize", "output": True},
|
||
"filter-audio":{"name":"filter-audio", "output": True},
|
||
"cut":{"name":"cut", "output": True},
|
||
"set-primary-audio":{"name":"set-primary-audio", "output": True},
|
||
"chapters":{"name":"chapters", "output": True},
|
||
"metadata-print":{"name":"metadata-print", "output": False},
|
||
"metadata-set":{"name":"metadata-set", "output": True},
|
||
"remove-russian-from-db":{"name":"remove-russian-from-db", "output": False},
|
||
}
|
||
# ------ Russian Dictonary --------------#
|
||
AUDIO_TRANSLATIONS = {
|
||
"Многоголосное озвучивание": "Multi-voice dub",
|
||
"Многоголосый": "Multi-voice",
|
||
"Любительский": "Amateur voice-over",
|
||
"Профессиональный": "Professional",
|
||
"Одноголосый": "Single voice",
|
||
"Двухголосый": "Dual voice",
|
||
"Авторский": "Author voice-over",
|
||
"Оригинал": "Original",
|
||
"Дубляж": "Dubbed",
|
||
"Русский": "Russian",
|
||
"Русская": "Russian",
|
||
"Русское": "Russian",
|
||
"rus": "Russian",
|
||
"ru": "Russian",
|
||
"ger": "German",
|
||
"deu": "German",
|
||
"eng": "English",
|
||
"und": "Unknown",
|
||
}
|
||
RUSSIAN_RE = re.compile(
|
||
r"(?:"
|
||
r"\brus\b|"
|
||
r"\bru\b|"
|
||
r"russian|"
|
||
r"dubbed|"
|
||
r"рус|"
|
||
r"русский|"
|
||
r"русская|"
|
||
r"дубляж|"
|
||
r"оригинал|"
|
||
r"многоголос|"
|
||
r"двухголос|"
|
||
r"одноголос|"
|
||
r"авторский|"
|
||
r"любительский"
|
||
r")",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
def is_russian_track(description: str) -> bool:
|
||
return bool(RUSSIAN_RE.search(description))
|
||
|
||
class ArgsBuffer:
|
||
input: Path | None = None
|
||
output: Path | None = None
|
||
kidflag: bool = False
|
||
loud: bool = False
|
||
key: str | None = None
|
||
value: str | None = None
|
||
audio_index: str | None = None
|
||
dry_run: bool | None = None
|
||
|
||
def translate_audio_description(description: str) -> str:
|
||
for src, dst in AUDIO_TRANSLATIONS.items():
|
||
description = re.sub(
|
||
rf"\b{re.escape(src)}\b",
|
||
dst,
|
||
description,
|
||
flags=re.IGNORECASE,
|
||
)
|
||
|
||
CYRILLIC_RE = re.compile(r"[А-Яа-яЁё]")
|
||
|
||
if CYRILLIC_RE.search(description):
|
||
logger.error(f"Contains unidentified Cyrillic Charachters {description}")
|
||
raise ValueError(
|
||
f"Cyrillic characters found in audio description: {description}"
|
||
)
|
||
return description
|
||
|
||
def translate_audio_tracks(tracks: list[dict]) -> list[dict]:
|
||
return [
|
||
{
|
||
**track,
|
||
"description": translate_audio_description(
|
||
track.get("description", "")
|
||
),
|
||
}
|
||
for track in tracks
|
||
]
|
||
|
||
# ---------- Utilities ---------- #
|
||
|
||
# ----------------------------
|
||
# HELPERS
|
||
# ----------------------------
|
||
def ask(prompt, default=None):
|
||
value = input(f"{prompt}" + (f" [{default}]" if default else "") + ": ").strip()
|
||
return value if value else default
|
||
|
||
def ensure(value, prompt):
|
||
if value is None:
|
||
return ask(prompt)
|
||
return value
|
||
|
||
def move_override_file(input_path, backup_files=False,silent=False):
|
||
if input_path.is_file():
|
||
input_path=input_path.parent
|
||
for override_path in input_path.glob("*_override.*"):
|
||
if not override_path.is_file():
|
||
continue
|
||
|
||
stem = override_path.stem
|
||
suffix = override_path.suffix
|
||
|
||
if not stem.endswith("_override"):
|
||
continue
|
||
|
||
base = stem.removesuffix("_override")
|
||
|
||
original = input_path / f"{base}{suffix}"
|
||
backup = input_path / f".backup_{base}{suffix}"
|
||
new_name = input_path / f"{base}{suffix}"
|
||
|
||
if not original.exists():
|
||
logger.info(f"Skip (missing original): {original}")
|
||
continue
|
||
|
||
if backup_files:
|
||
# move original -> backup
|
||
safe_rename(original, backup,silent=silent)
|
||
|
||
# move override -> original name
|
||
safe_rename(override_path, new_name,silent=silent)
|
||
|
||
|
||
LOG_FILE = Path.home() / "ffmpeg_log.log"
|
||
|
||
def run_ffmpeg(stream, overwrite=True, loud=False):
|
||
kwargs = {
|
||
"overwrite_output": overwrite,
|
||
}
|
||
|
||
|
||
kwargs.update(
|
||
capture_stdout=True,
|
||
capture_stderr=True,
|
||
quiet=True,
|
||
)
|
||
|
||
try:
|
||
out, err = ffmpeg.run(stream, **kwargs)
|
||
logger.trace(out)
|
||
if err:
|
||
logger.error(err)
|
||
except ffmpeg.Error as e:
|
||
logger.error(e.stderr)
|
||
logger.error(f"FFmpeg failed.\n")
|
||
sys.exit(1)
|
||
|
||
def probe_audio_streams(file):
|
||
"""
|
||
Probes metadata to identify audio tracks.
|
||
|
||
FFmpeg Equivalent:
|
||
ffprobe -v error -select_streams a -show_entries stream=index:tags=language -of json
|
||
"""
|
||
try:
|
||
probe = ffmpeg.probe(file)
|
||
except ffmpeg.Error as e:
|
||
logger.error(f"ffprobe failed:", e)
|
||
sys.exit(1)
|
||
|
||
return [
|
||
(s.get("index"), s.get("tags", {}).get("language", "unknown"))
|
||
for s in probe.get("streams", [])
|
||
if s.get("codec_type") == "audio"
|
||
]
|
||
|
||
def select_audio_stream(file, provided=None):
|
||
"""
|
||
Logical handler for stream selection. Ensures we target the correct
|
||
absolute stream index when building filter graphs.
|
||
"""
|
||
streams = probe_audio_streams(file)
|
||
|
||
if not streams:
|
||
logger.info(f"No audio streams found.")
|
||
sys.exit(1)
|
||
|
||
provided = int(provided) if provided else None
|
||
if provided is not None:
|
||
if any(s[0] == provided for s in streams):
|
||
return provided
|
||
logger.info(f"Provided index {provided} is not a valid audio stream.")
|
||
|
||
if len(streams) == 1:
|
||
return streams[0][0]
|
||
|
||
logger.info(f"Found {len(streams)} audio streams:")
|
||
for i, (idx, lang) in enumerate(streams):
|
||
logger.info(f"[{i}] Index: {idx}, Language: {lang}")
|
||
|
||
try:
|
||
choice = int(input("Select stream number (the list index [x]): "))
|
||
return streams[choice][0]
|
||
except (ValueError, IndexError):
|
||
logger.error(f"Invalid selection.")
|
||
sys.exit(1)
|
||
|
||
def parse_time_to_seconds(value: str) -> float:
|
||
"""
|
||
Convert human-readable time formats into seconds.
|
||
|
||
Supported formats:
|
||
- "100s" → 100
|
||
- "1m20s" → 80
|
||
- "2h10m5s" → 7805
|
||
- "0h1m5s" → 65
|
||
- "0:1:5" → 65 (hh:mm:ss or mm:ss)
|
||
"""
|
||
|
||
if value is None:
|
||
return None
|
||
|
||
value = str(value).strip()
|
||
|
||
# Case 1: pure seconds (e.g. 100s)
|
||
if value.endswith("s") and value[:-1].isdigit():
|
||
return float(value[:-1])
|
||
|
||
# Case 2: h/m/s format (e.g. 1h2m3s)
|
||
pattern = re.compile(
|
||
r"(?:(?P<h>\d+)h)?(?:(?P<m>\d+)m)?(?:(?P<s>\d+)s)?"
|
||
)
|
||
match = pattern.fullmatch(value)
|
||
if match and any(match.groups()):
|
||
h = int(match.group("h") or 0)
|
||
m = int(match.group("m") or 0)
|
||
s = int(match.group("s") or 0)
|
||
return h * 3600 + m * 60 + s
|
||
|
||
# Case 3: colon format (hh:mm:ss or mm:ss)
|
||
if ":" in value:
|
||
parts = list(map(int, value.split(":")))
|
||
if len(parts) == 2:
|
||
m, s = parts
|
||
return m * 60 + s
|
||
elif len(parts) == 3:
|
||
h, m, s = parts
|
||
return h * 3600 + m * 60 + s
|
||
|
||
# fallback: raw number
|
||
try:
|
||
return float(value)
|
||
except ValueError:
|
||
raise argparse.ArgumentTypeError(f"Invalid time format: {value}")
|
||
|
||
def get_scodec(output_path: str | Path):
|
||
ext = str(output_path).lower().split(".")[-1]
|
||
|
||
if ext == "mp4":
|
||
return "mov_text"
|
||
|
||
if ext == "mkv":
|
||
return "srt"
|
||
|
||
return "copy"
|
||
|
||
def get_vcodec(output_path: str):
|
||
ext = output_path.lower().split(".")[-1]
|
||
|
||
if ext == "mp4":
|
||
return "libx264"
|
||
|
||
return "copy"
|
||
|
||
def get_chapter_lengths(file_path):
|
||
"""
|
||
Returns an array of chapter durations (in seconds).
|
||
"""
|
||
|
||
cmd = [
|
||
"ffprobe",
|
||
"-v", "error",
|
||
"-print_format", "json",
|
||
"-show_chapters",
|
||
file_path
|
||
]
|
||
|
||
result = subprocess.run(cmd,
|
||
capture_output=True,
|
||
text=True,
|
||
check=True,
|
||
)
|
||
logger.trace(result.stdout)
|
||
if result.returncode != 0:
|
||
raise RuntimeError(result.stderr)
|
||
|
||
data = json.loads(result.stdout)
|
||
|
||
chapters = []
|
||
for ch in data.get("chapters", []):
|
||
chapters.append({
|
||
"start": float(ch["start_time"]),
|
||
"end": float(ch["end_time"])
|
||
})
|
||
|
||
return chapters
|
||
|
||
def natural_sort_key(path: Path):
|
||
s = str(path.name) # force string conversion
|
||
return [
|
||
int(part) if part.isdigit() else part.lower()
|
||
for part in re.split(r'(\d+)', s)
|
||
]
|
||
# ---------- Commands ---------- #
|
||
def cmd_info(args):
|
||
"""
|
||
Retrieves and prints metadata.
|
||
|
||
FFmpeg Equivalent:
|
||
ffprobe -show_format -show_streams -of json [input]
|
||
"""
|
||
try:
|
||
info = ffmpeg.probe(args.input)
|
||
logger.info(json.dumps(info, indent=2))
|
||
except ffmpeg.Error as e:
|
||
logger.error(f"Error probing file:", e)
|
||
|
||
def cmd_convert_av(args):
|
||
"""
|
||
Re-encodes both video and audio.
|
||
|
||
FFmpeg Equivalent:
|
||
ffmpeg -i [input] -c:v [video_codec] -c:a [audio_codec] -preset [preset] -crf [quality] [output]
|
||
|
||
Parameters:
|
||
- vcodec/c:v: Sets the video encoder (e.g., libx264).
|
||
- acodec/c:a: Sets the audio encoder (e.g., aac).
|
||
- preset: Trade-off between compression speed and efficiency (ultrafast to veryslow).
|
||
- qp/crf: Constant Rate Factor. Lower is higher quality (18-28 is standard).
|
||
"""
|
||
inp = ffmpeg.input(args.input)
|
||
out = ffmpeg.output(
|
||
inp,
|
||
args.output,
|
||
vcodec=args.video_codec,
|
||
acodec=args.audio_codec,
|
||
scodec=get_scodec(args.output),
|
||
preset=args.preset,
|
||
qp=args.quality,
|
||
# This maps all streams from the first input once
|
||
map='0',
|
||
# Prevents FFmpeg from trying to guess/add extra default streams
|
||
map_metadata=0
|
||
)
|
||
return out
|
||
|
||
def cmd_convert_audio(args):
|
||
"""
|
||
Transcodes audio while keeping video intact.
|
||
|
||
FFmpeg Equivalent:
|
||
ffmpeg -i [input] -map 0:v -map 0:a -c:v copy -c:a [audio_codec] [output]
|
||
|
||
Parameters:
|
||
- vcodec='copy': Enables 'stream copy' mode, bypassing the video decoder/encoder.
|
||
"""
|
||
inp = ffmpeg.input(args.input)
|
||
out = ffmpeg.output(
|
||
inp,
|
||
args.output,
|
||
vcodec="copy",
|
||
acodec=args.audio_codec,
|
||
scodec=get_scodec(args.output),
|
||
map='0',
|
||
map_metadata=0
|
||
)
|
||
return out
|
||
|
||
def cmd_convert_video(args):
|
||
"""
|
||
Transcodes video while keeping audio intact.
|
||
|
||
FFmpeg Equivalent:
|
||
ffmpeg -i [input] -map 0:v -map 0:a -c:v [video_codec] -c:a copy [output]
|
||
"""
|
||
inp = ffmpeg.input(args.input)
|
||
out = ffmpeg.output(
|
||
inp,
|
||
args.output,
|
||
vcodec=args.video_codec,
|
||
acodec="copy",
|
||
scodec=get_scodec(args.output),
|
||
map='0',
|
||
map_metadata=0
|
||
)
|
||
return out
|
||
|
||
def cmd_remove_subs(args):
|
||
"""
|
||
Strips all subtitle data.
|
||
|
||
FFmpeg Equivalent:
|
||
ffmpeg -i [input] -sn -c copy [output]
|
||
|
||
Parameters:
|
||
- sn: Disables subtitle recording.
|
||
"""
|
||
probe = ffmpeg.probe(args.input)
|
||
|
||
cmd = [
|
||
"ffmpeg",
|
||
"-y",
|
||
"-i", str(args.input),
|
||
]
|
||
|
||
if args.num is None:
|
||
# remove all subtitle streams
|
||
for stream in probe["streams"]:
|
||
if stream["codec_type"] != "subtitle":
|
||
cmd.extend(["-map", f"0:{stream['index']}"])
|
||
else:
|
||
# remove only specified subtitle streams
|
||
for stream in probe["streams"]:
|
||
if (
|
||
stream["codec_type"] == "subtitle"
|
||
and stream["index"] in args.num
|
||
):
|
||
continue
|
||
|
||
cmd.extend(["-map", f"0:{stream['index']}"])
|
||
|
||
cmd.extend([
|
||
"-c", "copy",
|
||
"-map_metadata", "0",
|
||
str(args.output),
|
||
])
|
||
|
||
logger.debug("%s", " ".join(cmd))
|
||
|
||
try:
|
||
subprocess.run(
|
||
cmd,
|
||
capture_output=True,
|
||
text=True,
|
||
check=True,
|
||
)
|
||
except subprocess.CalledProcessError as e:
|
||
logger.error("Command: %s", " ".join(e.cmd))
|
||
logger.error("Return code: %s", e.returncode)
|
||
logger.error("STDOUT:\n%s", e.stdout)
|
||
logger.error("STDERR:\n%s", e.stderr)
|
||
raise
|
||
return "No output needed"
|
||
def cmd_remove_audio_track(args):
|
||
"""
|
||
1. Removes a specific audio track from the video (Main Output).
|
||
"""
|
||
probe = ffmpeg.probe(args.input)
|
||
|
||
remove_idx = select_audio_stream(args.input, args.audio_index)
|
||
|
||
remaining_audio = [
|
||
{
|
||
**s,
|
||
"description": translate_audio_description(
|
||
s.get("description", "")
|
||
),
|
||
}
|
||
for s in probe["streams"]
|
||
if s["codec_type"] == "audio"
|
||
and s["index"] != remove_idx
|
||
]
|
||
|
||
removed_was_default = any(
|
||
s["index"] == remove_idx
|
||
and s.get("disposition", {}).get("default", 0) == 1
|
||
for s in probe["streams"]
|
||
)
|
||
|
||
cmd = [
|
||
"ffmpeg",
|
||
"-y",
|
||
"-i", str(args.input),
|
||
]
|
||
|
||
for stream in probe["streams"]:
|
||
if stream["index"] != remove_idx:
|
||
cmd.extend(["-map", f"0:{stream['index']}"])
|
||
|
||
cmd.extend([
|
||
"-c:v", "copy",
|
||
"-c:a", "copy",
|
||
"-c:s", get_scodec(args.output),
|
||
])
|
||
|
||
if removed_was_default and remaining_audio:
|
||
# First remaining audio stream in the output becomes default
|
||
cmd.extend([
|
||
"-disposition:a", "0",
|
||
"-disposition:a:0", "default",
|
||
])
|
||
|
||
cmd.append(str(args.output))
|
||
logger.debug(" ".join(cmd))
|
||
if args.dry_run:
|
||
return "No Output needed"
|
||
try:
|
||
result = subprocess.run(
|
||
cmd,
|
||
capture_output=True,
|
||
text=True,
|
||
check=True,
|
||
)
|
||
logger.debug("FFmpegsubprocess successful")
|
||
logger.trace(result.stdout)
|
||
return "No output needed"
|
||
|
||
except subprocess.CalledProcessError as e:
|
||
logging.error(
|
||
"Command failed (%s):\n%s\n\nstderr:\n%s",
|
||
e.returncode,
|
||
" ".join(map(str, e.cmd)),
|
||
e.stderr,
|
||
)
|
||
raise RuntimeError(f"Command failed: {' '.join(e.cmd)}") from e
|
||
|
||
def cmd_extract_audio_track(args):
|
||
"""
|
||
1. Saves that specific audio track as a standalone MP3 (Extra Output).
|
||
"""
|
||
inp = ffmpeg.input(args.input)
|
||
probe = ffmpeg.probe(args.input)
|
||
|
||
output_audio = Path(args.output)
|
||
output_audio = str(output_audio.with_name(f"{output_audio.stem}.mp3"))
|
||
|
||
remove_idx = select_audio_stream(args.input, args.audio_index)
|
||
|
||
extracted_audio = None
|
||
|
||
for s in probe["streams"]:
|
||
idx = s["index"]
|
||
if idx == remove_idx:
|
||
extracted_audio = inp[str(idx)]
|
||
|
||
out = ffmpeg.output(
|
||
extracted_audio,
|
||
output_audio,
|
||
acodec='libmp3lame',
|
||
audio_bitrate='192k'
|
||
)
|
||
return out
|
||
|
||
def cmd_downsize(args):
|
||
"""
|
||
Scales resolution and compresses.
|
||
|
||
FFmpeg Equivalent:
|
||
ffmpeg -i [input] -vf scale=[w]:[h] -c:v libx264 -crf [crf] -c:a copy [output]
|
||
|
||
Parameters:
|
||
- filter('scale'): Invokes the video filter (vf) for resizing.
|
||
- crf: Set to 23 (default) for balanced quality.
|
||
"""
|
||
inp = ffmpeg.input(args.input)
|
||
video = inp.video.filter("scale", args.width, args.height)
|
||
info = ffmpeg.probe(args.input)
|
||
streams = info.get("streams", [])
|
||
subtitle_streams = [s for s in streams if s.get("codec_type") == "subtitle"]
|
||
|
||
subs = {}
|
||
if len(subtitle_streams) > 0:
|
||
subs = {inp["s"]}
|
||
|
||
out = ffmpeg.output(
|
||
video,
|
||
inp.audio,
|
||
*subs,
|
||
args.output,
|
||
vcodec="libx264",
|
||
crf=args.crf,
|
||
acodec="copy",
|
||
scodec=get_scodec(args.output)
|
||
)
|
||
return out
|
||
|
||
def cmd_filter_audio(args):
|
||
"""
|
||
Multi-stage audio mastering chain.
|
||
|
||
FFmpeg Equivalent:
|
||
ffmpeg -i [input] -filter_complex "[0:a:X]highpass...,acompressor...,equalizer...out" \
|
||
-map 0:v -map out -c:v copy -c:a libvorbis [output]
|
||
|
||
Filter Descriptions:
|
||
1. highpass: Cuts frequencies below 100Hz (removes low-end rumble).
|
||
2. acompressor: Reduces volume of loud peaks to increase overall clarity.
|
||
3. equalizer: Adjusts specific frequency bands (2.5kHz and 4kHz for presence).
|
||
4. deesser: Dampens harsh 'S' sounds.
|
||
5. alimiter: Hard limit at 0.95 to prevent digital clipping (distortion).
|
||
6. dynaudnorm: Smooths out loudness levels over time.
|
||
"""
|
||
|
||
idx_filt = select_audio_stream(args.input, args.audio_index)
|
||
|
||
inp = ffmpeg.input(args.input)
|
||
probe = ffmpeg.probe(args.input)
|
||
|
||
audio_streams = []
|
||
audio_titles = {}
|
||
audio_languages = {}
|
||
|
||
for s in probe["streams"]:
|
||
if s.get("codec_type") == "audio":
|
||
idx = s["index"]
|
||
audio_streams.append(idx)
|
||
|
||
tags = s.get("tags", {})
|
||
audio_titles[idx] = tags.get("title", f"Audio {idx}")
|
||
audio_languages[idx] = tags.get("language", "und")
|
||
|
||
filtered_audios = []
|
||
metadata_args = []
|
||
|
||
out_audio_idx = 0
|
||
|
||
for idx in audio_streams:
|
||
audio = inp[str(idx)]
|
||
|
||
# -------------------
|
||
# ORIGINAL STREAM
|
||
# -------------------
|
||
filtered_audios.append(audio)
|
||
|
||
metadata_args += [
|
||
f"-metadata:s:a:{out_audio_idx}",
|
||
f"title={audio_titles[idx]}",
|
||
|
||
f"-metadata:s:a:{out_audio_idx}",
|
||
f"language={audio_languages[idx]}",
|
||
]
|
||
|
||
out_audio_idx += 1
|
||
|
||
# -------------------
|
||
# FILTERED STREAM
|
||
# -------------------
|
||
if idx == idx_filt:
|
||
audio_filtered = (
|
||
audio
|
||
.filter("highpass", f=100)
|
||
.filter("acompressor", threshold="-22dB", ratio=4, attack=10, release=150)
|
||
.filter("equalizer", f=2500, t="q", w=1, g=5)
|
||
.filter("equalizer", f=4000, t="q", w=1, g=4)
|
||
.filter("deesser", i=0.4, f=0.25)
|
||
.filter("alimiter", limit=0.95)
|
||
.filter("dynaudnorm", f=200, g=15)
|
||
)
|
||
|
||
filtered_audios.append(audio_filtered)
|
||
|
||
base_title = audio_titles[idx]
|
||
filtered_title = base_title + " + filtered"
|
||
lang = audio_languages[idx]
|
||
|
||
metadata_args += [
|
||
f"-metadata:s:a:{out_audio_idx}",
|
||
f"title={filtered_title}",
|
||
|
||
f"-metadata:s:a:{out_audio_idx}",
|
||
f"language={lang}",
|
||
|
||
f"-metadata:s:a:{out_audio_idx}",
|
||
f"handler_name={filtered_title}",
|
||
]
|
||
|
||
out_audio_idx += 1
|
||
|
||
# -------------------
|
||
# SUBTITLES
|
||
# -------------------
|
||
subs = []
|
||
if any(s.get("codec_type") == "subtitle" for s in probe["streams"]):
|
||
subs.append(inp["s"])
|
||
|
||
# -------------------
|
||
# OUTPUT (IMPORTANT FIX HERE)
|
||
# -------------------
|
||
input_args = [inp.video,
|
||
*filtered_audios,
|
||
*subs,]
|
||
|
||
out = ffmpeg.output(
|
||
*input_args,
|
||
args.output,
|
||
vcodec="copy",
|
||
acodec="libvorbis",
|
||
scodec=get_scodec(args.output)
|
||
).global_args(*metadata_args)
|
||
|
||
|
||
return out
|
||
|
||
def cmd_cut_segment(args):
|
||
"""
|
||
Cut a segment from a video or audio file.
|
||
|
||
FFmpeg parameters:
|
||
ss (start time): where to begin cutting (seconds or timestamp)
|
||
t (duration): length of segment to extract
|
||
c=copy: stream copy (fast, no re-encoding)
|
||
|
||
Notes:
|
||
- Using ss before input enables fast seek
|
||
- Copy mode is fast but may be less frame-accurate
|
||
"""
|
||
|
||
start = parse_time_to_seconds(args.start)
|
||
duration = parse_time_to_seconds(args.duration)
|
||
|
||
inp = ffmpeg.input(args.input, ss=start)
|
||
|
||
out = ffmpeg.output(
|
||
inp,
|
||
args.output,
|
||
t=duration,
|
||
map="0",
|
||
vcodec=get_vcodec(args.output),
|
||
acodec="copy",
|
||
scodec=get_scodec(args.output),
|
||
)
|
||
logger.trace(f"Command: {out}")
|
||
return out
|
||
|
||
def cmd_set_primary_audio(args):
|
||
"""
|
||
Set a chosen audio stream as the PRIMARY (first/default) audio track.
|
||
|
||
Uses select_audio_stream() to ensure correct stream selection.
|
||
|
||
FFmpeg behavior:
|
||
- We explicitly reorder audio streams
|
||
- Selected stream becomes first audio track
|
||
- All other audio streams are preserved
|
||
- Video and subtitles are kept
|
||
|
||
Important:
|
||
FFmpeg does NOT support "set default audio track" directly in MP4.
|
||
So we simulate it by stream reordering.
|
||
"""
|
||
|
||
inp = ffmpeg.input(args.input)
|
||
|
||
primary_audio = select_audio_stream(args.input, args.audio_index)
|
||
probe = ffmpeg.probe(args.input)
|
||
|
||
audio_streams = [
|
||
s["index"]
|
||
for s in probe["streams"]
|
||
if s.get("codec_type") == "audio"
|
||
]
|
||
|
||
reordered_audio = [primary_audio] + [
|
||
a for a in audio_streams if a != primary_audio
|
||
]
|
||
|
||
audio_inputs = [inp[f"a:{i - 1}"] for i in reordered_audio]
|
||
|
||
out_kwargs={}
|
||
for i, stream in enumerate(reordered_audio):
|
||
if stream == primary_audio:
|
||
out_kwargs[f"disposition:a:{i}"] = "default"
|
||
else:
|
||
out_kwargs[f"disposition:a:{i}"] = "0"
|
||
|
||
info = ffmpeg.probe(args.input)
|
||
streams = info.get("streams", [])
|
||
subtitle_streams = [s for s in streams if s.get("codec_type") == "subtitle"]
|
||
|
||
subs = {}
|
||
if len(subtitle_streams) > 0:
|
||
subs = {inp["s"]}
|
||
|
||
out = ffmpeg.output(
|
||
inp.video,
|
||
*audio_inputs,
|
||
*subs,
|
||
args.output,
|
||
vcodec="copy",
|
||
acodec="copy",
|
||
scodec=get_scodec(args.output),
|
||
**out_kwargs
|
||
)
|
||
|
||
return out
|
||
|
||
def cmd_show_metadata(args):
|
||
"""
|
||
Prints metadata directly from ffprobe output.
|
||
"""
|
||
|
||
probe = ffmpeg.probe(args.input)
|
||
|
||
logger.info(f"=== FORMAT METADATA ===")
|
||
format_tags = probe.get("format", {}).get("tags", {})
|
||
for k, v in format_tags.items():
|
||
logger.info(f"{k}: {v}")
|
||
|
||
#logger.info(f"\n=== STREAM METADATA ===")
|
||
#for i, stream in enumerate(probe.get("streams", [])):
|
||
# logger.info(f"\n--- Stream {i} ({stream.get('codec_type')}) ---")
|
||
|
||
# tags = stream.get("tags", {})
|
||
# if not tags:
|
||
# logger.info(f"No tags")
|
||
# else:
|
||
# for k, v in tags.items():
|
||
# logger.info(f"{k}: {v}")
|
||
return format_tags
|
||
|
||
|
||
def cmd_set_metadata(args):
|
||
"""
|
||
Set metadata for media files.
|
||
|
||
- Default: uses ffmpeg (stream copy)
|
||
- If file is .opus: uses kid3-cli (Vorbis tags)
|
||
"""
|
||
|
||
ext = os.path.splitext(args.input)[1].lower()
|
||
|
||
# ─────────────────────────────────────────────
|
||
# OPUS → use Kid3 CLI (Vorbis comments)
|
||
# ─────────────────────────────────────────────
|
||
if ext == ".opus" or args.kidflag:
|
||
cmd = [
|
||
"kid3-cli",
|
||
"-c", f"set {args.key} '{args.value}'",
|
||
args.input
|
||
]
|
||
|
||
result = subprocess.run(cmd,
|
||
capture_output=True,
|
||
text=True,
|
||
check=True,
|
||
)
|
||
logger.trace(result.stdout)
|
||
return None # Kid3 writes in-place
|
||
|
||
# ─────────────────────────────────────────────
|
||
# Default → ffmpeg (copy stream)
|
||
# ─────────────────────────────────────────────
|
||
inp = ffmpeg.input(args.input)
|
||
|
||
out = ffmpeg.output(
|
||
inp,
|
||
str(args.output),
|
||
c="copy",
|
||
metadata=f"{args.key}={args.value}"
|
||
)
|
||
return out
|
||
|
||
def cmd_apply_chapters(args):
|
||
"""
|
||
Apply chapter titles to an M4B using a simple title-only text file.
|
||
|
||
Chapters file format:
|
||
One title per line:
|
||
Introduction
|
||
Acknowledgment
|
||
The War Begins
|
||
|
||
Behavior:
|
||
- Evenly distributes chapters across media duration
|
||
- Generates ffmetadata internally
|
||
- Applies chapters using ffmpeg-python
|
||
"""
|
||
# ----------------------------
|
||
# read chapter titles
|
||
# ----------------------------
|
||
with open(args.chapter_file, "r", encoding="utf-8") as f:
|
||
titles = [line.strip() for line in f if line.strip()]
|
||
|
||
if not titles:
|
||
raise ValueError("Chapter file is empty")
|
||
|
||
if Path(args.input).suffix == ".m4b":
|
||
# ----------------------------
|
||
# build ffmetadata chapters
|
||
# ----------------------------
|
||
chapters = get_chapter_lengths(args.input)
|
||
|
||
if len(chapters) != len(titles):
|
||
logger.error(f"Chapter lengths do not match: {len(chapters)}, {len(titles)}")
|
||
for idx, title in enumerate(titles):
|
||
chapters[idx].update({
|
||
"title": title
|
||
})
|
||
|
||
metadata_text = ";FFMETADATA1\n\n"
|
||
for ch in chapters:
|
||
start_ms = int(float(ch["start"]) * 1000)
|
||
end_ms = int(float(ch["end"]) * 1000)
|
||
metadata_text += (
|
||
"[CHAPTER]\n"
|
||
"TIMEBASE=1/1000\n"
|
||
f"START={start_ms}\n"
|
||
f"END={end_ms}\n"
|
||
f"title={ch['title']}\n\n"
|
||
)
|
||
|
||
# ----------------------------
|
||
# run ffmpeg
|
||
# ----------------------------
|
||
with tempfile.NamedTemporaryFile("w+", suffix=".txt", delete=False) as f:
|
||
f.write(metadata_text)
|
||
meta_path = f.name
|
||
|
||
cmd = [
|
||
"ffmpeg",
|
||
"-y",
|
||
"-i", args.input,
|
||
"-f", "ffmetadata",
|
||
"-i", meta_path,
|
||
"-map_chapters", "1",
|
||
"-map_metadata", "0",
|
||
"-c", "copy",
|
||
args.output
|
||
]
|
||
|
||
result = subprocess.run(cmd,
|
||
capture_output=True,
|
||
text=True,
|
||
check=True,
|
||
)
|
||
logger.trace(result.stdout)
|
||
|
||
elif Path(args.input).suffix == ".mp3":
|
||
filename = Path(args.input)
|
||
logger.info(f"{args.file_counter}:{filename}")
|
||
title = titles[args.file_counter]
|
||
safe_title = title.replace("/", "-").replace(":", "-")
|
||
new_name = f"{args.file_counter + 1:03d} - {safe_title}.mp3"
|
||
|
||
new_path = filename.with_name(new_name)
|
||
|
||
logger.info(f"{filename.name} -> {new_name}")
|
||
audio = EasyID3(filename)
|
||
audio["title"] = title
|
||
audio["tracknumber"] = str(args.file_counter+1)
|
||
audio.save()
|
||
safe_rename(filename, new_path)
|
||
|
||
return "No output needed"
|
||
|
||
def cmd_remove_russian_from_db(args_in):
|
||
"""
|
||
Find all media with russian audio
|
||
remove russian audio tracks
|
||
"""
|
||
media_files = []
|
||
for ext in VIDEO_EXTS:
|
||
media_files.extend(list(args_in.input.rglob("*"+ext)))
|
||
for file in media_files:
|
||
audio_tracks = translate_audio_tracks(get_audio_tracks(file))
|
||
logger.trace(f"File: {file.name}: Audio tracks: {audio_tracks}")
|
||
remove_tracks = []
|
||
for track in audio_tracks:
|
||
if is_russian_track(track["description"]):
|
||
remove_tracks.append(track["number"])
|
||
logger.debug(f"Removing russian track {track['number']} from {file}")
|
||
for number in remove_tracks:
|
||
args=ArgsBuffer()
|
||
args.input = file
|
||
args.dry_run = args_in.dry_run
|
||
args.audio_index = number
|
||
args.output = file.with_name(f"{file.stem}_remove_russian{file.suffix}")
|
||
if cmd_remove_audio_track(args) == "No output needed":
|
||
logger.debug(safe_rename(args.output, file,overwrite=True))
|
||
|
||
return "No output needed"
|
||
|
||
|
||
def cmd_test(args):
|
||
"""
|
||
TEST MODE:
|
||
Runs each command individually on test.mp4 and probes outputs.
|
||
"""
|
||
|
||
test_file_dir = Path("/home/lucas/scripts/media_editing/testfiles")
|
||
test_files = [ "test_auto.mkv", "test_auto.mp4","test_auto_no_subs.mkv", "test_auto_no_subs.mp4"]
|
||
test_files = [ test_file_dir / file for file in test_files ]
|
||
test_dir = Path(tempfile.mkdtemp(prefix="ffmpeg_test_"))
|
||
logger.info(f"Using test dir: {test_dir}")
|
||
|
||
RED = "\033[91m"
|
||
GREEN = "\033[92m"
|
||
YELLOW = "\033[93m"
|
||
BLUE = "\033[94m"
|
||
RESET = "\033[0m"
|
||
|
||
def red_tag(tag: str) -> str:
|
||
return f"{RED}[{tag}]{RESET}"
|
||
|
||
def green_tag(tag: str) -> str:
|
||
return f"{GREEN}[{tag}]{RESET}"
|
||
|
||
def run_and_probe(name, cmd_func, args, test_dict: list=[]):
|
||
logger.info(f"TEST: {name}")
|
||
# Run FFmpeg for this single command only
|
||
outputs = cmd_func(args)
|
||
if not outputs:
|
||
logger.error(f"No outputs returned.")
|
||
return False
|
||
if outputs != "No output needed":
|
||
try:
|
||
ffmpeg.run(
|
||
outputs,
|
||
overwrite_output=True,
|
||
quiet=True,
|
||
capture_stderr=True
|
||
)
|
||
except ffmpeg.Error as e:
|
||
logger.error(f"FFmpeg failed:")
|
||
if e.stderr:
|
||
logger.info(e.stderr.decode(errors="ignore"))
|
||
return False
|
||
logger.info(f"FFmpeg succeeded")
|
||
# Probe result file(s)
|
||
out_files = []
|
||
|
||
if hasattr(args, "output") and args.output:
|
||
out_files.append(args.output)
|
||
|
||
for i, f in enumerate(out_files):
|
||
if not os.path.exists(f):
|
||
logger.info(f"{red_tag("FAIL")} Missing output: {f}")
|
||
continue
|
||
|
||
try:
|
||
ret_va = validate_ffprobe(f,test_dict[i])
|
||
except ffmpeg.Error as e:
|
||
logger.error(f"{red_tag("FAIL")} Probe error: {f}")
|
||
logger.error(e)
|
||
return False
|
||
|
||
return ret_va
|
||
|
||
def get_streams(test_file):
|
||
info = ffmpeg.probe(test_file)
|
||
streams = info.get("streams", [])
|
||
|
||
video_streams = [s for s in streams if s.get("codec_type") == "video" and s.get("codec_name") != "mjpeg" ]
|
||
audio_streams = [s for s in streams if s.get("codec_type") == "audio"]
|
||
subtitle_streams = [s for s in streams if s.get("codec_type") == "subtitle"]
|
||
cover_image =[s for s in streams if s.get("codec_type") == "video" and s.get("codec_name") == "mjpeg" ]
|
||
if not os.path.exists(test_file):
|
||
logger.info(f"test file not found in script directory.")
|
||
sys.exit(1)
|
||
|
||
return video_streams,audio_streams,subtitle_streams,cover_image
|
||
|
||
def validate_ffprobe(file_path, expected: dict):
|
||
"""
|
||
Validate ffprobe output using a dictionary-based expectation.
|
||
|
||
Expected dict format:
|
||
{
|
||
"video_count": int,
|
||
"audio_count": int,
|
||
"subtitle_count": int,
|
||
"audio_codecs": list[str],
|
||
"video_codecs": list[str]
|
||
}
|
||
"""
|
||
|
||
if file_path.is_file():
|
||
result = subprocess.run(
|
||
[
|
||
"ffprobe",
|
||
"-v", "quiet",
|
||
"-print_format", "json",
|
||
"-show_chapters",
|
||
"-show_format",
|
||
file_path,
|
||
],
|
||
capture_output=True,
|
||
text=True,
|
||
check=True,
|
||
)
|
||
|
||
info = json.loads(result.stdout)
|
||
video_streams, audio_streams, subtitle_streams, cover_image = get_streams(file_path)
|
||
args = ArgsBuffer()
|
||
args.input = file_path
|
||
tags = cmd_show_metadata(args)
|
||
|
||
return_value = True
|
||
|
||
# ---------- COUNT CHECKS ----------
|
||
if "video_count" in expected:
|
||
if len(video_streams) != expected["video_count"]:
|
||
logger.error(f"{red_tag("VIDEO COUNT FAIL")} expected {expected['video_count']}, got {len(video_streams)}")
|
||
return_value = return_value and False
|
||
|
||
if "audio_count" in expected:
|
||
if len(audio_streams) != expected["audio_count"]:
|
||
logger.error(f"{red_tag("AUDIO COUNT FAIL")} expected {expected['audio_count']}, got {len(audio_streams)}")
|
||
return_value = return_value and False
|
||
|
||
if "subtitle_count" in expected:
|
||
if len(subtitle_streams) != expected["subtitle_count"]:
|
||
logger.error(f"{red_tag("SUBTITLE COUNT FAIL")} expected {expected['subtitle_count']}, got {len(subtitle_streams)}")
|
||
return_value = return_value and False
|
||
|
||
# ---------- CODEC CHECKS ----------
|
||
if "audio_codecs" in expected:
|
||
actual = sorted([s.get("codec_name") for s in audio_streams])
|
||
expected_codecs = sorted(expected["audio_codecs"])
|
||
if len(expected_codecs) < len(actual):
|
||
last = expected_codecs[-1] if expected_codecs else None
|
||
expected_codecs = expected_codecs + [last] * (len(actual) - len(expected_codecs))
|
||
|
||
if actual != expected_codecs:
|
||
logger.error(f"{red_tag("AUDIO CODEC FAIL")}\nExpected: {expected_codecs}\nGot: {actual}")
|
||
return_value = return_value and False
|
||
|
||
if "video_codecs" in expected:
|
||
actual = sorted([s.get("codec_name") for s in video_streams])
|
||
expected_codecs = sorted(expected["video_codecs"])
|
||
if len(expected_codecs) < len(actual):
|
||
last = expected_codecs[-1] if expected_codecs else None
|
||
expected_codecs = expected_codecs + [last] * (len(actual) - len(expected_codecs))
|
||
if actual != expected_codecs:
|
||
logger.error(f"{red_tag("VIDEO CODEC FAIL")}\nExpected: {expected_codecs}\nGot: {actual}")
|
||
return_value = return_value and False
|
||
|
||
# ---------- DURATION CHECKS ----------
|
||
if "duration" in expected:
|
||
format_info = info.get("format", {})
|
||
actual_duration = float(format_info.get("duration", 0))
|
||
expected_duration = parse_time_to_seconds(expected["duration"])
|
||
|
||
# allow small ffmpeg timing variance (important!)
|
||
tolerance = expected.get("duration_tolerance", 0.2)
|
||
|
||
if abs(actual_duration - expected_duration) <= tolerance:
|
||
logger.error(f"{red_tag("DURATION FAIL")} expected ~{expected_duration} (±{tolerance}), got {actual_duration}s")
|
||
return_value = return_value and False
|
||
|
||
# ---------- CHAPTER CHECKS ----------
|
||
if "chapter_name" in expected:
|
||
chapter_name = [ chapter.get("tags", {}).get("title", "") for chapter in info.get("chapters", []) ]
|
||
expected_chapter_name = expected["chapter_name"]
|
||
if expected_chapter_name != chapter_name:
|
||
logger.error(f"Expected chapter_name: {expected_chapter_name}, got {chapter_name}")
|
||
return_value = return_value and False
|
||
|
||
if "chapter_count" in expected:
|
||
chapter_num = len(info.get("chapters", []))
|
||
expected_chapter_count = expected["chapter_count"]
|
||
if expected_chapter_count != chapter_num:
|
||
logger.error(f"Expected chapters: {expected_chapter_count}, got {chapter_num}")
|
||
return_value = return_value and False
|
||
|
||
# ---------- METADATA CHECKS ----------
|
||
|
||
if "metadata" in expected:
|
||
key = expected["metadata"].get("key", "")
|
||
value = expected["metadata"].get("value", "")
|
||
if key in tags.keys() and tags[key] == value:
|
||
logger.debug(f"{key}: {value}")
|
||
else:
|
||
logger.error(f"Key {key}: {value} not in expected tags")
|
||
return_value = return_value and False
|
||
|
||
if "no_russian_dir" in expected:
|
||
media_files = []
|
||
for ext in VIDEO_EXTS:
|
||
media_files.extend(list(file_path.rglob("*"+ext)))
|
||
|
||
for file in media_files:
|
||
audio_tracks = translate_audio_tracks(get_audio_tracks(file))
|
||
subtitle_tracks = translate_audio_tracks(get_subtitle_tracks(file))
|
||
|
||
logger.debug(f"Audio tracks: {audio_tracks}")
|
||
for audio_track in audio_tracks:
|
||
if is_russian_track(audio_track["description"]):
|
||
logger.error(f"Found russian track: {audio_track} in file {file}")
|
||
return return_value and False
|
||
|
||
#logger.debug(f"Subtitle tracks: {subtitle_tracks}")
|
||
#for subtitle_track in subtitle_tracks:
|
||
# if is_russian_track(audio_track["description"]):
|
||
# logger.error(f"Found russian track: {subtitle_track} in file {file}")
|
||
# return return_value and False
|
||
|
||
logger.info(f"{green_tag("PASS")} {file_path}")
|
||
return return_value
|
||
|
||
|
||
|
||
ret = True
|
||
# ---------- INFO ----------
|
||
#args = ArgsBuffer()
|
||
#args.input = test_file_mp4
|
||
#test_dict = {}
|
||
#run_and_probe(cmd["name"], cmd_info, args, test_dict)
|
||
#args.input = test_file_mkv
|
||
#ret = ret and run_and_probe(cmd["name"], cmd_info, args, test_dict)
|
||
|
||
# ---------- DOWNSIZE ----------
|
||
args_test = ArgsBuffer()
|
||
|
||
cmd = commands["downsize"]
|
||
if cmd["name"] in args.choices:
|
||
for test_file in test_files:
|
||
video_streams, audio_streams, subtitle_streams, cover_image = get_streams(test_file)
|
||
args_test.input = test_file
|
||
args_test.output = os.path.join(test_dir, "out_downsize" + os.path.splitext(test_file)[1])
|
||
args_test.width = 640
|
||
args_test.height = 360
|
||
args_test.crf = 28
|
||
test_dict = [{"video_count": len(video_streams), "audio_count": len(audio_streams),
|
||
"subtitle_count": len(subtitle_streams), "video_codecs": ["h264"]}]
|
||
ret = ret and run_and_probe(cmd["name"], cmd_downsize, args_test, test_dict)
|
||
|
||
# ---------- REMOVE AUDIO TRACK ----------
|
||
cmd = commands["remove-audio"]
|
||
if cmd["name"] in args.choices:
|
||
args_test = ArgsBuffer()
|
||
args_test.audio_index = 1
|
||
|
||
for test_file in test_files:
|
||
video_streams, audio_streams, subtitle_streams, cover_image = get_streams(test_file)
|
||
args_test.input = test_file
|
||
test_dict = [{"video_count": len(video_streams), "audio_count": 1, "subtitle_count": len(subtitle_streams)}]
|
||
args_test.output = os.path.join(test_dir, "out_remove_audio" + os.path.splitext(test_file)[1])
|
||
ret = ret and run_and_probe(cmd["name"], cmd_remove_audio_track, args_test, test_dict)
|
||
|
||
# ---------- EXTRACT AUDIO TRACK ----------
|
||
cmd = commands["extract-audio"]
|
||
if cmd["name"] in args.choices:
|
||
args_test = ArgsBuffer()
|
||
args_test.audio_index = 1
|
||
|
||
for test_file in test_files:
|
||
args_test.input = test_file
|
||
test_dict = [{"video_count": 0, "audio_count": 1, "subtitle_count": 0}]
|
||
args_test.output = os.path.join(test_dir, "extracted_audio_" + os.path.splitext(test_file)[1][1:] + ".mp3")
|
||
ret = ret and run_and_probe(cmd["name"], cmd_extract_audio_track, args_test, test_dict)
|
||
|
||
# ---------- CONVERT AV ----------
|
||
cmd = commands["convert-av"]
|
||
if cmd["name"] in args.choices:
|
||
args_test = ArgsBuffer()
|
||
for test_file in test_files:
|
||
video_streams, audio_streams, subtitle_streams, cover_image = get_streams(test_file)
|
||
args_test.input = test_file
|
||
args_test.output = os.path.join(test_dir, "out_convert_av" + os.path.splitext(test_file)[1])
|
||
args_test.video_codec = "libx264"
|
||
args_test.audio_codec = "aac"
|
||
args_test.preset = "fast"
|
||
args_test.quality = 23
|
||
test_dict = [ {"video_count": len(video_streams),"audio_count": len(audio_streams),
|
||
"subtitle_count": len(subtitle_streams), "video_codecs": ["h264"],"audio_codecs": ["aac"]}]
|
||
ret = ret and run_and_probe(cmd["name"], cmd_convert_av, args_test, test_dict)
|
||
|
||
# ---------- CONVERT AUDIO ----------
|
||
cmd = commands["convert-audio"]
|
||
|
||
if cmd["name"] in args.choices:
|
||
args_test = ArgsBuffer()
|
||
for test_file in test_files:
|
||
video_streams, audio_streams, subtitle_streams, cover_image = get_streams(test_file)
|
||
args_test.input = test_file
|
||
args_test.output = os.path.join(test_dir, "out_convert_audio" + os.path.splitext(test_file)[1])
|
||
args_test.audio_codec = "aac"
|
||
test_dict = [ {"video_count": len(video_streams), "audio_count": len(audio_streams),
|
||
"subtitle_count": len(subtitle_streams), "audio_codecs": ["aac"]}]
|
||
ret = ret and run_and_probe(cmd["name"], cmd_convert_audio, args_test, test_dict)
|
||
|
||
# ---------- CONVERT VIDEO ----------
|
||
cmd = commands["convert-video"]
|
||
if cmd["name"] in args.choices:
|
||
args_test = ArgsBuffer()
|
||
for test_file in test_files:
|
||
video_streams, audio_streams, subtitle_streams, cover_image = get_streams(test_file)
|
||
args_test.input = test_file
|
||
args_test.output = os.path.join(test_dir, "out_convert_video" + os.path.splitext(test_file)[1])
|
||
args_test.video_codec = "libx264"
|
||
test_dict = [ {"video_count": len(video_streams), "audio_count": len(audio_streams),
|
||
"subtitle_count": len(subtitle_streams), "video_codecs": ["h264"]}]
|
||
ret = ret and run_and_probe(cmd["name"], cmd_convert_video, args_test, test_dict)
|
||
|
||
# ---------- REMOVE SUBS ----------
|
||
cmd = commands["remove-subs"]
|
||
if cmd["name"] in args.choices:
|
||
args_test = ArgsBuffer()
|
||
for test_file in test_files:
|
||
video_streams, audio_streams, subtitle_streams, cover_image = get_streams(test_file)
|
||
args_test.input = test_file
|
||
test_dict = [ {"video_count": len(video_streams), "audio_count": len(audio_streams), "subtitle_count": 0}]
|
||
args_test.output = os.path.join(test_dir, "out_no_subs" + os.path.splitext(test_file)[1])
|
||
ret = ret and run_and_probe(cmd["name"], cmd_remove_subs, args_test, test_dict)
|
||
|
||
# ---------- FILTER AUDIO ----------
|
||
cmd = commands["filter-audio"]
|
||
if cmd["name"] in args.choices:
|
||
args_test = ArgsBuffer()
|
||
for test_file in test_files:
|
||
video_streams, audio_streams, subtitle_streams, cover_image = get_streams(test_file)
|
||
args_test.input = test_file
|
||
args_test.output = os.path.join(test_dir, "out_filter_audio" + os.path.splitext(test_file)[1])
|
||
args_test.audio_index = 1
|
||
test_dict = [ {"video_count": len(video_streams), "audio_count": len(audio_streams)+1,
|
||
"subtitle_count": len(subtitle_streams), "audio_codecs": ["vorbis"]}]
|
||
ret = ret and run_and_probe(cmd["name"], cmd_filter_audio, args_test, test_dict)
|
||
|
||
# ---------- CUT ----------
|
||
cmd = commands["cut"]
|
||
if cmd["name"] in args.choices:
|
||
args_test = ArgsBuffer()
|
||
for test_file in test_files:
|
||
video_streams, audio_streams, subtitle_streams, cover_image = get_streams(test_file)
|
||
args_test.input = test_file
|
||
args_test.output = os.path.join(test_dir, "out_cut" + os.path.splitext(test_file)[1])
|
||
args_test.start = "0:0:5"
|
||
args_test.duration = "10s"
|
||
test_dict = [ {"video_count": len(video_streams), "audio_count": len(audio_streams),
|
||
"subtitle_count": len(subtitle_streams), "duration": "10s"}]
|
||
ret = ret and run_and_probe(cmd["name"], cmd_cut_segment, args_test, test_dict)
|
||
|
||
# ---------- SET PRIMARY AUDIO ----------
|
||
cmd = commands["set-primary-audio"]
|
||
if cmd["name"] in args.choices:
|
||
args_test = ArgsBuffer()
|
||
for test_file in test_files:
|
||
video_streams, audio_streams, subtitle_streams, cover_image = get_streams(test_file)
|
||
args_test.input = test_file
|
||
args_test.output = os.path.join(test_dir, "out_primary_audio" + os.path.splitext(test_file)[1])
|
||
args_test.audio_index = 2
|
||
test_dict = [ {"video_count": len(video_streams), "audio_count": len(audio_streams),
|
||
"subtitle_count": len(subtitle_streams)}]
|
||
ret = ret and run_and_probe(cmd["name"], cmd_set_primary_audio, args_test, test_dict)
|
||
|
||
# ---------- APPLY CHAPTERS-----
|
||
cmd = commands["chapters"]
|
||
if cmd["name"] in args.choices:
|
||
args_test = ArgsBuffer()
|
||
test_file = test_file_dir / "test_auto.m4b"
|
||
video_streams, audio_streams, subtitle_streams, cover_image = get_streams(test_file)
|
||
args_test.input = test_file
|
||
args_test.output = Path(test_dir) / f"out_chapters{Path(test_file).suffix}"
|
||
args_test.chapter_file = test_file_dir / "test_auto_chapters.txt"
|
||
lines = Path(args_test.chapter_file).read_text(encoding="utf-8").splitlines()
|
||
test_dict = [{"video_count": len(video_streams), "audio_count": len(audio_streams),
|
||
"chapter_name": lines, "chapter_count": len(lines) }]
|
||
ret = ret and run_and_probe(cmd["name"], cmd_apply_chapters, args_test, test_dict)
|
||
|
||
|
||
# ---------- WRITE METADATA-----
|
||
cmd = commands["metadata-set"]
|
||
if cmd["name"] in args.choices:
|
||
args_test = ArgsBuffer()
|
||
test_file = test_file_dir / "test_auto.m4b"
|
||
video_streams, audio_streams, subtitle_streams, cover_image = get_streams(test_file)
|
||
args_test.input = test_file
|
||
args_test.output = Path(test_dir) / f"out_metadata{Path(test_file).suffix}"
|
||
args_test.key = "artist"
|
||
args_test.value = "TEST"
|
||
args_test.kidflag = False
|
||
test_dict = [{"video_count": len(video_streams), "audio_count": len(audio_streams),
|
||
"metadata":{"key":"TEST"}} ]
|
||
ret = ret and run_and_probe(cmd["name"], cmd_set_metadata, args_test, test_dict)
|
||
|
||
# ---------- REMOVE RUSSIAN AUDIO -----
|
||
cmd = commands["remove-russian-from-db"]
|
||
if cmd["name"] in args.choices:
|
||
args_test = ArgsBuffer()
|
||
args_test.input = test_dir / "out_remove_russian"
|
||
args_test.output = test_dir / "out_remove_russian"
|
||
args_test.loud = False
|
||
shutil.copytree(
|
||
Path(test_file_dir),
|
||
Path(args_test.input),
|
||
dirs_exist_ok=True,
|
||
)
|
||
test_dict = [{"no_russian_dir": test_dir} ]
|
||
ret = ret and run_and_probe(cmd["name"], cmd_remove_russian_from_db, args_test, test_dict)
|
||
|
||
if ret:
|
||
logger.info(green_tag("ALL TESTS SUCCEEDED"))
|
||
else:
|
||
logger.info(red_tag("SOME TESTS FAILED"))
|
||
|
||
# ---------- CLI Setup ---------- #
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(
|
||
description="Media processing tool using ffmpeg-python (CLI + interactive fallback)",
|
||
formatter_class=argparse.ArgumentDefaultsHelpFormatter
|
||
)
|
||
|
||
parser.add_argument("-ll","--log-level",type=str.upper,default="INFO",
|
||
choices=["TRACE", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
|
||
help="Set log level",
|
||
)
|
||
parser.add_argument(
|
||
"--dry-run",
|
||
action="store_true",
|
||
help="Dry run only",
|
||
)
|
||
# ----------------------------
|
||
# BASE ARGS (shared)
|
||
# ----------------------------
|
||
base_p = argparse.ArgumentParser(add_help=False)
|
||
base_p.add_argument("input", nargs="?", help="Path to input media file")
|
||
base_p.add_argument("-o", "--output", help="Path for output file")
|
||
base_p.add_argument("--override", action="store_true", help="Override input")
|
||
base_p.add_argument("--backup", action="store_true", help="Backup original file when overriding")
|
||
base_p.add_argument("--loud", action="store_true", help="Run ffmpeg loud")
|
||
base_p.add_argument("--kidflag", action="store_true", help="Use kid3 instead of ffmpeg")
|
||
sub = parser.add_subparsers(dest="command") # command is optional now
|
||
|
||
# ----------------------------
|
||
# INFO
|
||
# ----------------------------
|
||
cmd=commands["info"]
|
||
p_info = sub.add_parser(cmd["name"], aliases=["i"], parents=[base_p], help="Inspect media metadata")
|
||
p_info.set_defaults(func=cmd_info, command=cmd["name"])
|
||
|
||
# TEST
|
||
cmd=commands["test"]
|
||
p_test = sub.add_parser(cmd["name"], aliases=["t"], parents=[base_p], help="Run test function")
|
||
|
||
p_test.add_argument("--choices", nargs="+",choices=list(commands.keys()),default=list(commands.keys()),
|
||
help="Run a predefined test")
|
||
p_test.set_defaults(func=cmd_test, command=cmd["name"])
|
||
|
||
# ----------------------------
|
||
# CONVERT AV
|
||
# ----------------------------
|
||
cmd=commands["convert-av"]
|
||
p_av = sub.add_parser(cmd["name"], aliases=["cav"], parents=[base_p], help="Full re-encode of video/audio")
|
||
p_av.add_argument("--video-codec", default="libx264",
|
||
help="Video codec to use (default: libx264, e.g. libx265, vp9)")
|
||
p_av.add_argument("--audio-codec", default="aac", help="Audio codec to use (default: aac, e.g. mp3, opus)")
|
||
p_av.add_argument("--preset", default="medium",
|
||
help="Encoding speed/quality preset (default: medium, e.g. ultrafast, fast, slow, veryslow)")
|
||
p_av.add_argument("--quality", type=int,
|
||
help="Output quality level (e.g. CRF value or bitrate depending on encoder)")
|
||
p_av.set_defaults(func=cmd_convert_av, command=cmd["name"])
|
||
|
||
# ----------------------------
|
||
# CONVERT AUDIO
|
||
# ----------------------------
|
||
cmd=commands["convert-audio"]
|
||
p_a = sub.add_parser(cmd["name"], aliases=["ca"], parents=[base_p], help="Change audio format only")
|
||
p_a.add_argument("--audio-codec", default="aac", help="Audio codec to use (default: aac, e.g. mp3, opus)")
|
||
p_a.set_defaults(func=cmd_convert_audio, command=cmd["name"])
|
||
|
||
# ----------------------------
|
||
# CONVERT VIDEO
|
||
# ----------------------------
|
||
cmd=commands["convert-video"]
|
||
p_v = sub.add_parser(cmd["name"], aliases=["cv"], parents=[base_p], help="Change video format only")
|
||
p_v.add_argument("--video-codec", default="libx264",
|
||
help="Video codec to use (default: libx264, e.g. libx265, vp9)")
|
||
p_v.set_defaults(func=cmd_convert_video, command=cmd["name"])
|
||
|
||
# ----------------------------
|
||
# REMOVE SUBS
|
||
# ----------------------------
|
||
cmd=commands["remove-subs"]
|
||
p_rs = sub.add_parser(cmd["name"], aliases=["rs"], parents=[base_p], help="Strip all subtitle streams (-sn)")
|
||
p_rs.set_defaults(func=cmd_remove_subs, command=cmd["name"])
|
||
|
||
# ----------------------------
|
||
# REMOVE AUDIO TRACK
|
||
# ----------------------------
|
||
cmd=commands["remove-audio"]
|
||
p_ra = sub.add_parser(cmd["name"], aliases=["ra"], parents=[base_p], help="Remove one audio track")
|
||
p_ra.add_argument("-a", "--audio-index", type=int)
|
||
p_ra.set_defaults(func=cmd_remove_audio_track, command=cmd["name"])
|
||
|
||
# ----------------------------
|
||
# EXTRACT AUDIO TRACK
|
||
# ----------------------------
|
||
cmd=commands["extract-audio"]
|
||
p_ea = sub.add_parser(cmd["name"], aliases=["ea"], parents=[base_p], help="Extract one audio track")
|
||
p_ea.add_argument("-a", "--audio-index", type=int)
|
||
p_ea.set_defaults(func=cmd_extract_audio_track, command=cmd["name"])
|
||
|
||
# ----------------------------
|
||
# DOWNSIZE
|
||
# ----------------------------
|
||
cmd=commands["downsize"]
|
||
p_ds = sub.add_parser(cmd["name"], aliases=["d"], parents=[base_p], help="Scale video dimensions (-vf scale)")
|
||
p_ds.add_argument("-w", "--width", type=int, help="width: eg 1080")
|
||
p_ds.add_argument("-he", "--height", type=int, help="height: eg 720")
|
||
p_ds.add_argument("-c" , "--crf", type=int, help="crf: eg 23")
|
||
p_ds.set_defaults(func=cmd_downsize, command=cmd["name"])
|
||
|
||
# ----------------------------
|
||
# FILTER AUDIO
|
||
# ----------------------------
|
||
cmd=commands["filter-audio"]
|
||
p_fa = sub.add_parser(cmd["name"], aliases=["fa"], parents=[base_p], help="Apply audio enhancement chain")
|
||
p_fa.add_argument("-a", "--audio-index", type=int)
|
||
p_fa.set_defaults(func=cmd_filter_audio, command=cmd["name"])
|
||
|
||
# ----------------------------
|
||
# CUT
|
||
# ----------------------------
|
||
cmd=commands["cut"]
|
||
p_c = sub.add_parser(cmd["name"], aliases=["c"], parents=[base_p], help="Cut X seconds out of video")
|
||
p_c.add_argument("--start")
|
||
p_c.add_argument("--duration")
|
||
p_c.set_defaults(func=cmd_cut_segment, command=cmd["name"])
|
||
|
||
# ----------------------------
|
||
# SET PRIMARY AUDIO
|
||
# ----------------------------
|
||
cmd=commands["set-primary-audio"]
|
||
p_pa = sub.add_parser(cmd["name"], aliases=["spa"], parents=[base_p], help="Set primary audio track")
|
||
p_pa.add_argument("-a", "--audio-index", type=int)
|
||
p_pa.set_defaults(func=cmd_set_primary_audio, command=cmd["name"])
|
||
|
||
# ----------------------------
|
||
# REMOVE ALL RUSSIAN AUDIO
|
||
# ----------------------------
|
||
cmd = commands["remove-russian-from-db"]
|
||
p_ch = sub.add_parser(cmd["name"], aliases=["rrfd"], parents=[base_p],
|
||
help="Remove all russian audio tracks from media files in path")
|
||
p_ch.set_defaults(func=cmd_remove_russian_from_db, command=cmd["name"])
|
||
|
||
# ----------------------------
|
||
# APPLY CHAPTERS
|
||
# ----------------------------
|
||
cmd=commands["chapters"]
|
||
p_ch = sub.add_parser(cmd["name"],aliases=["ch"],parents=[base_p],help="Apply chapter metadata from ffmetadata file")
|
||
p_ch.add_argument("-cf","--chapter-file",help="Path to ffmetadata chapters file (e.g. chapters.txt)")
|
||
p_ch.set_defaults(func=cmd_apply_chapters, command=cmd["name"])
|
||
|
||
# ----------------------------
|
||
# METADATA
|
||
# ----------------------------
|
||
|
||
# PRINT METADATA
|
||
cmd=commands["metadata-print"]
|
||
p_ch = sub.add_parser(cmd["name"], aliases=["mp"], parents=[base_p],
|
||
help="Show metadata")
|
||
p_ch.set_defaults(func=cmd_show_metadata, command=cmd["name"])
|
||
|
||
# SET METADATA
|
||
cmd = commands["metadata-set"]
|
||
p_ch = sub.add_parser(cmd["name"],aliases=["ms"],parents=[base_p],help="Apply metadata key and value to input file")
|
||
p_ch.add_argument("-k","--key",help="Metadata key name)")
|
||
p_ch.add_argument("-v","--value",help="Metadata key value")
|
||
p_ch.set_defaults(func=cmd_set_metadata, command=cmd["name"])
|
||
|
||
# =========================================================
|
||
# PARSE
|
||
# =========================================================
|
||
args = parser.parse_args()
|
||
|
||
logging.basicConfig(
|
||
level=LOG_LEVELS[args.log_level.upper()],
|
||
format="%(asctime)s %(levelname)s %(message)s",
|
||
)
|
||
# ----------------------------
|
||
# COMMAND VALIDATION
|
||
# ----------------------------
|
||
if not hasattr(args, "func"):
|
||
logger.info(f"\nAvailable commands:")
|
||
|
||
seen = set()
|
||
for name, parser in sub.choices.items():
|
||
if id(parser) in seen:
|
||
continue
|
||
seen.add(id(parser))
|
||
logger.info(f" - {name}")
|
||
|
||
args.command = ensure(None, "Command")
|
||
|
||
if args.command not in sub.choices:
|
||
parser.error("Unknown command")
|
||
|
||
# re-parse subcommand properly
|
||
sub_args = sub.choices[args.command].parse_args([])
|
||
args = argparse.Namespace( **vars(sub_args))
|
||
|
||
# ----------------------------
|
||
# INTERACTIVE FALLBACK (ensure ALL missing values)
|
||
# ----------------------------
|
||
if args.command != "test":
|
||
args.input = ensure(getattr(args, "input", None), f"Input file ({Path.cwd()})")
|
||
args.input = Path(args.input).expanduser().resolve()
|
||
if not Path(args.input).exists():
|
||
parser.error(f"Input file {args.input} does not exist!")
|
||
if commands[args.command]["output"] and not getattr(args, "output", None):
|
||
input_path = Path(args.input)
|
||
if args.override:
|
||
args.output = str(input_path.with_name(f"{input_path.stem}_override{input_path.suffix}"))
|
||
else:
|
||
args.output = ensure(getattr(args, "output", None), "Output file (enter for default file)")
|
||
if args.output == "" or args.output is None:
|
||
input_path = Path(args.input)
|
||
args.output = str(input_path.with_name(f"{input_path.stem}_{args.command}{input_path.suffix}"))
|
||
# ----------------------------
|
||
# COMMAND-SPECIFIC REQUIRED FIELDS
|
||
# ----------------------------
|
||
|
||
if args.command == "convert-av":
|
||
args.preset = ensure(getattr(args, "preset", None),
|
||
"Encoding speed/quality preset (default: medium, e.g. ultrafast, fast, slow, veryslow)")
|
||
args.quality = ensure(getattr(args, "quality", None),
|
||
"Output quality level (e.g. CRF value or bitrate depending on encoder)")
|
||
if args.command == "convert-av" or args.command == "convert-audio":
|
||
args.audio_codec = ensure(getattr(args, "audio_codec", None),
|
||
"Audio codec to use (default: aac, e.g. mp3, opus)")
|
||
if args.command == "convert-av" or args.command == "convert-video":
|
||
args.video_codec = ensure(getattr(args, "video_codec", None),
|
||
"Video codec to use (default: libx264, e.g. libx265, vp9)")
|
||
|
||
if args.command == "cut":
|
||
args.start = ensure(getattr(args, "start", None), "Start time (e.g. 10s, 1:00)")
|
||
args.duration = ensure(getattr(args, "duration", None), "Duration (e.g. 30s)")
|
||
|
||
if args.command == "downsize":
|
||
args.width = ensure(getattr(args, "width", None), "width: eg 1080")
|
||
args.height = ensure(getattr(args, "height", None), "height: eg 720")
|
||
args.crf = ensure(getattr(args, "crf", None), "crf: eg 23")
|
||
|
||
if args.command in ["filter-audio", "remove-audio", "extract-audio", "set-primary-audio"]:
|
||
args.audio_index = ensure(getattr(args, "audio_index", None), "Audio index")
|
||
if args.audio_index is None:
|
||
args.audio_index = 1
|
||
|
||
if args.command == "chapters":
|
||
args.chapter_file = ensure(getattr(args, "chapter_file", None), "File with chapternames")
|
||
|
||
if args.command == "metadata-set":
|
||
args.chapter_file = ensure(getattr(args, "key", None), "Key to change")
|
||
args.chapter_file = ensure(getattr(args, "value", None), "Key value")
|
||
|
||
# ----------------------------
|
||
# EXECUTE COMMAND
|
||
# ----------------------------
|
||
try:
|
||
if args.command == "test":
|
||
args.func(args)
|
||
sys.exit(0)
|
||
outputs = []
|
||
input_path = Path(args.input)
|
||
# ---- DIRECTORY ----
|
||
if input_path.is_dir():
|
||
counter=0
|
||
sorted_files = sorted(input_path.iterdir(), key=natural_sort_key)
|
||
for file in sorted_files:
|
||
if (file.suffix.lower() not in VIDEO_EXTS or file.suffix.lower() not in AUDIO_EXTS):
|
||
continue
|
||
logger.info(f"Processing File: {file}")
|
||
args.input = str(file)
|
||
if args.override:
|
||
out_path = file.with_name(f"{file.stem}_override{file.suffix}")
|
||
else:
|
||
out_path = file.with_name(f"{file.stem}_{args.command}{file.suffix}")
|
||
args.output = str(out_path)
|
||
args.file_counter = counter
|
||
out = args.func(args)
|
||
if out != "No output needed":
|
||
outputs.append(out)
|
||
counter+=1
|
||
# ---- SINGLE FILE ----
|
||
elif input_path.is_file():
|
||
out = args.func(args)
|
||
if out != "No output needed":
|
||
outputs.append(out)
|
||
else:
|
||
raise ValueError(f"Invalid input path: {args.input}")
|
||
|
||
if any(item is None for item in outputs) or len(outputs) == 0:
|
||
logger.info(f"Output is NONE")
|
||
else:
|
||
if not isinstance(outputs, list):
|
||
outputs = [outputs]
|
||
|
||
if args.command != "test" and outputs is not None:
|
||
run_ffmpeg(
|
||
ffmpeg.merge_outputs(*outputs),
|
||
overwrite=True,
|
||
loud=args.loud
|
||
)
|
||
if args.override:
|
||
move_override_file(input_path, args.backup,args.loud)
|
||
except ffmpeg.Error as e:
|
||
logger.error(f"FFmpeg failed.\n")
|
||
if e.stderr:
|
||
logger.error(e.stderr.decode(errors="ignore"))
|
||
sys.exit(1)
|
||
|
||
sys.exit(0)
|
||
|
||
if __name__ == "__main__":
|
||
main()
|