From 2ea486a8c0686de9bbf11c4fc69cee6d51bed143 Mon Sep 17 00:00:00 2001 From: Lucas Vivo Date: Sun, 25 Oct 2026 18:32:51 +0100 Subject: [PATCH] ffmpeg handling: change directory handling for remove russian audio change directory handling for remove-russian to default use Path everywhere transfer LOG vars to serverlibrary --- media_editing/ffmpeg_handling.py | 370 +++++++++++++++---------------- serverlibrary/__init__.py | 31 +++ serverlibrary/core.py | 25 +++ 3 files changed, 234 insertions(+), 192 deletions(-) diff --git a/media_editing/ffmpeg_handling.py b/media_editing/ffmpeg_handling.py index df37a43..f56d319 100755 --- a/media_editing/ffmpeg_handling.py +++ b/media_editing/ffmpeg_handling.py @@ -13,32 +13,14 @@ 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) - +from serverlibrary import * 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"} @@ -86,22 +68,8 @@ AUDIO_TRANSLATIONS = { "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")", + r"\brus\b|\bru\b|russian|dubbed|рус|русский|русская|дубляж|" +r"оригинал|многоголос|двухголос|одноголос|авторский|любительский", re.IGNORECASE, ) @@ -111,12 +79,22 @@ def is_russian_track(description: str) -> bool: class ArgsBuffer: input: Path | None = None output: Path | None = None - kidflag: bool = False - loud: bool = False + kidflag: bool | None = False + loud: bool | None = False key: str | None = None value: str | None = None - audio_index: str | None = None + audio_index: int | None = None dry_run: bool | None = None + audio_codec: str | None = None + video_codec: str | None = None + preset: str | None = None + quality: int | None = None + width: int | None = None + height: int | None = None + crf: int | None = None + num: int | None = None + start: str | None = None + duration: str | None = None def translate_audio_description(description: str) -> str: for src, dst in AUDIO_TRANSLATIONS.items(): @@ -314,8 +292,8 @@ def parse_time_to_seconds(value: str) -> float: except ValueError: raise argparse.ArgumentTypeError(f"Invalid time format: {value}") -def get_scodec(output_path: str | Path): - ext = str(output_path).lower().split(".")[-1] +def get_scodec(output_path: Path): + ext = output_path.suffix if ext == "mp4": return "mov_text" @@ -325,9 +303,8 @@ def get_scodec(output_path: str | Path): return "copy" -def get_vcodec(output_path: str): - ext = output_path.lower().split(".")[-1] - +def get_vcodec(output_path: Path): + ext = output_path.suffix if ext == "mp4": return "libx264" @@ -402,7 +379,7 @@ def cmd_convert_av(args): inp = ffmpeg.input(args.input) out = ffmpeg.output( inp, - args.output, + str(args.output), vcodec=args.video_codec, acodec=args.audio_codec, scodec=get_scodec(args.output), @@ -428,7 +405,7 @@ def cmd_convert_audio(args): inp = ffmpeg.input(args.input) out = ffmpeg.output( inp, - args.output, + str(args.output), vcodec="copy", acodec=args.audio_codec, scodec=get_scodec(args.output), @@ -447,7 +424,7 @@ def cmd_convert_video(args): inp = ffmpeg.input(args.input) out = ffmpeg.output( inp, - args.output, + str(args.output), vcodec=args.video_codec, acodec="copy", scodec=get_scodec(args.output), @@ -637,7 +614,7 @@ def cmd_downsize(args): video, inp.audio, *subs, - args.output, + str(args.output), vcodec="libx264", crf=args.crf, acodec="copy", @@ -753,7 +730,7 @@ def cmd_filter_audio(args): out = ffmpeg.output( *input_args, - args.output, + str(args.output), vcodec="copy", acodec="libvorbis", scodec=get_scodec(args.output) @@ -783,7 +760,7 @@ def cmd_cut_segment(args): out = ffmpeg.output( inp, - args.output, + str(args.output), t=duration, map="0", vcodec=get_vcodec(args.output), @@ -846,7 +823,7 @@ def cmd_set_primary_audio(args): inp.video, *audio_inputs, *subs, - args.output, + str(args.output), vcodec="copy", acodec="copy", scodec=get_scodec(args.output), @@ -855,17 +832,18 @@ def cmd_set_primary_audio(args): return out -def cmd_show_metadata(args): +def cmd_show_metadata(args,show_metadata=False): """ 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}") + + if show_metadata: + logger.info(f"=== 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", [])): @@ -888,7 +866,7 @@ def cmd_set_metadata(args): - If file is .opus: uses kid3-cli (Vorbis tags) """ - ext = os.path.splitext(args.input)[1].lower() + ext = args.input.suffix.lower() # ───────────────────────────────────────────── # OPUS → use Kid3 CLI (Vorbis comments) @@ -1019,27 +997,23 @@ 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)) + file = args_in.input + args = ArgsBuffer() - return "No output needed" + 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.input = file + args.dry_run = args_in.dry_run + args.audio_index = number + args.output = args_in.output + + return cmd_remove_audio_track(args) def cmd_test(args): @@ -1054,22 +1028,19 @@ def cmd_test(args): 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}" + return f"{red}[{tag}]{reset}" def green_tag(tag: str) -> str: - return f"{GREEN}[{tag}]{RESET}" + return f"{green}[{tag}]{reset}" - def run_and_probe(name, cmd_func, args, test_dict: list=[]): + def run_and_probe(name, cmd_func, args_probe, _test_dict=None): + if _test_dict is None: + _test_dict = [] logger.info(f"TEST: {name}") # Run FFmpeg for this single command only - outputs = cmd_func(args) + outputs = cmd_func(args_probe) if not outputs: logger.error(f"No outputs returned.") return False @@ -1090,36 +1061,56 @@ def cmd_test(args): # Probe result file(s) out_files = [] - if hasattr(args, "output") and args.output: - out_files.append(args.output) + if hasattr(args_probe, "output") and args_probe.output: + out_files.append(args_probe.output) - for i, f in enumerate(out_files): - if not os.path.exists(f): - logger.info(f"{red_tag("FAIL")} Missing output: {f}") + for idx, file in enumerate(out_files): + if not file.exists(): + logger.info(f"{red_tag("FAIL")} Missing output: {file}") continue - try: - ret_va = validate_ffprobe(f,test_dict[i]) + ret_va = validate_ffprobe(file,_test_dict[idx]) + return ret_va except ffmpeg.Error as e: - logger.error(f"{red_tag("FAIL")} Probe error: {f}") + logger.error(f"{red_tag("FAIL")} Probe error: {file}") logger.error(e) return False - return ret_va + def is_cover_art(stream): + if stream["codec_type"] != "video": + return False - def get_streams(test_file): - info = ffmpeg.probe(test_file) + if stream.get("disposition", {}).get("attached_pic") == 1: + return True + + if stream.get("nb_frames") and int(stream.get("nb_frames")) <= 1: + return True + tags = stream.get("tags", {}) + if "DURATION" in tags: + h, m, s = tags["DURATION"].split(":") + if int(h) * 3600 + int(m) * 60 + float(s) < 1: + return True + if "MIMETYPE" in tags and tags["MIMETYPE"] == "image/jpg": + return True + if "FILENAME" in tags and tags["FILENAME"] == "cover.jpg": + return True + + + return False + + 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): + _video_streams = [s for s in streams if s.get("codec_type") == "video" and not is_cover_art(s) ] + _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 is_cover_art(s) ] + 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 + return _video_streams,_audio_streams,_subtitle_streams,_cover_image def validate_ffprobe(file_path, expected: dict): """ @@ -1134,27 +1125,27 @@ def cmd_test(args): "video_codecs": list[str] } """ + if not file_path.is_file(): + logger.error(f"Test file {file_path} is not a file") + result = subprocess.run( + [ + "ffprobe", + "-v", "quiet", + "-print_format", "json", + "-show_chapters", + "-show_format", + file_path, + ], + capture_output=True, + text=True, + check=True, + ) - 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) + 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 @@ -1227,27 +1218,21 @@ def cmd_test(args): # ---------- 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 + keys = expected["metadata"].keys() + values = expected["metadata"].values() + for key, value in zip(keys,values): + 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)) - + 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"]): - logger.error(f"Found russian track: {audio_track} in file {file}") + logger.error(f"Found russian track: {audio_track} in file {file_path}") return return_value and False #logger.debug(f"Subtitle tracks: {subtitle_tracks}") @@ -1278,7 +1263,7 @@ def cmd_test(args): 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.output = Path(test_dir) / f"out_{str(cmd["name"]).replace("-","_")}_{Path(test_file).stem}{Path(test_file).suffix}" args_test.width = 640 args_test.height = 360 args_test.crf = 28 @@ -1296,7 +1281,7 @@ def cmd_test(args): 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]) + args_test.output = 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) # ---------- EXTRACT AUDIO TRACK ---------- @@ -1308,7 +1293,7 @@ def cmd_test(args): 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") + args_test.output = Path(test_dir) / f"extracted_audio_{Path(test_file).suffix[1:]}.mp3" ret = ret and run_and_probe(cmd["name"], cmd_extract_audio_track, args_test, test_dict) # ---------- CONVERT AV ---------- @@ -1318,7 +1303,7 @@ def cmd_test(args): 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.output = Path(test_dir) / f"out_{str(cmd["name"]).replace("-","_")}_{Path(test_file).stem}{Path(test_file).suffix}" args_test.video_codec = "libx264" args_test.audio_codec = "aac" args_test.preset = "fast" @@ -1335,7 +1320,7 @@ def cmd_test(args): 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.output = Path(test_dir) / f"out_{str(cmd["name"]).replace("-","_")}_{Path(test_file).stem}{Path(test_file).suffix}" 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"]}] @@ -1348,7 +1333,7 @@ def cmd_test(args): 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.output = Path(test_dir) / f"out_{str(cmd["name"]).replace("-","_")}_{Path(test_file).stem}{Path(test_file).suffix}" 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"]}] @@ -1361,8 +1346,9 @@ def cmd_test(args): 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.num = None 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]) + args_test.output = 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_subs, args_test, test_dict) # ---------- FILTER AUDIO ---------- @@ -1372,7 +1358,7 @@ def cmd_test(args): 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.output = Path(test_dir) / f"out_{str(cmd["name"]).replace("-","_")}_{Path(test_file).stem}{Path(test_file).suffix}" 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"]}] @@ -1385,7 +1371,7 @@ def cmd_test(args): 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.output = Path(test_dir) / f"out_{str(cmd["name"]).replace("-","_")}_{Path(test_file).stem}{Path(test_file).suffix}" args_test.start = "0:0:5" args_test.duration = "10s" test_dict = [ {"video_count": len(video_streams), "audio_count": len(audio_streams), @@ -1399,7 +1385,7 @@ def cmd_test(args): 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.output = Path(test_dir) / f"out_{str(cmd["name"]).replace("-","_")}_{Path(test_file).stem}{Path(test_file).suffix}" args_test.audio_index = 2 test_dict = [ {"video_count": len(video_streams), "audio_count": len(audio_streams), "subtitle_count": len(subtitle_streams)}] @@ -1412,7 +1398,7 @@ def cmd_test(args): 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.output = Path(test_dir) / f"out_{str(cmd["name"]).replace("-","_")}_{Path(test_file).stem}{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), @@ -1427,28 +1413,24 @@ def cmd_test(args): 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.output = Path(test_dir) / f"out_{str(cmd["name"]).replace("-","_")}_{Path(test_file).stem}{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"}} ] + "metadata":{"artist":"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) + for test_file in test_files: + args_test = ArgsBuffer() + args_test.input = test_file + args_test.output = Path(test_dir) / f"out_{str(cmd["name"]).replace("-","_")}_{Path(test_file).stem}{Path(test_file).suffix}" + args_test.loud = False + 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")) @@ -1463,15 +1445,7 @@ def main(): 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) # ---------------------------- @@ -1482,20 +1456,25 @@ def main(): 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") + base_p.add_argument("-r", "--recursive", action="store_true") + base_p.add_argument("-ll","--log-level",type=str.upper,default="INFO", + choices=["TRACE", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], + help="Set log level", ) + base_p.add_argument("--dry-run",action="store_true",help="Dry run only",) 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 = sub.add_parser(str(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()), + p_test = sub.add_parser(str(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"]) @@ -1503,7 +1482,7 @@ def main(): # 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 = sub.add_parser(str(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)") @@ -1517,7 +1496,7 @@ def main(): # 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 = sub.add_parser(str(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"]) @@ -1525,7 +1504,7 @@ def main(): # 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 = sub.add_parser(str(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"]) @@ -1534,14 +1513,14 @@ def main(): # 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 = sub.add_parser(str(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 = sub.add_parser(str(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"]) @@ -1549,7 +1528,7 @@ def main(): # 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 = sub.add_parser(str(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"]) @@ -1557,7 +1536,7 @@ def main(): # DOWNSIZE # ---------------------------- cmd=commands["downsize"] - p_ds = sub.add_parser(cmd["name"], aliases=["d"], parents=[base_p], help="Scale video dimensions (-vf scale)") + p_ds = sub.add_parser(str(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") @@ -1567,7 +1546,7 @@ def main(): # 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 = sub.add_parser(str(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"]) @@ -1575,7 +1554,7 @@ def main(): # 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 = sub.add_parser(str(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"]) @@ -1584,7 +1563,7 @@ def main(): # 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 = sub.add_parser(str(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"]) @@ -1592,7 +1571,7 @@ def main(): # REMOVE ALL RUSSIAN AUDIO # ---------------------------- cmd = commands["remove-russian-from-db"] - p_ch = sub.add_parser(cmd["name"], aliases=["rrfd"], parents=[base_p], + p_ch = sub.add_parser(str(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"]) @@ -1600,7 +1579,7 @@ def main(): # 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 = sub.add_parser(str(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"]) @@ -1610,13 +1589,13 @@ def main(): # PRINT METADATA cmd=commands["metadata-print"] - p_ch = sub.add_parser(cmd["name"], aliases=["mp"], parents=[base_p], + p_ch = sub.add_parser(str(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 = sub.add_parser(str(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"]) @@ -1718,12 +1697,19 @@ def main(): # ---- DIRECTORY ---- if input_path.is_dir(): counter=0 - sorted_files = sorted(input_path.iterdir(), key=natural_sort_key) + if args.recursive: + sorted_files = sorted(( p for p in input_path.rglob("*") + if p.is_file() and p.suffix.lower() in VIDEO_EXTS), + key=natural_sort_key,) + else: + sorted_files = sorted(( p for p in input_path.iterdir() + if p.is_file() and p.suffix.lower() in VIDEO_EXTS), + 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): + if ((file.suffix.lower() not in VIDEO_EXTS and file.suffix.lower() not in AUDIO_EXTS)): continue logger.info(f"Processing File: {file}") - args.input = str(file) + args.input = Path(file) if args.override: out_path = file.with_name(f"{file.stem}_override{file.suffix}") else: diff --git a/serverlibrary/__init__.py b/serverlibrary/__init__.py index 7f32da0..fbe2d27 100755 --- a/serverlibrary/__init__.py +++ b/serverlibrary/__init__.py @@ -1,7 +1,14 @@ from .core import ( + TRACE, + LOG_LEVELS, + CustomLogger, + blue, combined_score, get_audio_tracks, get_subtitle_tracks, + green, + red, + reset, safe_delete, safe_rename, score_match_fuzzy, @@ -11,4 +18,28 @@ from .core import ( tvdb_get, word_similarity, words, + yellow, ) + +__all__ = [ + "TRACE", + "LOG_LEVELS", + "CustomLogger", + "blue", + "combined_score", + "get_audio_tracks", + "get_subtitle_tracks", + "green", + "red", + "reset", + "safe_delete", + "safe_rename", + "score_match_fuzzy", + "score_match_jaccard", + "score_match_recall_jaccard", + "tmdb_get", + "tvdb_get", + "word_similarity", + "words", + "yellow", +] \ No newline at end of file diff --git a/serverlibrary/core.py b/serverlibrary/core.py index 912cf8a..a44a923 100755 --- a/serverlibrary/core.py +++ b/serverlibrary/core.py @@ -6,8 +6,33 @@ from difflib import SequenceMatcher from pathlib import Path import requests +import logging +# ----------------------------- +# Logger +# ----------------------------- +TRACE=5 + +red = "\033[91m" +green = "\033[92m" +yellow = "\033[93m" +blue = "\033[94m" +reset = "\033[0m" + +class CustomLogger(logging.Logger): + def trace(self, msg: str, *args, **kwargs) -> None: + if self.isEnabledFor(TRACE): + self._log(TRACE, msg, args, **kwargs) + +LOG_LEVELS = { + "TRACE": TRACE, + "DEBUG": logging.DEBUG, + "INFO": logging.INFO, + "WARNING": logging.WARNING, + "ERROR": logging.ERROR, + "CRITICAL": logging.CRITICAL, +} # ----------------------------- # Helpers # -----------------------------