From c4a717a8a8baa62f6eb4e0a2e60ed48995274d61 Mon Sep 17 00:00:00 2001 From: Lucas Vivo Date: Tue, 27 Oct 2026 09:48:06 +0100 Subject: [PATCH] feat(ffmpeg_handling): remove subs with remove_russian 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 --- media_editing/ffmpeg_handling.py | 240 ++++++++++++++++--------------- 1 file changed, 121 insertions(+), 119 deletions(-) diff --git a/media_editing/ffmpeg_handling.py b/media_editing/ffmpeg_handling.py index 4500f53..1cc2060 100755 --- a/media_editing/ffmpeg_handling.py +++ b/media_editing/ffmpeg_handling.py @@ -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 )