From 6d229c59f7f353e8964a0e3cb91778c724da481a Mon Sep 17 00:00:00 2001 From: Lucas Media Date: Mon, 26 Oct 2026 09:44:05 +0100 Subject: [PATCH] fix(ffmpeg handling): add translation, revise logging add more translations add setup logging function for file and stdout logger combine translation functions: translate_audio_tracks, translate_audio_descriptions formatting changes fix move_override function --- media_editing/ffmpeg_handling.py | 1149 +++++++++++++++++++----------- serverlibrary/__init__.py | 5 +- serverlibrary/core.py | 35 +- 3 files changed, 782 insertions(+), 407 deletions(-) diff --git a/media_editing/ffmpeg_handling.py b/media_editing/ffmpeg_handling.py index 4d24792..9bd9fc7 100755 --- a/media_editing/ffmpeg_handling.py +++ b/media_editing/ffmpeg_handling.py @@ -7,16 +7,18 @@ Parameters are documented with their equivalent FFmpeg command-line flags. """ import argparse -import os -import sys import json -import ffmpeg -import re import logging +import os +import re import subprocess +import sys import tempfile from pathlib import Path + +import ffmpeg from mutagen.easyid3 import EasyID3 + from serverlibrary import * logging.setLoggerClass(CustomLogger) @@ -29,53 +31,115 @@ AUDIO_EXTS = {".mp3", ".m4a", ".m4b", ".opus"} # OUTPUT ENFORCEMENT (clean + centralized) # ---------------------------- commands = { -"test":{"name":"test", "output": False}, -"info":{"name":"info", "output": False}, -"convert-av":{"name":"convert-av", "output": True}, -"convert-audio":{"name":"convert-audio", "output": True}, -"convert-video":{"name":"convert-video", "output": True}, -"remove-subs":{"name":"remove-subs", "output": True}, -"remove-audio":{"name":"remove-audio", "output": True}, -"extract-audio":{"name":"extract-audio", "output": True}, -"downsize":{"name":"downsize", "output": True}, -"filter-audio":{"name":"filter-audio", "output": True}, -"cut":{"name":"cut", "output": True}, -"set-primary-audio":{"name":"set-primary-audio", "output": True}, -"chapters":{"name":"chapters", "output": True}, -"metadata-print":{"name":"metadata-print", "output": False}, -"metadata-set":{"name":"metadata-set", "output": True}, -"remove-russian-from-db":{"name":"remove-russian-from-db", "output": False}, + "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": True}, } # ------ 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", + # Additional descriptors + "Закадровый": "Voice-over", + "Закадровое озвучивание": "Voice-over", + "Многоголосый закадровый": "Multi-voice voice-over", + "Многоголосное закадровое озвучивание": "Multi-voice voice-over", + # Variants found in metadata + "Оригинален език": "Original language", + "Многоголосное озвучивание": "Multi-voice dub", + "Многоголосый": "Multi-voice", + "Любительский": "Amateur voice-over", + "многоголосый": "Multi-voice", + "Профессиональный": "Professional", + "Одноголосый": "Single voice", + "Двухголосый": "Dual voice", + "Авторский": "Author voice-over", + "закадровый": "Voice-over", + "Оригинал": "Original", + "Дубляж": "Dubbed", + # Russian groups + "Кубик в кубе": "Kubik v Kube", + "LostFilm": "LostFilm", + "AlexFilm": "AlexFilm", + "Baibako": "Baibako", + "NewStudio": "NewStudio", + "Jaskier": "Jaskier", + "Kerob": "Kerob", + "TVShows": "TVShows", + "HDRezka": "HDRezka", + # Groups / studios + "Синема УС": "Cinema US", + "KinoPub": "KinoPub", + "Озвучка Rumble": "Rumble", + "Rumble": "Rumble", + # Voice actor + "Юрий Сербин": "Yuriy Serbin", + "Ю. Сербин": "Y. Serbin", + # Languages + "Английски": "English", + "Русский": "Russian", + "Русская": "Russian", + "Русское": "Russian", + # ISO codes + "rus": "Russian", + "ru": "Russian", + "ger": "German", + "deu": "German", + "eng": "English", + "und": "Unknown", } RUSSIAN_RE = re.compile( - r"\brus\b|\bru\b|russian|dubbed|рус|русский|русская|дубляж|" -r"оригинал|многоголос|двухголос|одноголос|авторский|любительский", - re.IGNORECASE, - ) + r"\brus\b|" + r"\bru\b|" + r"russian|" + r"рус|" + r"русский|" + r"русская|" + r"русское|" + r"дубляж|" + r"оригинал|" + r"многоголос|" + r"двухголос|" + r"одноголос|" + r"авторский|" + r"любительский|" + r"профессиональн|" + r"закадров|" + r"юрий\s+сербин|" + r"ю\.\s*сербин|" + r"сербин|" + r"синема\s+ус|" + r"kinopub|" + r"rumble|" + r"озвучка\s+rumble|" + r"кубик\s+в\s+кубе|" + r"lostfilm|" + r"alexfilm|" + r"baibako|" + r"newstudio|" + r"jaskier|" + r"kerob|" + r"tvshows|" + r"hdrezka", + 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 @@ -89,44 +153,49 @@ class ArgsBuffer: video_codec: str | None = None preset: str | None = None quality: int | None = None - width: int | None = None - height: 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(): - description = re.sub( - rf"\b{re.escape(src)}\b", - dst, - description, - flags=re.IGNORECASE, - ) - CYRILLIC_RE = re.compile(r"[А-Яа-яЁё]") +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 - ] + translated = [] + + for track in tracks: + description = track.get("description", "") + + for src, dst in AUDIO_TRANSLATIONS.items(): + description = re.sub( + rf"\b{re.escape(src)}\b", + dst, + description, + flags=re.IGNORECASE, + ) + + if CYRILLIC_RE.search(description): + raise ValueError( + f"Contains unidentified Cyrillic characters: {description}" + ) + + translated.append( + { + **track, + "description": description, + } + ) + + return translated + # ---------- Utilities ---------- # + # ---------------------------- # HELPERS # ---------------------------- @@ -134,50 +203,49 @@ def ask(prompt, default=None): value = input(f"{prompt}" + (f" [{default}]" if default else "") + ": ").strip() return value if value else default + def ensure(value, prompt): if value is None: return ask(prompt) return value -def move_override_file(input_path, backup_files=False,silent=False): - if input_path.is_file(): - input_path=input_path.parent - for override_path in input_path.glob("*_override.*"): - if not override_path.is_file(): - continue - stem = override_path.stem - suffix = override_path.suffix +def move_override_file(files_move_from, files_move_to, args): - if not stem.endswith("_override"): - continue + backup_files = args.backup + silent = not args.loud + for file_to, file_from in zip(files_move_to, files_move_from): + if not file_to.is_file(): + logger.error(f"Override Target File does not exist: {file_to}") + return False - base = stem.removesuffix("_override") - - original = input_path / f"{base}{suffix}" - backup = input_path / f".backup_{base}{suffix}" - new_name = input_path / f"{base}{suffix}" - - if not original.exists(): - logger.info(f"Skip (missing original): {original}") + if not file_from.is_file(): + logger.trace(f"Override File does not exist: {file_from}") continue + backup = file_to.parent / f"{file_to.stem}_backup{file_from.suffix}" if backup_files: # move original -> backup - safe_rename(original, backup,silent=silent) + logger.trace(f"Move original to backup: {backup}") + safe_rename(file_to, backup, silent=silent) # move override -> original name - safe_rename(override_path, new_name,silent=silent) + retstr, ret = safe_rename(file_from, file_to, silent=silent, overwrite=True) + logger.trace(retstr) + if not ret: + logger.error(f"Safe rename failed for: {file_from} -> {file_to}") + return False + return True LOG_FILE = Path.home() / "ffmpeg_log.log" -def run_ffmpeg(stream, overwrite=True, loud=False,dry_run = False): + +def run_ffmpeg(stream, overwrite=True, loud=False, dry_run=False): kwargs = { "overwrite_output": overwrite, } - kwargs.update( capture_stdout=True, capture_stderr=True, @@ -194,9 +262,10 @@ def run_ffmpeg(stream, overwrite=True, loud=False,dry_run = False): logger.error(err) except ffmpeg.Error as e: logger.error(e.stderr) - logger.error(f"FFmpeg failed.\n") + logger.error("FFmpeg failed.\n") sys.exit(1) + def probe_audio_streams(file): """ Probes metadata to identify audio tracks. @@ -207,7 +276,7 @@ def probe_audio_streams(file): try: probe = ffmpeg.probe(file) except ffmpeg.Error as e: - logger.error(f"ffprobe failed:", e) + logger.error(f"ffprobe failed:{e}") sys.exit(1) return [ @@ -216,6 +285,7 @@ def probe_audio_streams(file): if s.get("codec_type") == "audio" ] + def select_audio_stream(file, provided=None): """ Logical handler for stream selection. Ensures we target the correct @@ -224,7 +294,7 @@ def select_audio_stream(file, provided=None): streams = probe_audio_streams(file) if not streams: - logger.info(f"No audio streams found.") + logger.info("No audio streams found.") sys.exit(1) provided = int(provided) if provided else None @@ -244,9 +314,10 @@ def select_audio_stream(file, provided=None): choice = int(input("Select stream number (the list index [x]): ")) return streams[choice][0] except (ValueError, IndexError): - logger.error(f"Invalid selection.") + logger.error("Invalid selection.") sys.exit(1) + def parse_time_to_seconds(value: str) -> float: """ Convert human-readable time formats into seconds. @@ -269,9 +340,7 @@ def parse_time_to_seconds(value: str) -> float: return float(value[:-1]) # Case 2: h/m/s format (e.g. 1h2m3s) - pattern = re.compile( - r"(?:(?P\d+)h)?(?:(?P\d+)m)?(?:(?P\d+)s)?" - ) + pattern = re.compile(r"(?:(?P\d+)h)?(?:(?P\d+)m)?(?:(?P\d+)s)?") match = pattern.fullmatch(value) if match and any(match.groups()): h = int(match.group("h") or 0) @@ -295,6 +364,7 @@ def parse_time_to_seconds(value: str) -> float: except ValueError: raise argparse.ArgumentTypeError(f"Invalid time format: {value}") + def get_scodec(output_path: Path): ext = output_path.suffix @@ -306,6 +376,7 @@ def get_scodec(output_path: Path): return "copy" + def get_vcodec(output_path: Path): ext = output_path.suffix if ext == "mp4": @@ -313,6 +384,7 @@ def get_vcodec(output_path: Path): return "copy" + def get_chapter_lengths(file_path): """ Returns an array of chapter durations (in seconds). @@ -320,17 +392,20 @@ def get_chapter_lengths(file_path): cmd = [ "ffprobe", - "-v", "error", - "-print_format", "json", + "-v", + "error", + "-print_format", + "json", "-show_chapters", - file_path + file_path, ] - result = subprocess.run(cmd, - capture_output=True, - text=True, - check=True, - ) + result = subprocess.run( + cmd, + capture_output=True, + text=True, + check=True, + ) logger.trace(result.stdout) if result.returncode != 0: raise RuntimeError(result.stderr) @@ -339,19 +414,20 @@ def get_chapter_lengths(file_path): chapters = [] for ch in data.get("chapters", []): - chapters.append({ - "start": float(ch["start_time"]), - "end": float(ch["end_time"]) - }) + chapters.append( + {"start": float(ch["start_time"]), "end": float(ch["end_time"])} + ) return chapters + def natural_sort_key(path: Path): - s = str(path.name) # force string conversion + s = str(path.name) # force string conversion return [ - int(part) if part.isdigit() else part.lower() - for part in re.split(r'(\d+)', s) + int(part) if part.isdigit() else part.lower() for part in re.split(r"(\d+)", s) ] + + # ---------- Commands ---------- # def cmd_info(args): """ @@ -364,7 +440,8 @@ def cmd_info(args): info = ffmpeg.probe(str(args.input)) logger.info(json.dumps(info, indent=2)) except ffmpeg.Error as e: - logger.error(f"Error probing file:", e) + logger.error("Error probing file:", e) + def cmd_convert_av(args): """ @@ -389,12 +466,13 @@ def cmd_convert_av(args): preset=args.preset, qp=args.quality, # This maps all streams from the first input once - map='0', + map="0", # Prevents FFmpeg from trying to guess/add extra default streams - map_metadata=0 + map_metadata=0, ) return out + def cmd_convert_audio(args): """ Transcodes audio while keeping video intact. @@ -412,11 +490,12 @@ def cmd_convert_audio(args): vcodec="copy", acodec=args.audio_codec, scodec=get_scodec(args.output), - map='0', - map_metadata=0 + map="0", + map_metadata=0, ) return out + def cmd_convert_video(args): """ Transcodes video while keeping audio intact. @@ -431,11 +510,12 @@ def cmd_convert_video(args): vcodec=args.video_codec, acodec="copy", scodec=get_scodec(args.output), - map='0', - map_metadata=0 + map="0", + map_metadata=0, ) return out + def cmd_remove_subs(args): """ Strips all subtitle data. @@ -451,7 +531,8 @@ def cmd_remove_subs(args): cmd = [ "ffmpeg", "-y", - "-i", str(args.input), + "-i", + str(args.input), ] if args.num is None: @@ -462,23 +543,24 @@ def cmd_remove_subs(args): else: # remove only specified subtitle streams for stream in probe["streams"]: - if ( - stream["codec_type"] == "subtitle" - and stream["index"] in args.num - ): + 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), - ]) + 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)}") + logger.debug(f"Dry run: {' '.join(cmd)}") return "nooutputneeded" try: subprocess.run( @@ -494,59 +576,69 @@ def cmd_remove_subs(args): logger.error("STDERR:\n%s", e.stderr) raise return "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_idx = select_audio_stream(args.input, args.audio_index) remaining_audio = [ - { - **s, - "description": translate_audio_description( - s.get("description", "") - ), - } + s for s in probe["streams"] - if s["codec_type"] == "audio" - and s["index"] != remove_idx + if s["codec_type"] == "audio" and s["index"] != remove_idx ] + if not remaining_audio: + logger.error("No audiotracks remaining") + return False + remaining_audio = translate_audio_tracks(remaining_audio) removed_was_default = any( - s["index"] == remove_idx - and s.get("disposition", {}).get("default", 0) == 1 + s["index"] == remove_idx and s.get("disposition", {}).get("default", 0) == 1 for s in probe["streams"] ) cmd = [ "ffmpeg", "-y", - "-i", str(args.input), + "-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), - ]) + 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.extend( + [ + "-disposition:a", + "0", + "-disposition:a:0", + "default", + ] + ) cmd.append(str(args.output)) - logger.debug(" ".join(cmd)) + logger.trace(" ".join(cmd)) if args.dry_run: - logger.debug(f"Dry run: {" ".join(cmd)}") + logger.debug(f"Dry run: {' '.join(cmd)}") return "nooutputneeded" try: result = subprocess.run( @@ -555,12 +647,12 @@ def cmd_remove_audio_track(args): text=True, check=True, ) - logger.debug("FFmpegsubprocess successful") + logger.debug("FFmpeg subprocess successful") logger.trace(result.stdout) return "nooutputneeded" except subprocess.CalledProcessError as e: - logging.error( + logger.error( "Command failed (%s):\n%s\n\nstderr:\n%s", e.returncode, " ".join(map(str, e.cmd)), @@ -568,6 +660,7 @@ def cmd_remove_audio_track(args): ) raise RuntimeError(f"Command failed: {' '.join(e.cmd)}") from e + def cmd_extract_audio_track(args): """ 1. Saves that specific audio track as a standalone MP3 (Extra Output). @@ -588,13 +681,11 @@ def cmd_extract_audio_track(args): extracted_audio = inp[str(idx)] out = ffmpeg.output( - extracted_audio, - output_audio, - acodec='libmp3lame', - audio_bitrate='192k' + extracted_audio, output_audio, acodec="libmp3lame", audio_bitrate="192k" ) return out + def cmd_downsize(args): """ Scales resolution and compresses. @@ -624,10 +715,11 @@ def cmd_downsize(args): vcodec="libx264", crf=args.crf, acodec="copy", - scodec=get_scodec(args.output) + scodec=get_scodec(args.output), ) return out + def cmd_filter_audio(args): """ Multi-stage audio mastering chain. @@ -679,7 +771,6 @@ def cmd_filter_audio(args): metadata_args += [ f"-metadata:s:a:{out_audio_idx}", f"title={audio_titles[idx]}", - f"-metadata:s:a:{out_audio_idx}", f"language={audio_languages[idx]}", ] @@ -691,9 +782,10 @@ def cmd_filter_audio(args): # ------------------- if idx == idx_filt: audio_filtered = ( - audio - .filter("highpass", f=100) - .filter("acompressor", threshold="-22dB", ratio=4, attack=10, release=150) + audio.filter("highpass", f=100) + .filter( + "acompressor", threshold="-22dB", ratio=4, attack=10, release=150 + ) .filter("equalizer", f=2500, t="q", w=1, g=5) .filter("equalizer", f=4000, t="q", w=1, g=4) .filter("deesser", i=0.4, f=0.25) @@ -710,10 +802,8 @@ def cmd_filter_audio(args): metadata_args += [ f"-metadata:s:a:{out_audio_idx}", f"title={filtered_title}", - f"-metadata:s:a:{out_audio_idx}", f"language={lang}", - f"-metadata:s:a:{out_audio_idx}", f"handler_name={filtered_title}", ] @@ -730,21 +820,23 @@ def cmd_filter_audio(args): # ------------------- # OUTPUT (IMPORTANT FIX HERE) # ------------------- - input_args = [inp.video, + input_args = [ + inp.video, *filtered_audios, - *subs,] + *subs, + ] out = ffmpeg.output( *input_args, str(args.output), vcodec="copy", acodec="libvorbis", - scodec=get_scodec(args.output) + scodec=get_scodec(args.output), ).global_args(*metadata_args) - return out + def cmd_cut_segment(args): """ Cut a segment from a video or audio file. @@ -776,6 +868,7 @@ def cmd_cut_segment(args): logger.trace(f"Command: {out}") return out + def cmd_set_primary_audio(args): """ Set a chosen audio stream as the PRIMARY (first/default) audio track. @@ -799,18 +892,14 @@ def cmd_set_primary_audio(args): probe = ffmpeg.probe(str(args.input)) audio_streams = [ - s["index"] - for s in probe["streams"] - if s.get("codec_type") == "audio" + s["index"] for s in probe["streams"] if s.get("codec_type") == "audio" ] - reordered_audio = [primary_audio] + [ - a for a in audio_streams if a != primary_audio - ] + reordered_audio = [primary_audio] + [a for a in audio_streams if a != primary_audio] audio_inputs = [inp[f"a:{i - 1}"] for i in reordered_audio] - out_kwargs={} + out_kwargs = {} for i, stream in enumerate(reordered_audio): if stream == primary_audio: out_kwargs[f"disposition:a:{i}"] = "default" @@ -833,12 +922,13 @@ def cmd_set_primary_audio(args): vcodec="copy", acodec="copy", scodec=get_scodec(args.output), - **out_kwargs + **out_kwargs, ) return out -def cmd_show_metadata(args,show_metadata=False): + +def cmd_show_metadata(args, show_metadata=False): """ Prints metadata directly from ffprobe output. """ @@ -847,20 +937,20 @@ def cmd_show_metadata(args,show_metadata=False): format_tags = probe.get("format", {}).get("tags", {}) if show_metadata: - logger.info(f"=== FORMAT 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')}) ---") + # logger.info(f"\n=== STREAM METADATA ===") + # for i, stream in enumerate(probe.get("streams", [])): + # logger.info(f"\n--- Stream {i} ({stream.get('codec_type')}) ---") - # tags = stream.get("tags", {}) - # if not tags: - # logger.info(f"No tags") - # else: - # for k, v in tags.items(): - # logger.info(f"{k}: {v}") + # tags = stream.get("tags", {}) + # if not tags: + # logger.info(f"No tags") + # else: + # for k, v in tags.items(): + # logger.info(f"{k}: {v}") return format_tags @@ -878,19 +968,16 @@ def cmd_set_metadata(args): # OPUS → use Kid3 CLI (Vorbis comments) # ───────────────────────────────────────────── if ext == ".opus" or args.kidflag: - cmd = [ - "kid3-cli", - "-c", f"set {args.key} '{args.value}'", - args.input - ] + cmd = ["kid3-cli", "-c", f"set {args.key} '{args.value}'", args.input] if args.dry_run: - logger.debug(f"Dry run: {" ".join(cmd)}") + logger.debug(f"Dry run: {' '.join(cmd)}") return "nooutputneeded" - result = subprocess.run(cmd, - capture_output=True, - text=True, - check=True, - ) + result = subprocess.run( + cmd, + capture_output=True, + text=True, + check=True, + ) logger.trace(result.stdout) return None # Kid3 writes in-place @@ -900,13 +987,11 @@ def cmd_set_metadata(args): inp = ffmpeg.input(args.input) out = ffmpeg.output( - inp, - str(args.output), - c="copy", - metadata=f"{args.key}={args.value}" + inp, str(args.output), c="copy", metadata=f"{args.key}={args.value}" ) return out + def cmd_apply_chapters(args): """ Apply chapter titles to an M4B using a simple title-only text file. @@ -938,11 +1023,11 @@ def cmd_apply_chapters(args): chapters = get_chapter_lengths(args.input) if len(chapters) != len(titles): - logger.error(f"Chapter lengths do not match: {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 - }) + chapters[idx].update({"title": title}) metadata_text = ";FFMETADATA1\n\n" for ch in chapters: @@ -966,54 +1051,66 @@ def cmd_apply_chapters(args): cmd = [ "ffmpeg", "-y", - "-i", args.input, - "-f", "ffmetadata", - "-i", meta_path, - "-map_chapters", "1", - "-map_metadata", "0", - "-c", "copy", - args.output + "-i", + args.input, + "-f", + "ffmetadata", + "-i", + meta_path, + "-map_chapters", + "1", + "-map_metadata", + "0", + "-c", + "copy", + args.output, ] if args.dry_run: - logger.debug(f"Dry run: {" ".join(cmd)}") + logger.debug(f"Dry run: {' '.join(cmd)}") return "nooutputneeded" - result = subprocess.run(cmd, - capture_output=True, - text=True, - 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) - 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" + filename = Path(args.input) + logger.info(f"{args.file_counter}:{filename}") + title = titles[args.file_counter] + safe_title = title.replace("/", "-").replace(":", "-") + new_name = f"{args.file_counter + 1:03d} - {safe_title}.mp3" - new_path = filename.with_name(new_name) + new_path = filename.with_name(new_name) - logger.info(f"{filename.name} -> {new_name}") - audio = EasyID3(filename) - audio["title"] = title - audio["tracknumber"] = str(args.file_counter+1) - if args.dry_run: - logger.debug(f"Dry run") - return "nooutputneeded" - audio.save() - safe_rename(filename, new_path) + logger.info(f"{filename.name} -> {new_name}") + audio = EasyID3(filename) + audio["title"] = title + audio["tracknumber"] = str(args.file_counter + 1) + if args.dry_run: + logger.debug("Dry run") + return "nooutputneeded" + audio.save() + safe_rename(filename, new_path) return "nooutputneeded" + def cmd_remove_russian_from_db(args_in): """ Find all media with russian audio remove russian audio tracks """ file = args_in.input - args = ArgsBuffer() + logger.trace("Run remove russian cmd") + try: + audio_tracks = translate_audio_tracks(get_audio_tracks(file)) + except ValueError as e: + logger.warning("%s", e) + return "nooutputneeded" - 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: @@ -1021,12 +1118,10 @@ def cmd_remove_russian_from_db(args_in): 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 = args_in args.audio_index = number - args.output = args_in.output - - return cmd_remove_audio_track(args) + cmd_remove_audio_track(args) + return "nooutputneeded" def cmd_test(args): @@ -1036,12 +1131,16 @@ def cmd_test(args): """ test_file_dir = Path("/home/lucas/scripts/media_editing/testfiles") - test_files = [ "test_auto.mkv", "test_auto.mp4","test_auto_no_subs.mkv", "test_auto_no_subs.mp4"] - test_files = [ test_file_dir / file for file in test_files ] + test_files = [ + "test_auto.mkv", + "test_auto.mp4", + "test_auto_no_subs.mkv", + "test_auto_no_subs.mp4", + ] + test_files = [test_file_dir / file for file in test_files] test_dir = Path(tempfile.mkdtemp(prefix="ffmpeg_test_")) logger.info(f"Using test dir: {test_dir}") - def red_tag(tag: str) -> str: return f"{red}[{tag}]{reset}" @@ -1055,22 +1154,19 @@ def cmd_test(args): # Run FFmpeg for this single command only outputs = cmd_func(args_probe) if not outputs: - logger.error(f"No outputs returned.") + logger.error("No outputs returned.") return False if outputs != "nooutputneeded": try: ffmpeg.run( - outputs, - overwrite_output=True, - quiet=True, - capture_stderr=True + outputs, overwrite_output=True, quiet=True, capture_stderr=True ) except ffmpeg.Error as e: - logger.error(f"FFmpeg failed:") + logger.error("FFmpeg failed:") if e.stderr: logger.info(e.stderr.decode(errors="ignore")) return False - logger.info(f"FFmpeg succeeded") + logger.info("FFmpeg succeeded") # Probe result file(s) out_files = [] @@ -1079,13 +1175,13 @@ def cmd_test(args): for idx, file in enumerate(out_files): if not file.exists(): - logger.info(f"{red_tag("FAIL")} Missing output: {file}") + logger.info(f"{red_tag('FAIL')} Missing output: {file}") continue try: - ret_va = validate_ffprobe(file,_test_dict[idx]) + ret_va = validate_ffprobe(file, _test_dict[idx]) return ret_va except ffmpeg.Error as e: - logger.error(f"{red_tag("FAIL")} Probe error: {file}") + logger.error(f"{red_tag('FAIL')} Probe error: {file}") logger.error(e) return False @@ -1108,22 +1204,25 @@ def cmd_test(args): 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 not is_cover_art(s) ] + _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) ] + _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.") + logger.info("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): """ @@ -1143,8 +1242,10 @@ def cmd_test(args): result = subprocess.run( [ "ffprobe", - "-v", "quiet", - "-print_format", "json", + "-v", + "quiet", + "-print_format", + "json", "-show_chapters", "-show_format", file_path, @@ -1155,7 +1256,9 @@ def cmd_test(args): ) info = json.loads(result.stdout) - video_streams, audio_streams, subtitle_streams, cover_image = get_streams(file_path) + video_streams, audio_streams, subtitle_streams, cover_image = get_streams( + file_path + ) args = ArgsBuffer() args.input = file_path tags = cmd_show_metadata(args) @@ -1165,18 +1268,24 @@ def cmd_test(args): # ---------- COUNT CHECKS ---------- if "video_count" in expected: if len(video_streams) != expected["video_count"]: - logger.error(f"{red_tag("VIDEO COUNT FAIL")} expected {expected['video_count']}, got {len(video_streams)}") - return_value = return_value and False + logger.error( + f"{red_tag('VIDEO COUNT FAIL')} expected {expected['video_count']}, got {len(video_streams)}" + ) + return_value = return_value and False if "audio_count" in expected: if len(audio_streams) != expected["audio_count"]: - logger.error(f"{red_tag("AUDIO COUNT FAIL")} expected {expected['audio_count']}, got {len(audio_streams)}") - return_value = return_value and False + logger.error( + f"{red_tag('AUDIO COUNT FAIL')} expected {expected['audio_count']}, got {len(audio_streams)}" + ) + return_value = return_value and False if "subtitle_count" in expected: if len(subtitle_streams) != expected["subtitle_count"]: - logger.error(f"{red_tag("SUBTITLE COUNT FAIL")} expected {expected['subtitle_count']}, got {len(subtitle_streams)}") - return_value = return_value and False + 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: @@ -1184,56 +1293,73 @@ def cmd_test(args): expected_codecs = sorted(expected["audio_codecs"]) if len(expected_codecs) < len(actual): last = expected_codecs[-1] if expected_codecs else None - expected_codecs = expected_codecs + [last] * (len(actual) - len(expected_codecs)) + expected_codecs = expected_codecs + [last] * ( + len(actual) - len(expected_codecs) + ) if actual != expected_codecs: - logger.error(f"{red_tag("AUDIO CODEC FAIL")}\nExpected: {expected_codecs}\nGot: {actual}") - return_value = return_value and False + logger.error( + f"{red_tag('AUDIO CODEC FAIL')}\nExpected: {expected_codecs}\nGot: {actual}" + ) + return_value = return_value and False if "video_codecs" in expected: actual = sorted([s.get("codec_name") for s in video_streams]) expected_codecs = sorted(expected["video_codecs"]) if len(expected_codecs) < len(actual): last = expected_codecs[-1] if expected_codecs else None - expected_codecs = expected_codecs + [last] * (len(actual) - len(expected_codecs)) + expected_codecs = expected_codecs + [last] * ( + len(actual) - len(expected_codecs) + ) if actual != expected_codecs: - logger.error(f"{red_tag("VIDEO CODEC FAIL")}\nExpected: {expected_codecs}\nGot: {actual}") - return_value = return_value and False + 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"]) + expected_duration = parse_time_to_seconds(expected["duration"]) # allow small ffmpeg timing variance (important!) tolerance = expected.get("duration_tolerance", 0.2) if abs(actual_duration - expected_duration) <= tolerance: - logger.error(f"{red_tag("DURATION FAIL")} expected ~{expected_duration} (±{tolerance}), got {actual_duration}s") - return_value = return_value and False + logger.error( + f"{red_tag('DURATION FAIL')} expected ~{expected_duration} (±{tolerance}), got {actual_duration}s" + ) + return_value = return_value and False # ---------- CHAPTER CHECKS ---------- if "chapter_name" in expected: - chapter_name = [ chapter.get("tags", {}).get("title", "") for chapter in info.get("chapters", []) ] + 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 + logger.error( + f"Expected chapter_name: {expected_chapter_name}, got {chapter_name}" + ) + return_value = return_value and False if "chapter_count" in expected: chapter_num = len(info.get("chapters", [])) expected_chapter_count = expected["chapter_count"] if expected_chapter_count != chapter_num: - logger.error(f"Expected chapters: {expected_chapter_count}, got {chapter_num}") - return_value = return_value and False + logger.error( + f"Expected chapters: {expected_chapter_count}, got {chapter_num}" + ) + return_value = return_value and False # ---------- METADATA CHECKS ---------- if "metadata" in expected: keys = expected["metadata"].keys() values = expected["metadata"].values() - for key, value in zip(keys,values): + for key, value in zip(keys, values): if key in tags.keys() and tags[key] == value: logger.debug(f"{key}: {value}") else: @@ -1241,32 +1367,32 @@ def cmd_test(args): return_value = return_value and False if "no_russian_dir" in expected: - 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_path}") - return return_value and False + 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_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 + # 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}") + logger.info(f"{green_tag('PASS')} {file_path}") return return_value - - ret = True # ---------- INFO ---------- - #args = ArgsBuffer() - #args.input = test_file_mp4 - #test_dict = {} - #run_and_probe(cmd["name"], cmd_info, args, test_dict) - #args.input = test_file_mkv - #ret = ret and run_and_probe(cmd["name"], cmd_info, args, test_dict) + # args = ArgsBuffer() + # args.input = test_file_mp4 + # test_dict = {} + # run_and_probe(cmd["name"], cmd_info, args, test_dict) + # args.input = test_file_mkv + # ret = ret and run_and_probe(cmd["name"], cmd_info, args, test_dict) # ---------- DOWNSIZE ---------- args_test = ArgsBuffer() @@ -1274,15 +1400,26 @@ def cmd_test(args): 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) + 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_{str(cmd["name"]).replace("-","_")}_{Path(test_file).stem}{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.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) + test_dict = [ + { + "video_count": len(video_streams), + "audio_count": len(audio_streams), + "subtitle_count": len(subtitle_streams), + "video_codecs": ["h264"], + } + ] + ret = ret and run_and_probe(cmd["name"], cmd_downsize, args_test, test_dict) # ---------- REMOVE AUDIO TRACK ---------- cmd = commands["remove-audio"] @@ -1291,11 +1428,24 @@ def cmd_test(args): args_test.audio_index = 1 for test_file in test_files: - video_streams, audio_streams, subtitle_streams, cover_image = get_streams(test_file) + 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 = 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) + test_dict = [ + { + "video_count": len(video_streams), + "audio_count": 1, + "subtitle_count": len(subtitle_streams), + } + ] + 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 ---------- cmd = commands["extract-audio"] @@ -1306,24 +1456,42 @@ 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 = 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) + 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 ---------- 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) + 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_{str(cmd["name"]).replace("-","_")}_{Path(test_file).stem}{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.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) + test_dict = [ + { + "video_count": len(video_streams), + "audio_count": len(audio_streams), + "subtitle_count": len(subtitle_streams), + "video_codecs": ["h264"], + "audio_codecs": ["aac"], + } + ] + ret = ret and run_and_probe( + cmd["name"], cmd_convert_av, args_test, test_dict + ) # ---------- CONVERT AUDIO ---------- cmd = commands["convert-audio"] @@ -1331,108 +1499,207 @@ def cmd_test(args): 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) + 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_{str(cmd["name"]).replace("-","_")}_{Path(test_file).stem}{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.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) + test_dict = [ + { + "video_count": len(video_streams), + "audio_count": len(audio_streams), + "subtitle_count": len(subtitle_streams), + "audio_codecs": ["aac"], + } + ] + ret = ret and run_and_probe( + cmd["name"], cmd_convert_audio, args_test, test_dict + ) # ---------- CONVERT VIDEO ---------- cmd = commands["convert-video"] if cmd["name"] in args.choices: args_test = ArgsBuffer() for test_file in test_files: - video_streams, audio_streams, subtitle_streams, cover_image = get_streams(test_file) + 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_{str(cmd["name"]).replace("-","_")}_{Path(test_file).stem}{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.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) + test_dict = [ + { + "video_count": len(video_streams), + "audio_count": len(audio_streams), + "subtitle_count": len(subtitle_streams), + "video_codecs": ["h264"], + } + ] + ret = ret and run_and_probe( + cmd["name"], cmd_convert_video, args_test, test_dict + ) # ---------- REMOVE SUBS ---------- cmd = commands["remove-subs"] if cmd["name"] in args.choices: args_test = ArgsBuffer() for test_file in test_files: - video_streams, audio_streams, subtitle_streams, cover_image = get_streams(test_file) + 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 = 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) + test_dict = [ + { + "video_count": len(video_streams), + "audio_count": len(audio_streams), + "subtitle_count": 0, + } + ] + 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 ---------- 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) + 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_{str(cmd["name"]).replace("-","_")}_{Path(test_file).stem}{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.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) + test_dict = [ + { + "video_count": len(video_streams), + "audio_count": len(audio_streams) + 1, + "subtitle_count": len(subtitle_streams), + "audio_codecs": ["vorbis"], + } + ] + ret = ret and run_and_probe( + cmd["name"], cmd_filter_audio, args_test, test_dict + ) # ---------- CUT ---------- cmd = commands["cut"] if cmd["name"] in args.choices: args_test = ArgsBuffer() for test_file in test_files: - video_streams, audio_streams, subtitle_streams, cover_image = get_streams(test_file) + 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_{str(cmd["name"]).replace("-","_")}_{Path(test_file).stem}{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.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) + test_dict = [ + { + "video_count": len(video_streams), + "audio_count": len(audio_streams), + "subtitle_count": len(subtitle_streams), + "duration": "10s", + } + ] + ret = ret and run_and_probe( + cmd["name"], cmd_cut_segment, args_test, test_dict + ) # ---------- SET PRIMARY AUDIO ---------- cmd = commands["set-primary-audio"] if cmd["name"] in args.choices: args_test = ArgsBuffer() for test_file in test_files: - video_streams, audio_streams, subtitle_streams, cover_image = get_streams(test_file) + 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_{str(cmd["name"]).replace("-","_")}_{Path(test_file).stem}{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.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) + test_dict = [ + { + "video_count": len(video_streams), + "audio_count": len(audio_streams), + "subtitle_count": len(subtitle_streams), + } + ] + ret = ret and run_and_probe( + cmd["name"], cmd_set_primary_audio, args_test, test_dict + ) # ---------- APPLY CHAPTERS----- cmd = commands["chapters"] if cmd["name"] in args.choices: args_test = ArgsBuffer() test_file = test_file_dir / "test_auto.m4b" - video_streams, audio_streams, subtitle_streams, cover_image = get_streams(test_file) + 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_{str(cmd["name"]).replace("-","_")}_{Path(test_file).stem}{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), - "chapter_name": lines, "chapter_count": len(lines) }] - ret = ret and run_and_probe(cmd["name"], cmd_apply_chapters, args_test, test_dict) - + 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) + 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_{str(cmd["name"]).replace("-","_")}_{Path(test_file).stem}{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":{"artist":"TEST"}} ] - ret = ret and run_and_probe(cmd["name"], cmd_set_metadata, args_test, test_dict) + test_dict = [ + { + "video_count": len(video_streams), + "audio_count": len(audio_streams), + "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"] @@ -1440,18 +1707,41 @@ def cmd_test(args): 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.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) + test_dict = [{"no_russian_dir": test_dir}] + ret = ret and run_and_probe( + cmd["name"], cmd_remove_russian_from_db, args_test, test_dict + ) + + ## -----------FOLDER TEST--------------- + ## ---------- REMOVE RUSSIAN AUDIO ----- + # cmd = commands["remove-russian-from-db"] + # if cmd["name"] in args.choices: + # args_test = ArgsBuffer() + # args_test.input = test_file_dir + # 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")) else: logger.info(red_tag("SOME TESTS FAILED")) + # ---------- CLI Setup ---------- # + def add_command( subparsers, name, @@ -1494,13 +1784,17 @@ def main(): base_p = argparse.ArgumentParser(add_help=False) - base_p.add_argument("input", nargs="?", help="Input file or directory") - base_p.add_argument("-o", "--output", help="Output file") + base_p.add_argument("input", nargs="?", type=Path, help="Input file or directory") + base_p.add_argument("-o", "--output", type=Path, help="Output file") base_p.add_argument("--override", action="store_true", help="Overwrite input") base_p.add_argument("--backup", action="store_true", help="Backup original") base_p.add_argument("--loud", action="store_true", help="Show ffmpeg output") - base_p.add_argument("--kidflag", action="store_true", help="Use kid3 instead of ffmpeg") - base_p.add_argument("-r", "--recursive", action="store_true", help="Process recursively") + base_p.add_argument( + "--kidflag", action="store_true", help="Use kid3 instead of ffmpeg" + ) + base_p.add_argument( + "-r", "--recursive", action="store_true", help="Process recursively" + ) base_p.add_argument( "-ll", @@ -1771,15 +2065,12 @@ def main(): args = parser.parse_args() - logging.basicConfig( - level=LOG_LEVELS[args.log_level.upper()], - format="%(asctime)s %(levelname)s %(message)s", - ) + setup_logging(args.log_level, "ffmpeg_handling") # ---------------------------- # COMMAND VALIDATION # ---------------------------- if not hasattr(args, "func"): - logger.info(f"\nAvailable commands:") + logger.info("\nAvailable commands:") seen = set() for name, parser in sub.choices.items(): @@ -1795,40 +2086,59 @@ def main(): # re-parse subcommand properly sub_args = sub.choices[args.command].parse_args([]) - args = argparse.Namespace( **vars(sub_args)) + args = argparse.Namespace(**vars(sub_args)) # ---------------------------- # INTERACTIVE FALLBACK (ensure ALL missing values) # ---------------------------- if args.command != "test": - args.input = ensure(getattr(args, "input", None), f"Input file ({Path.cwd()})") + args.input = Path( + ensure(getattr(args, "input", None), f"Input file ({Path.cwd()})") + ) args.input = Path(args.input).expanduser().resolve() if not Path(args.input).exists(): parser.error(f"Input file {args.input} does not exist!") if commands[args.command]["output"] and not getattr(args, "output", None): input_path = Path(args.input) if args.override: - args.output = str(input_path.with_name(f"{input_path.stem}_override{input_path.suffix}")) + args.output = input_path.with_name( + f"{input_path.stem}_override{input_path.suffix}" + ) else: - args.output = ensure(getattr(args, "output", None), "Output file (enter for default file)") + args.output = ensure( + getattr(args, "output", None), + "Output file (enter for default file)", + ) if args.output == "" or args.output is None: input_path = Path(args.input) - args.output = str(input_path.with_name(f"{input_path.stem}_{args.command}{input_path.suffix}")) + args.output = input_path.with_name( + f"{input_path.stem}_{args.command}{input_path.suffix}" + ) + else: + args.output = Path(args.output) # ---------------------------- # COMMAND-SPECIFIC REQUIRED FIELDS # ---------------------------- if args.command == "convert-av": - args.preset = ensure(getattr(args, "preset", None), - "Encoding speed/quality preset (default: medium, e.g. ultrafast, fast, slow, veryslow)") - args.quality = ensure(getattr(args, "quality", None), - "Output quality level (e.g. CRF value or bitrate depending on encoder)") + args.preset = ensure( + getattr(args, "preset", None), + "Encoding speed/quality preset (default: medium, e.g. ultrafast, fast, slow, veryslow)", + ) + args.quality = ensure( + getattr(args, "quality", None), + "Output quality level (e.g. CRF value or bitrate depending on encoder)", + ) if args.command == "convert-av" or args.command == "convert-audio": - args.audio_codec = ensure(getattr(args, "audio_codec", None), - "Audio codec to use (default: aac, e.g. mp3, opus)") + args.audio_codec = ensure( + getattr(args, "audio_codec", None), + "Audio codec to use (default: aac, e.g. mp3, opus)", + ) if args.command == "convert-av" or args.command == "convert-video": - args.video_codec = ensure(getattr(args, "video_codec", None), - "Video codec to use (default: libx264, e.g. libx265, vp9)") + args.video_codec = ensure( + getattr(args, "video_codec", None), + "Video codec to use (default: libx264, e.g. libx265, vp9)", + ) if args.command == "cut": args.start = ensure(getattr(args, "start", None), "Start time (e.g. 10s, 1:00)") @@ -1839,13 +2149,20 @@ def main(): args.height = ensure(getattr(args, "height", None), "height: eg 720") args.crf = ensure(getattr(args, "crf", None), "crf: eg 23") - if args.command in ["filter-audio", "remove-audio", "extract-audio", "set-primary-audio"]: + if args.command in [ + "filter-audio", + "remove-audio", + "extract-audio", + "set-primary-audio", + ]: args.audio_index = ensure(getattr(args, "audio_index", None), "Audio index") if args.audio_index is None: args.audio_index = 1 if args.command == "chapters": - args.chapter_file = ensure(getattr(args, "chapter_file", None), "File with chapternames") + args.chapter_file = ensure( + getattr(args, "chapter_file", None), "File with chapternames" + ) if args.command == "metadata-set": args.chapter_file = ensure(getattr(args, "key", None), "Key to change") @@ -1860,32 +2177,52 @@ def main(): sys.exit(0) outputs = [] input_path = Path(args.input) + files_move_from = [] + files_move_to = [] # ---- DIRECTORY ---- if input_path.is_dir(): - counter=0 + counter = 0 + logger.info(f"Processing dir: {input_path}") 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,) + 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, ) + 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 and 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}") + logger.trace(f"Processing File: {file}") args.input = Path(file) if args.override: out_path = file.with_name(f"{file.stem}_override{file.suffix}") + files_move_from.append(out_path) + files_move_to.append(file) else: - out_path = file.with_name(f"{file.stem}_{args.command}{file.suffix}") + out_path = file.with_name( + f"{file.stem}_{args.command}{file.suffix}" + ) args.output = out_path args.file_counter = counter out = args.func(args) if out != "nooutputneeded": outputs.append(out) - counter+=1 + counter += 1 # ---- SINGLE FILE ---- elif input_path.is_file(): out = args.func(args) @@ -1895,26 +2232,28 @@ def main(): raise ValueError(f"Invalid input path: {args.input}") if any(item is None for item in outputs) or len(outputs) == 0: - logger.info(f"Output is NONE") + logger.info("Output is NONE") else: if not isinstance(outputs, list): outputs = [outputs] if args.command != "test" and outputs is not None: run_ffmpeg( - ffmpeg.merge_outputs(*outputs), - overwrite=True, - loud=args.loud + ffmpeg.merge_outputs(*outputs), overwrite=True, loud=args.loud ) - if args.override: - move_override_file(input_path, args.backup,args.loud) + if args.override and not move_override_file( + files_move_from, files_move_to, args + ): + logger.error("move_override failed") + sys.exit(1) except ffmpeg.Error as e: - logger.error(f"FFmpeg failed.\n") + logger.error("FFmpeg failed.\n") if e.stderr: logger.error(e.stderr.decode(errors="ignore")) sys.exit(1) sys.exit(0) + if __name__ == "__main__": main() diff --git a/serverlibrary/__init__.py b/serverlibrary/__init__.py index fbe2d27..a9caf26 100755 --- a/serverlibrary/__init__.py +++ b/serverlibrary/__init__.py @@ -2,6 +2,7 @@ from .core import ( TRACE, LOG_LEVELS, CustomLogger, + setup_logging, blue, combined_score, get_audio_tracks, @@ -25,6 +26,7 @@ __all__ = [ "TRACE", "LOG_LEVELS", "CustomLogger", + "setup_logging", "blue", "combined_score", "get_audio_tracks", @@ -42,4 +44,5 @@ __all__ = [ "word_similarity", "words", "yellow", -] \ No newline at end of file +] + diff --git a/serverlibrary/core.py b/serverlibrary/core.py index a44a923..e34a5c6 100755 --- a/serverlibrary/core.py +++ b/serverlibrary/core.py @@ -1,9 +1,11 @@ import json import re +import sys import shutil import subprocess from difflib import SequenceMatcher from pathlib import Path +from datetime import datetime import requests import logging @@ -12,7 +14,7 @@ import logging # ----------------------------- # Logger # ----------------------------- -TRACE=5 +TRACE = 5 red = "\033[91m" green = "\033[92m" @@ -20,11 +22,13 @@ 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, @@ -33,6 +37,35 @@ LOG_LEVELS = { "ERROR": logging.ERROR, "CRITICAL": logging.CRITICAL, } + + +def setup_logging(log_level: str = "INFO", logfile: str = "logfile") -> None: + formatter = logging.Formatter("%(asctime)s %(levelname)s %(name)s %(message)s") + + stdout_handler = logging.StreamHandler(sys.stdout) + stdout_handler.setFormatter(formatter) + stdout_handler.setLevel(LOG_LEVELS[log_level.upper()]) + + ext = Path(logfile).suffix + if not ext: + ext = ".log" + logfile_data = ( + Path("/home/lucas/logs/") / f"{logfile}_{datetime.now():%Y%m%d_%H%M%S}{ext}" + ) + + file_handler = logging.FileHandler(logfile_data, encoding="utf-8") + file_handler.setFormatter(formatter) + file_handler.setLevel(TRACE) + + root_logger = logging.getLogger() + root_logger.setLevel(TRACE) + root_logger.handlers.clear() # avoid duplicate logging + + root_logger.addHandler(stdout_handler) + root_logger.addHandler(file_handler) + root_logger.info(f"Set logger to {log_level}, Logfile: {logfile_data}") + + # ----------------------------- # Helpers # -----------------------------