lucas/Server-scripts
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ffmpeg_handling): remove subs with remove_russian

Browse Source 
add remove stream function
change remove audio and remove subs to use this function
fix: show_metadata returns default values and return_tags is given as
parameter
This commit is contained in:
Lucas Vivo committed 2026-10-27 09:48:06 +01:00
1 parent 7f2048cda6
commit c4a717a8a8
1 file changed
+121 -119
Download Patch File Download Diff File Expand all files Collapse all files
media_editing/ffmpeg_handling.py
+121 -119
View File
@@ -520,91 +520,31 @@ def cmd_convert_video(args):
)
return True, out
def cmd_remove_subs(args):
"""
Strips all subtitle data.
FFmpeg Equivalent:
ffmpeg -i [input] -sn -c copy [output]
Parameters:
- sn: Disables subtitle recording.
"""
from collections.abc import Callable
def remove_streams(
args,
remove_indices: set[int] | list[int],
*,
output_subtitle_codec: bool = False,
) -> tuple[bool, str]:
probe = ffmpeg.probe(str(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))
if args.dry_run:
logger.debug(f"Dry run: {' '.join(cmd)}")
return True, "nooutputneeded"
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 True, "nooutputneeded"
def cmd_remove_audio_track(args):
"""
1. Removes a specific audio track from the video (Main Output).
"""
probe = ffmpeg.probe(str(args.input))
remove_indices = []
for audio_index in args.audio_index:
remove_indices.append(select_audio_stream(args.input, audio_index))
remove_indices = set(remove_indices)
remaining_audio = [
s
for s in probe["streams"]
if s["codec_type"] == "audio" and s["index"] not in remove_indices
if s["codec_type"] == "audio"
and s["index"] not in remove_indices
]
if not remaining_audio:
logger.error("No audiotracks remaining")
return False, "nooutputneeded"
remaining_audio = translate_audio_tracks(remaining_audio)
removed_was_default = any(
s["index"] in remove_indices and s.get("disposition", {}).get("default", 0) == 1
if not remaining_audio:
logger.warning("No audiotracks remaining")
return False, "nooutputneeded"
removed_default_audio = any(
s["codec_type"] == "audio"
and s["index"] in remove_indices
and s.get("disposition", {}).get("default", 0) == 1
for s in probe["streams"]
)
@@ -625,13 +565,22 @@ def cmd_remove_audio_track(args):
"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
if output_subtitle_codec:
cmd.extend(["-c:s", get_scodec(args.output)])
else:
cmd.extend(["-c:s", "copy"])
cmd.extend(
[
"-map_metadata",
"0",
]
)
if removed_default_audio:
cmd.extend(
[
"-disposition:a",
@@ -642,10 +591,13 @@ def cmd_remove_audio_track(args):
)
cmd.append(str(args.output))
logger.trace(" ".join(cmd))
if args.dry_run:
logger.debug(f"Dry run: {' '.join(cmd)}")
logger.debug("Dry run: %s", " ".join(cmd))
return True, "nooutputneeded"
try:
result = subprocess.run(
cmd,
@@ -653,20 +605,47 @@ def cmd_remove_audio_track(args):
text=True,
check=True,
)
logger.debug("FFmpeg subprocess successful")
logger.trace(result.stdout)
return True, "nooutputneeded"
except subprocess.CalledProcessError as e:
logger.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
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
def cmd_remove_subs(args):
probe = ffmpeg.probe(str(args.input))
if args.num is None:
remove_indices = {
s["index"]
for s in probe["streams"]
if s["codec_type"] == "subtitle"
}
else:
remove_indices = set(args.num)
return remove_streams(
args=args,
remove_indices=remove_indices,
)
def cmd_remove_audio_track(args):
remove_indices = {
select_audio_stream(args.input, audio_index)
for audio_index in args.audio_index
}
return remove_streams(
args=args,
remove_indices=remove_indices,
output_subtitle_codec=True,
)
def cmd_extract_audio_track(args):
"""
1. Saves that specific audio track as a standalone MP3 (Extra Output).
@@ -950,20 +929,20 @@ def cmd_set_primary_audio(args):
return True, out
def cmd_show_metadata(args, show_metadata=False):
"""
Prints metadata directly from ffprobe output.
"""
def cmd_show_metadata(args, show_metadata=False, return_tags=None):
probe = ffmpeg.probe(str(args.input))
format_tags = probe.get("format", {}).get("tags", {})
if return_tags is not None:
return_tags.clear()
return_tags.update(format_tags)
if show_metadata:
logger.info("=== FORMAT METADATA ===")
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')}) ---")
@@ -974,7 +953,7 @@ def cmd_show_metadata(args, show_metadata=False):
# else:
# for k, v in tags.items():
# logger.info(f"{k}: {v}")
return False, format_tags
return False, "nooutputneeded"
def cmd_set_metadata(args):
@@ -1121,34 +1100,55 @@ def cmd_apply_chapters(args):
return True, "nooutputneeded"
def cmd_remove_russian_from_db(args_in):
def cmd_remove_russian_from_db(args):
"""
Find all media with russian audio
remove russian audio tracks
"""
file = args_in.input
file = args.input
logger.trace("Run remove russian cmd")
try:
audio_tracks = translate_audio_tracks(get_audio_tracks(file))
subtitle_tracks = translate_audio_tracks(get_subtitle_tracks(file))
except ValueError as e:
logger.warning("%s", e)
return False, "nooutputneeded"
logger.trace(f"Remove Russian from File: {file.name}")
logger.trace(f"With Audio tracks: {audio_tracks}")
remove_tracks = []
logger.trace("Remove Russian from File: %s", file.name)
remove_indices = {
track["number"]
for track in audio_tracks + subtitle_tracks
if is_russian_track(track)
}
for track in audio_tracks:
if is_russian_track(track):
remove_tracks.append(track["number"])
logger.debug(f"Removing russian track {track['number']} from {file}")
args = args_in
args.audio_index = remove_tracks
if remove_tracks:
cmd_remove_audio_track(args)
return True, "nooutputneeded"
else:
if track["number"] in remove_indices:
logger.debug(
"Removing russian audio track %s from %s",
track["number"],
file,
)
for track in subtitle_tracks:
if track["number"] in remove_indices:
logger.debug(
"Removing russian subtitle track %s from %s",
track["number"],
file,
)
if not remove_indices:
return False, "nooutputneeded"
return remove_streams(
args=args,
remove_indices=remove_indices,
)
def cmd_test(args):
"""
@@ -1289,7 +1289,8 @@ def cmd_test(args):
)
args = ArgsBuffer()
args.input = file_path
tags = cmd_show_metadata(args)
tags = {}
cmd_show_metadata(args,return_tags=tags)
return_value = True
@@ -1398,17 +1399,18 @@ def cmd_test(args):
audio_tracks = translate_audio_tracks(get_audio_tracks(file_path))
logger.debug(f"Audio tracks: {audio_tracks}")
for audio_track in audio_tracks:
if is_russian_track(audio_track["description"]):
if is_russian_track(audio_track):
logger.error(
f"Found russian track: {audio_track} in file {file_path}"
)
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
subtitle_tracks = translate_audio_tracks(get_audio_tracks(file_path))
logger.debug(f"Subtitle tracks: {subtitle_tracks}")
for subtitle_track in subtitle_tracks:
if is_russian_track(subtitle_track):
logger.error(f"Found russian track: {subtitle_track} in file {file_path}")
return return_value and False
logger.info(f"{green_tag('PASS')} {file_path}")
return return_value
@@ -1471,7 +1473,7 @@ def cmd_test(args):
Path(test_dir)
/ f"out_{str(cmd['name']).replace('-', '_')}_{Path(test_file).stem}{Path(test_file).suffix}"
)
ret = ret and run_and_probe(
cmd["name"], cmd_remove_audio_track, args_test, test_dict
)