From c5cf55fe2197b49d833bec6385a07f21813ea64d Mon Sep 17 00:00:00 2001 From: Lucas Vivo Date: Mon, 19 Oct 2026 13:24:04 +0200 Subject: [PATCH] ffmpeg handling py proper testing for chapters and metadata remove all russian finction --- media_editing/extract_dvds.py | 48 +- media_editing/ffmpeg_handling.py | 980 ++++++++++++------ serverlibary/__init__.py | 3 - serverlibary/core.py | 49 - serverlibrary/__init__.py | 4 + serverlibrary/core.py | 200 ++++ .../pyproject.toml | 2 +- 7 files changed, 894 insertions(+), 392 deletions(-) delete mode 100755 serverlibary/__init__.py delete mode 100755 serverlibary/core.py create mode 100755 serverlibrary/__init__.py create mode 100755 serverlibrary/core.py rename {serverlibary => serverlibrary}/pyproject.toml (91%) diff --git a/media_editing/extract_dvds.py b/media_editing/extract_dvds.py index 765b7a9..edc883c 100755 --- a/media_editing/extract_dvds.py +++ b/media_editing/extract_dvds.py @@ -7,7 +7,7 @@ import shutil import subprocess import sys from pathlib import Path - +from serverlibrary import get_audio_tracks def check_command(command): if shutil.which(command) is None: @@ -60,52 +60,6 @@ def scan_dvd(dvd_name: Path): return titles -def get_audio_tracks(dvd, title): - """Return HandBrake audio track information for a DVD title.""" - - result = subprocess.run( - [ - "HandBrakeCLI", - "-i", dvd, - "-t", str(title), - "--scan", - ], - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - text=True, - check=True, - ) - - output = result.stdout - - # Examples HandBrake may produce: - # - # + 1, Deutsch (AC3, 2.0 ch, 192 kbps) - # + 2, English (AC3, 2.0 ch, 224 kbps) - # - # Also handles German/Englisch/deu/eng where present. - - tracks = [] - - pattern = re.compile( - r"\+\s+(\d+),\s+(.+?)\s+\((?:AC3|AAC|MP3|E-AC-3|DTS)" - ) - - for match in pattern.finditer(output): - track_number = int(match.group(1)) - description = match.group(2).strip() - - tracks.append({ - "number": track_number, - "description": description, - }) - if not tracks: - tracks.append({ - "number": 0, - "description": "", - }) - return tracks - def choose_audio_tracks(tracks): """ Select German first, English second. diff --git a/media_editing/ffmpeg_handling.py b/media_editing/ffmpeg_handling.py index 86ee122..ba86d8c 100755 --- a/media_editing/ffmpeg_handling.py +++ b/media_editing/ffmpeg_handling.py @@ -1,5 +1,4 @@ -#!/home/lucas/.venv_general/bin/python3 - +#!/home/lucas/scripts/.venv/bin/python3 """ FFmpeg Python CLI Tool @@ -13,31 +12,139 @@ 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 -from serverlibary import safe_rename -VIDEO_EXTS = {".mp4", ".mkv", ".avi", ".mov", ".webm",".mp3", ".m4a", ".m4b", ".opus"} + +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) # ---------------------------- -REQUIRES_OUTPUT = { - "chapters", - "convert-av", - "convert-audio", - "convert-video", - "remove-subs", - "remove-audio", - "extract-audio", - "downsize", - "filter-audio", - "cut", - "set-primary-audio", - "metaset", +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 + +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 ---------- # # ---------------------------- @@ -72,7 +179,7 @@ def move_override_file(input_path, backup_files=False,silent=False): new_name = input_path / f"{base}{suffix}" if not original.exists(): - print(f"Skip (missing original): {original}") + logger.info(f"Skip (missing original): {original}") continue if backup_files: @@ -90,29 +197,21 @@ def run_ffmpeg(stream, overwrite=True, loud=False): "overwrite_output": overwrite, } - if loud: - kwargs.update( - capture_stdout=False, - capture_stderr=True, - quiet=True, - ) + + kwargs.update( + capture_stdout=True, + capture_stderr=True, + quiet=True, + ) try: out, err = ffmpeg.run(stream, **kwargs) - - if loud and err: - with open(LOG_FILE, "ab") as log: - log.write(err) - + logger.trace(out) + if err: + logger.error(err) except ffmpeg.Error as e: - if loud: - with open(LOG_FILE, "ab") as log: - log.write(e.stderr or b"") - print(f"FFmpeg failed. See log: {LOG_FILE}") - else: - print("FFmpeg failed.\n") - if e.stderr: - print(e.stderr.decode(errors="ignore")) + logger.error(e.stderr) + logger.error(f"FFmpeg failed.\n") sys.exit(1) def probe_audio_streams(file): @@ -125,7 +224,7 @@ def probe_audio_streams(file): try: probe = ffmpeg.probe(file) except ffmpeg.Error as e: - print("ffprobe failed:", e) + logger.error(f"ffprobe failed:", e) sys.exit(1) return [ @@ -142,27 +241,27 @@ def select_audio_stream(file, provided=None): streams = probe_audio_streams(file) if not streams: - print("No audio streams found.") + 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 - print(f"Provided index {provided} is not a valid audio stream.") + logger.info(f"Provided index {provided} is not a valid audio stream.") if len(streams) == 1: return streams[0][0] - print(f"Found {len(streams)} audio streams:") + logger.info(f"Found {len(streams)} audio streams:") for i, (idx, lang) in enumerate(streams): - print(f"[{i}] Index: {idx}, Language: {lang}") + 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): - print("Invalid selection.") + logger.error(f"Invalid selection.") sys.exit(1) def parse_time_to_seconds(value: str) -> float: @@ -213,8 +312,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): - ext = output_path.lower().split(".")[-1] +def get_scodec(output_path: str | Path): + ext = str(output_path).lower().split(".")[-1] if ext == "mp4": return "mov_text" @@ -245,8 +344,12 @@ def get_chapter_lengths(file_path): file_path ] - result = subprocess.run(cmd, capture_output=True, text=True) - + result = subprocess.run(cmd, + capture_output=True, + text=True, + check=True, + ) + logger.trace(result.stdout) if result.returncode != 0: raise RuntimeError(result.stderr) @@ -277,9 +380,9 @@ def cmd_info(args): """ try: info = ffmpeg.probe(args.input) - print(json.dumps(info, indent=2)) + logger.info(json.dumps(info, indent=2)) except ffmpeg.Error as e: - print("Error probing file:", e) + logger.error(f"Error probing file:", e) def cmd_convert_av(args): """ @@ -361,41 +464,122 @@ def cmd_remove_subs(args): Parameters: - sn: Disables subtitle recording. """ - inp = ffmpeg.input(args.input) - out = ffmpeg.output( - inp, - args.output, - sn=None, - c="copy", - map='0', - map_metadata=0 - ) - return out + 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). """ - inp = ffmpeg.input(args.input) probe = ffmpeg.probe(args.input) remove_idx = select_audio_stream(args.input, args.audio_index) - streams_to_keep = [] + 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 + ] - for s in probe["streams"]: - idx = s["index"] - if idx != remove_idx: - streams_to_keep.append(inp[str(idx)]) - - out = ffmpeg.output( - *streams_to_keep, - args.output, - vcodec="copy", - acodec="copy", - scodec= get_scodec(args.output), + removed_was_default = any( + s["index"] == remove_idx + and s.get("disposition", {}).get("default", 0) == 1 + for s in probe["streams"] ) - return out + + 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)) + 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): """ @@ -591,19 +775,18 @@ def cmd_cut_segment(args): start = parse_time_to_seconds(args.start) duration = parse_time_to_seconds(args.duration) - inp = ffmpeg.input(args.input) + inp = ffmpeg.input(args.input, ss=start) out = ffmpeg.output( inp, args.output, - ss=start, t=duration, + map="0", vcodec=get_vcodec(args.output), acodec="copy", scodec=get_scodec(args.output), - map="0" ) - + logger.trace(f"Command: {out}") return out def cmd_set_primary_audio(args): @@ -675,22 +858,22 @@ def cmd_show_metadata(args): probe = ffmpeg.probe(args.input) - print("=== FORMAT METADATA ===") + logger.info(f"=== FORMAT METADATA ===") format_tags = probe.get("format", {}).get("tags", {}) for k, v in format_tags.items(): - print(f"{k}: {v}") + logger.info(f"{k}: {v}") - #print("\n=== STREAM METADATA ===") + #logger.info(f"\n=== STREAM METADATA ===") #for i, stream in enumerate(probe.get("streams", [])): - # print(f"\n--- Stream {i} ({stream.get('codec_type')}) ---") + # logger.info(f"\n--- Stream {i} ({stream.get('codec_type')}) ---") # tags = stream.get("tags", {}) # if not tags: - # print("No tags") + # logger.info(f"No tags") # else: # for k, v in tags.items(): - # print(f"{k}: {v}") - return + # logger.info(f"{k}: {v}") + return format_tags def cmd_set_metadata(args): @@ -713,7 +896,12 @@ def cmd_set_metadata(args): args.input ] - subprocess.run(cmd, check=True) + result = subprocess.run(cmd, + capture_output=True, + text=True, + check=True, + ) + logger.trace(result.stdout) return None # Kid3 writes in-place # ───────────────────────────────────────────── @@ -723,7 +911,7 @@ def cmd_set_metadata(args): out = ffmpeg.output( inp, - args.output, + str(args.output), c="copy", metadata=f"{args.key}={args.value}" ) @@ -731,7 +919,7 @@ def cmd_set_metadata(args): def cmd_apply_chapters(args): """ - Apply chapters to an M4B using a simple title-only text file. + Apply chapter titles to an M4B using a simple title-only text file. Chapters file format: One title per line: @@ -754,12 +942,13 @@ def cmd_apply_chapters(args): 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 @@ -796,24 +985,70 @@ def cmd_apply_chapters(args): args.output ] - subprocess.run(cmd, check=True) + 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) - print(f"{args.file_counter}:{filename}") + 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) - print(f"{filename.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 None + 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)) + subtitle_tracks = translate_audio_tracks(get_subtitle_tracks(file)) + logger.trace(f"File: {file.name}: Audio tracks: {audio_tracks}") + remove_tracks = [] + remove_subtitle_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 track in subtitle_tracks: + if is_russian_track(track["description"]): + remove_subtitle_tracks.append(track["number"]) + logger.debug(f"Removing russian subtitle {track['number']} from {file}") + for number in remove_tracks: + args=ArgsBuffer() + args.input = file + 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)) + args=ArgsBuffer() + args.input = file + args.num = [remove_subtitle_tracks] + args.output = file.with_name(f"{file.stem}_remove_russian{file.suffix}") + + if cmd_remove_subs(args) == "No output needed": + logger.debug(safe_rename(args.output, file)) + + return "No output needed" + def cmd_test(args): """ @@ -821,11 +1056,11 @@ def cmd_test(args): Runs each command individually on test.mp4 and probes outputs. """ - script_dir = os.path.dirname(os.path.abspath(__file__)) + 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 = [ os.path.join(script_dir,file) for file in test_files ] - test_dir = "/tmp/" - + 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" @@ -840,27 +1075,26 @@ def cmd_test(args): return f"{GREEN}[{tag}]{RESET}" def run_and_probe(name, cmd_func, args, test_dict: list=[]): - print(f"TEST: {name}") + logger.info(f"TEST: {name}") # Run FFmpeg for this single command only outputs = cmd_func(args) - if not outputs: - print("No outputs returned.") - return 1 - - try: - ffmpeg.run( - outputs, - overwrite_output=True, - quiet=True, - capture_stderr=True - ) - except ffmpeg.Error as e: - print("FFmpeg failed:") - if e.stderr: - print(e.stderr.decode(errors="ignore")) - return 1 - + 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 = [] @@ -869,15 +1103,15 @@ def cmd_test(args): for i, f in enumerate(out_files): if not os.path.exists(f): - print(f"{red_tag("FAIL")} Missing output: {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: - print(f"{red_tag("FAIL")} Probe error: {f}") - print(e) - return 1 + logger.error(f"{red_tag("FAIL")} Probe error: {f}") + logger.error(e) + return False return ret_va @@ -885,14 +1119,15 @@ def cmd_test(args): info = ffmpeg.probe(test_file) streams = info.get("streams", []) - video_streams = [s for s in streams if s.get("codec_type") == "video"] + 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): - print("test file not found in script directory.") + logger.info(f"test file not found in script directory.") sys.exit(1) - return video_streams,audio_streams,subtitle_streams + return video_streams,audio_streams,subtitle_streams,cover_image def validate_ffprobe(file_path, expected: dict): """ @@ -908,29 +1143,44 @@ def cmd_test(args): } """ - info = ffmpeg.probe(file_path) - streams = info.get("streams", []) + 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, + ) - video_streams = [s for s in streams if s.get("codec_type") == "video"] - 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"] - ret = 0 + 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"]: - print(f"{red_tag("VIDEO COUNT FAIL")} expected {expected['video_count']}, got {len(video_streams)}") - ret = 1 + 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"]: - print(f"{red_tag("AUDIO COUNT FAIL")} expected {expected['audio_count']}, got {len(audio_streams)}") - ret = 1 + 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"]: - print(f"{red_tag("SUBTITLE COUNT FAIL")} expected {expected['subtitle_count']}, got {len(subtitle_streams)}") - ret = 1 + 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: @@ -941,8 +1191,8 @@ def cmd_test(args): expected_codecs = expected_codecs + [last] * (len(actual) - len(expected_codecs)) if actual != expected_codecs: - print(f"{red_tag("AUDIO CODEC FAIL")}\nExpected: {expected_codecs}\nGot: {actual}") - ret = 1 + 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]) @@ -951,156 +1201,267 @@ def cmd_test(args): last = expected_codecs[-1] if expected_codecs else None expected_codecs = expected_codecs + [last] * (len(actual) - len(expected_codecs)) if actual != expected_codecs: - print(f"{red_tag("VIDEO CODEC FAIL")}\nExpected: {expected_codecs}\nGot: {actual}") - ret = 1 + 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: - print(f"{red_tag("DURATION FAIL")} expected ~{expected_duration} (±{tolerance}), got {actual_duration}s") - ret = 1 + logger.error(f"{red_tag("DURATION FAIL")} expected ~{expected_duration} (±{tolerance}), got {actual_duration}s") + return_value = return_value and False - print(f"{green_tag("PASS")} {file_path}") - return ret + # ---------- 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 - class Args: - pass + 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 - ret = 0 + # ---------- 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 = Args() + #args = ArgsBuffer() #args.input = test_file_mp4 #test_dict = {} - #run_and_probe("info", cmd_info, args, test_dict) + #run_and_probe(cmd["name"], cmd_info, args, test_dict) #args.input = test_file_mkv - #ret += run_and_probe("info", cmd_info, args, test_dict) + #ret = ret and run_and_probe(cmd["name"], cmd_info, args, test_dict) # ---------- DOWNSIZE ---------- - args = Args() - for test_file in test_files: - video_streams, audio_streams, subtitle_streams = get_streams(test_file) - args.input = test_file - args.output = os.path.join(test_dir, "out_downsize" + os.path.splitext(test_file)[1]) - args.width = 640 - args.height = 360 - args.crf = 28 - test_dict = [{"video_count": len(video_streams), "audio_count": len(audio_streams), - "subtitle_count": len(subtitle_streams), "video_codecs": ["h264"]}] - ret += run_and_probe("downsize", cmd_downsize, args, test_dict) + 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 ---------- - args = Args() - args.audio_index = 1 + 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 = get_streams(test_file) - args.input = test_file - test_dict = [{"video_count": len(video_streams), "audio_count": 1, "subtitle_count": len(subtitle_streams)}] - args.output = os.path.join(test_dir, "out_remove_audio" + os.path.splitext(test_file)[1]) - ret += run_and_probe("remove-audio-track", cmd_remove_audio_track, args, test_dict) + 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 ---------- - args = Args() - args.audio_index = 1 + 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.input = test_file - test_dict = [{"video_count": 0, "audio_count": 1, "subtitle_count": 0}] - args.output = os.path.join(test_dir, "extracted_audio_" + os.path.splitext(test_file)[1][1:] + ".mp3") - ret += run_and_probe("remove-audio-track", cmd_extract_audio_track, args, test_dict) + 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 ---------- - args = Args() - for test_file in test_files: - video_streams, audio_streams, subtitle_streams = get_streams(test_file) - args.input = test_file - args.output = os.path.join(test_dir, "out_convert_av" + os.path.splitext(test_file)[1]) - args.video_codec = "libx264" - args.audio_codec = "aac" - args.preset = "fast" - args.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 += run_and_probe("convert-av", cmd_convert_av, args, test_dict) - + 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 ---------- - args = Args() - for test_file in test_files: - video_streams, audio_streams, subtitle_streams = get_streams(test_file) - args.input = test_file - args.output = os.path.join(test_dir, "out_convert_audio" + os.path.splitext(test_file)[1]) - args.audio_codec = "aac" - test_dict = [ {"video_count": len(video_streams), "audio_count": len(audio_streams), - "subtitle_count": len(subtitle_streams), "audio_codecs": ["aac"]}] - ret += run_and_probe("convert-audio", cmd_convert_audio, args, test_dict) + 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 ---------- - args = Args() - for test_file in test_files: - video_streams, audio_streams, subtitle_streams = get_streams(test_file) - args.input = test_file - args.output = os.path.join(test_dir, "out_convert_video" + os.path.splitext(test_file)[1]) - args.video_codec = "libx264" - test_dict = [ {"video_count": len(video_streams), "audio_count": len(audio_streams), - "subtitle_count": len(subtitle_streams), "video_codecs": ["h264"]}] - ret += run_and_probe("convert-video", cmd_convert_video, args, test_dict) + 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 ---------- - args = Args() - for test_file in test_files: - video_streams, audio_streams, subtitle_streams = get_streams(test_file) - args.input = test_file - test_dict = [ {"video_count": len(video_streams), "audio_count": len(audio_streams), "subtitle_count": 0}] - args.output = os.path.join(test_dir, "out_no_subs" + os.path.splitext(test_file)[1]) - ret += run_and_probe("remove-subs", cmd_remove_subs, args, test_dict) + 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 ---------- - args = Args() - for test_file in test_files: - video_streams, audio_streams, subtitle_streams = get_streams(test_file) - args.input = test_file - args.output = os.path.join(test_dir, "out_filter_audio" + os.path.splitext(test_file)[1]) - args.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 += run_and_probe("filter-audio", cmd_filter_audio, args, test_dict) + 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 ---------- - args = Args() - for test_file in test_files: - video_streams, audio_streams, subtitle_streams = get_streams(test_file) - args.input = test_file - args.output = os.path.join(test_dir, "out_cut" + os.path.splitext(test_file)[1]) - args.start = "0:0:5" - args.duration = "10s" - test_dict = [ {"video_count": len(video_streams), "audio_count": len(audio_streams), - "subtitle_count": len(subtitle_streams), "duration": "10s"}] - ret += run_and_probe("cut", cmd_cut_segment, args, test_dict) + 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 ---------- - args = Args() - for test_file in test_files: - video_streams, audio_streams, subtitle_streams = get_streams(test_file) - args.input = test_file - args.output = os.path.join(test_dir, "out_primary_audio" + os.path.splitext(test_file)[1]) - args.audio_index = 2 - test_dict = [ {"video_count": len(video_streams), "audio_count": len(audio_streams), - "subtitle_count": len(subtitle_streams)}] - ret += run_and_probe("set-primary-audio", cmd_set_primary_audio, args, test_dict) + 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) - if ret == 0: - print(green_tag("ALL TESTS SUCCEEDED")) + # ---------- 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: - print(red_tag("SOME TESTS FAILED")) + logger.info(red_tag("SOME TESTS FAILED")) # ---------- CLI Setup ---------- # @@ -1114,28 +1475,40 @@ def main(): # BASE ARGS (shared) # ---------------------------- base_p = argparse.ArgumentParser(add_help=False) + + + 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("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_false", help="Run ffmpeg loud") + 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 # ---------------------------- - p_info = sub.add_parser("info", aliases=["i"], parents=[base_p], help="Inspect media metadata") - p_info.set_defaults(func=cmd_info, command="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 - p_test = sub.add_parser("test", aliases=["t"], help="Run test function") - p_test.set_defaults(func=cmd_test, command="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 # ---------------------------- - p_av = sub.add_parser("convert-av", aliases=["cav"], parents=[base_p], help="Full re-encode of video/audio") + 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)") @@ -1143,115 +1516,137 @@ def main(): 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="convert-av") + p_av.set_defaults(func=cmd_convert_av, command=cmd["name"]) # ---------------------------- # CONVERT AUDIO # ---------------------------- - p_a = sub.add_parser("convert-audio", aliases=["ca"], parents=[base_p], help="Change audio format only") + 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="convert-audio") + p_a.set_defaults(func=cmd_convert_audio, command=cmd["name"]) # ---------------------------- # CONVERT VIDEO # ---------------------------- - p_v = sub.add_parser("convert-video", aliases=["cv"], parents=[base_p], help="Change video format only") + 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="convert-video") + p_v.set_defaults(func=cmd_convert_video, command=cmd["name"]) # ---------------------------- # REMOVE SUBS # ---------------------------- - p_rs = sub.add_parser("remove-subs", aliases=["rs"], parents=[base_p], help="Strip all subtitle streams (-sn)") - p_rs.set_defaults(func=cmd_remove_subs, command="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 # ---------------------------- - p_ra = sub.add_parser("remove-audio", aliases=["ra"], parents=[base_p], help="Remove one 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="remove-audio") + p_ra.set_defaults(func=cmd_remove_audio_track, command=cmd["name"]) # ---------------------------- - # REMOVE AUDIO TRACK + # EXTRACT AUDIO TRACK # ---------------------------- - p_ea = sub.add_parser("extract-audio", aliases=["ea"], parents=[base_p], help="Extract one 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="extract-audio") + p_ea.set_defaults(func=cmd_extract_audio_track, command=cmd["name"]) # ---------------------------- # DOWNSIZE # ---------------------------- - p_ds = sub.add_parser("downsize", aliases=["d"], parents=[base_p], help="Scale video dimensions (-vf scale)") + 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="downsize") + p_ds.set_defaults(func=cmd_downsize, command=cmd["name"]) # ---------------------------- # FILTER AUDIO # ---------------------------- - p_fa = sub.add_parser("filter-audio", aliases=["fa"], parents=[base_p], help="Apply audio enhancement chain") + 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="filter-audio") + p_fa.set_defaults(func=cmd_filter_audio, command=cmd["name"]) # ---------------------------- # CUT # ---------------------------- - p_c = sub.add_parser("cut", aliases=["c"], parents=[base_p], help="Cut X seconds out of video") + 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="cut") + p_c.set_defaults(func=cmd_cut_segment, command=cmd["name"]) # ---------------------------- # SET PRIMARY AUDIO # ---------------------------- - p_pa = sub.add_parser("set-primary-audio", aliases=["spa"], parents=[base_p], help="Set primary audio track") + 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="set-primary-audio") + p_pa.set_defaults(func=cmd_set_primary_audio, command=cmd["name"]) # ---------------------------- - # CHAPTERS + # REMOVE ALL RUSSIAN AUDIO # ---------------------------- - p_ch = sub.add_parser("chapters",aliases=["ch"],parents=[base_p],help="Apply chapter metadata from ffmetadata file") + 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="chapters") + p_ch.set_defaults(func=cmd_apply_chapters, command=cmd["name"]) # ---------------------------- # METADATA # ---------------------------- - # PRINT - - p_ch = sub.add_parser("metaprint", aliases=["mp"], parents=[base_p], + # 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="metaprint") + p_ch.set_defaults(func=cmd_show_metadata, command=cmd["name"]) - # SET - - p_ch = sub.add_parser("metaset",aliases=["ms"],parents=[base_p],help="Apply metadata key and value to input file") + # 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="metaset") + 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"): - print("\nAvailable commands:") + logger.info(f"\nAvailable commands:") seen = set() for name, parser in sub.choices.items(): if id(parser) in seen: continue seen.add(id(parser)) - print(f" - {name}") + logger.info(f" - {name}") args.command = ensure(None, "Command") @@ -1270,7 +1665,7 @@ def main(): args.input = Path(args.input).resolve() if not Path(args.input).exists(): parser.error("Input file does not exist!") - if args.command in REQUIRES_OUTPUT and not getattr(args, "output", None): + 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}")) @@ -1279,7 +1674,6 @@ def main(): 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}")) - print(args.output) # ---------------------------- # COMMAND-SPECIFIC REQUIRED FIELDS # ---------------------------- @@ -1313,7 +1707,7 @@ def main(): if args.command == "chapters": args.chapter_file = ensure(getattr(args, "chapter_file", None), "File with chapternames") - if args.command == "metaset": + 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") @@ -1331,9 +1725,9 @@ def main(): 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: + if (file.suffix.lower() not in VIDEO_EXTS or file.suffix.lower() not in AUDIO_EXTS): continue - print(f"Processing File: {file}") + logger.info(f"Processing File: {file}") args.input = str(file) if args.override: out_path = file.with_name(f"{file.stem}_override{file.suffix}") @@ -1341,7 +1735,9 @@ def main(): out_path = file.with_name(f"{file.stem}_{args.command}{file.suffix}") args.output = str(out_path) args.file_counter = counter - outputs.append(args.func(args)) + out = args.func(args) + if out != "No output needed": + outputs.append(out) counter+=1 # ---- SINGLE FILE ---- elif input_path.is_file(): @@ -1350,7 +1746,7 @@ def main(): raise ValueError(f"Invalid input path: {args.input}") if any(item is None for item in outputs) or len(outputs) == 0: - print("Output is NONE") + logger.info(f"Output is NONE") else: if not isinstance(outputs, list): outputs = [outputs] @@ -1364,9 +1760,9 @@ def main(): if args.override: move_override_file(input_path, args.backup,args.loud) except ffmpeg.Error as e: - print("FFmpeg failed.\n") + logger.error(f"FFmpeg failed.\n") if e.stderr: - print(e.stderr.decode(errors="ignore")) + logger.error(e.stderr.decode(errors="ignore")) sys.exit(1) sys.exit(0) diff --git a/serverlibary/__init__.py b/serverlibary/__init__.py deleted file mode 100755 index 0fae776..0000000 --- a/serverlibary/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ - -from .core import safe_rename -from .core import safe_delete diff --git a/serverlibary/core.py b/serverlibary/core.py deleted file mode 100755 index 2b4ccd8..0000000 --- a/serverlibary/core.py +++ /dev/null @@ -1,49 +0,0 @@ - -from pathlib import Path -import shutil - -# ----------------------------- -# Helpers -# ----------------------------- -def safe_delete(path: Path): - if path.is_dir(): - shutil.rmtree(path) - else: - path.unlink() - -def safe_rename(src: Path, dst: Path, dry: bool=False, confirm: bool=False, silent=False): - """ - Safely rename a file or directory without overwriting existing files. - Automatically appends _1, _2, ... if needed. - """ - - src = Path(src) - dst = Path(dst) - - # nothing to do - if src.resolve() == dst.resolve(): - return - - directory = dst.parent - base = dst.stem - ext = dst.suffix - - final_dst = dst - counter = 1 - - # find free name - while final_dst.exists(): - final_dst = directory / f"{base}_{counter}{ext}" - counter += 1 - if not silent: - print(f"{src} → {dst}") - - if dry: - return - - if confirm: - input("Press Enter to confirm...") - - dst.parent.mkdir(parents=True, exist_ok=True) - src.rename(dst) - diff --git a/serverlibrary/__init__.py b/serverlibrary/__init__.py new file mode 100755 index 0000000..99dc9a5 --- /dev/null +++ b/serverlibrary/__init__.py @@ -0,0 +1,4 @@ +from .core import safe_rename +from .core import safe_delete +from .core import get_audio_tracks +from .core import get_subtitle_tracks diff --git a/serverlibrary/core.py b/serverlibrary/core.py new file mode 100755 index 0000000..adb41dc --- /dev/null +++ b/serverlibrary/core.py @@ -0,0 +1,200 @@ + +from pathlib import Path +import shutil +import subprocess +import json +import re +from typing import Any + + +# ----------------------------- +# Helpers +# ----------------------------- +def safe_delete(path: Path): + if path.is_dir(): + shutil.rmtree(path) + else: + path.unlink() + +def safe_rename(src: Path, dst: Path, dry: bool=False, confirm: bool=False, silent=False): + """ + Safely rename a file or directory without overwriting existing files. + Automatically appends _1, _2, ... if needed. + """ + + src = Path(src) + dst = Path(dst) + + # nothing to do + if src.resolve() == dst.resolve(): + return + + directory = dst.parent + base = dst.stem + ext = dst.suffix + + final_dst = dst + counter = 1 + + # find free name + while final_dst.exists(): + final_dst = directory / f"{base}_{counter}{ext}" + counter += 1 + ret=f"{src} → {dst}" + if not silent: + print(ret) + + if dry: + return + + if confirm: + input("Press Enter to confirm...") + + dst.parent.mkdir(parents=True, exist_ok=True) + src.rename(dst) + return ret + +def get_audio_tracks( source: str, title: int | None = None, handbrake: bool = False) -> list[dict]: + """ + Return audio track information from a media file or DVD title. + + When ``handbrake`` is True, audio tracks are detected using + HandBrakeCLI's scan output. This is useful for DVDs and other + optical media sources where ffprobe may provide incomplete track + information. + + When ``handbrake`` is False, ffprobe is used to inspect the media + file and return all audio streams. + + Args: + source: + Path to the media file or DVD device/directory. + + title: + DVD title number to scan when using HandBrakeCLI. + Ignored when using ffprobe. + + handbrake: + If True, use HandBrakeCLI for track detection. + If False, use ffprobe. + + Returns: + List of dictionaries containing: + + - number: audio track number/index + - description: language, title, or track description + + Example: + get_audio_tracks("movie.mkv") + [{'number': 1, 'description': 'eng'}] + + get_audio_tracks("/dev/dvd", title=1, handbrake=True) + [{'number': 1, 'description': 'Deutsch'}] + """ + tracks: list[dict] = [] + + if handbrake: + result = subprocess.run( + [ + "HandBrakeCLI", + "-i", source, + "-t", str(title), + "--scan", + ], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + check=True, + ) + + pattern = re.compile( + r"\+\s+(\d+),\s+(.+?)\s+\((?:AC3|AAC|MP3|E-AC-3|DTS)" + ) + + for match in pattern.finditer(result.stdout): + tracks.append({ + "number": int(match.group(1)), + "description": match.group(2).strip(), + }) + + else: + result = subprocess.run( + [ + "ffprobe", + "-v", "quiet", + "-print_format", "json", + "-show_streams", + source, + ], + capture_output=True, + text=True, + check=True, + ) + + info = json.loads(result.stdout) + + for stream in info["streams"]: + if stream.get("codec_type") != "audio": + continue + + tracks.append({ + "number": stream["index"], + "description": ( + stream.get("tags", {}).get("title") + or stream.get("tags", {}).get("language") + or f"Track {stream['index']}" + ), + "primary": stream.get("codec_type") == "audio" and stream.get("disposition", {}).get("default") == 1 + }) + + if not tracks: + tracks.append({ + "number": 0, + "description": "", + "primary": False + }) + + return tracks + +def get_subtitle_tracks(source): + result = subprocess.run( + [ + "ffprobe", + "-v", "quiet", + "-print_format", "json", + "-show_streams", + source, + ], + capture_output=True, + text=True, + check=True, + ) + + info = json.loads(result.stdout) + + tracks = [] + + for stream in info["streams"]: + if stream.get("codec_type") != "subtitle": + continue + + tracks.append({ + "number": stream["index"], + "description": ( + stream.get("tags", {}).get("title") + or stream.get("tags", {}).get("language") + or f"Subtitle {stream['index']}" + ), + "primary": ( + stream.get("disposition", {}).get("default") == 1 + ), + }) + + if not tracks: + tracks.append({ + "number": 0, + "description": "", + "primary": False, + }) + + return tracks \ No newline at end of file diff --git a/serverlibary/pyproject.toml b/serverlibrary/pyproject.toml similarity index 91% rename from serverlibary/pyproject.toml rename to serverlibrary/pyproject.toml index c3502bf..9d2d106 100755 --- a/serverlibary/pyproject.toml +++ b/serverlibrary/pyproject.toml @@ -3,7 +3,7 @@ requires = ["setuptools", "wheel"] build-backend = "setuptools.build_meta" [project] -name = "serverlibary" +name = "serverlibrary" version = "0.1.0" description = "Serverhost Python library" authors = [{ name = "Lucas" }]