#!/bin/python import sys import os import subprocess import json import argparse from check_video import check_video_ext, normalize_language, normalize_lang_code color = True try: from termcolor import colored except ImportError: if os.name == "posix": print("For nicer output install termcolor:\nsudo \'your installer\' python-termcolor") else: print("For nicer output install termcolor:\npip install termcolor") color = False NORMAL_STYLE = ("white", None, []) ERROR_STYLE = ("red", None, ["bold"]) WARN_STYLE = ("yellow", None, ["bold"]) INFO_STYLE = ("cyan", None, []) SUCCESS_STYLE = ("green", None, ["bold"]) DEBUG_STYLE = ("magenta", None, ["dark"]) def np(string, style, end = "\n"): if color: print(colored(string, *style), end=end) else: print(string, end=end) def human_readable_size(size, decimal_places=2): for unit in ['B','KB','MB','GB','TB']: if size < 1024: return f"{size:.{decimal_places}f} {unit}" size /= 1024 def calculate_aspect(width: int, height: int) -> str: temp = 0 def gcd(a, b): """The GCD (greatest common divisor) is the highest number that evenly divides both width and height.""" return a if b == 0 else gcd(b, a % b) if width == height: return "1:1" if width < height: temp = width width = height height = temp divisor = gcd(width, height) x = int(width / divisor) if not temp else int(height / divisor) y = int(height / divisor) if not temp else int(width / divisor) return f"{x}:{y}" def get_interlace_label(fo): if not fo: return "Progressive" fo = str(fo).lower() # Map ffprobe codes to standard labels if fo in ["tt", "tff", "tb"]: return "Interlaced (TFF)" elif fo in ["bb", "bff", "bt"]: return "Interlaced (BFF)" elif "progressive" in fo: return "Progressive" return "Progressive" # Default assumption for modern web video def fixed_width(s, width, align="left", fill=" "): s = str(s) if len(s) > width: return s[:width] if align == "right": return s.rjust(width, fill) if align == "center": return s.center(width, fill) return s.ljust(width, fill) def short_codec_name(codec): if not codec: return "" codec = codec.lower() match codec: # ---- Audio ---- case "pcm_s16le": new_codec = "PCM16" case "pcm_s24le": new_codec = "PCM24" case "pcm_s32le": new_codec = "PCM32" case "pcm_f32le": new_codec = "PCMF" case "truehd": new_codec = "THD" # ---- Video ---- case "mpeg2video": new_codec = "MPG2" case "prores": new_codec = "PRRS" case "prores_ks": new_codec = "PRRSK" # ---- Subtitles ---- case "subrip": new_codec = "SRT" case "webvtt": new_codec = "VTT" case "hdmv_pgs_subtitle": new_codec = "PGS" case "dvb_subtitle": new_codec = "DVB" case _: new_codec = codec return fixed_width(new_codec.upper(), 5) class video_lines: def __init__(self, stream, hdr_type): if stream.get("index"): self.id = stream.get("index") else: self.id = None if stream.get("name"): self.name = stream.get("name") else: self.name = "" if stream.get("name"): self.duration = seconds_to_hms(stream.get("duration")) else: self.duration = None if stream.get("codec_name"): self.codec = stream.get("codec_name") else: self.codec = "" if stream.get("width"): self.width = stream.get("width") else: self.width = "" if stream.get("height"): self.height = stream.get("height") else: self.height = "" self.resolution = f"{self.width}x{self.height}" if stream.get("r_frame_rate"): num, den = map(int, stream.get("r_frame_rate").split("/")) self.framerate = round(num / den, 2) else: self.framerate = "" if stream.get("display_aspect_ratio"): self.aspect_ratio = stream.get("display_aspect_ratio") elif self.resolution != "x": self.aspect_ratio = calculate_aspect(self.width, self.height) else: self.aspect_ratio = "" if stream.get("pix_fmt"): self.pix_fmt = stream.get("pix_fmt") else: self.pix_fmt = "" if stream.get("color_space"): self.color_space = stream.get("color_space") else: self.color_space = "" if stream.get("field_order"): self.field_order = get_interlace_label(stream.get("field_order")) else: self.field_order = "" if hdr_type: self.hdr_type = hdr_type else: self.hdr_type = "" def __str__(self): string = fixed_width("Video", 6) if self.id != None: string += f" {self.id:02d}: " else: string += f": " string += f"{short_codec_name(self.codec)}" if self.duration: string += f" {self.duration}s" if self.resolution != "x": string += f" ({self.resolution}" if self.framerate != "x": string += f"@{self.framerate}" string += f" {self.hdr_type})" elif self.hdr_type != "Unknown HDR": string += f" ({self.hdr_type})" if self.aspect_ratio: string += f" [{self.aspect_ratio}]" if self.pix_fmt and self.color_space: string += f" [{self.pix_fmt}, {self.color_space}]" if self.field_order: string += f" [{self.field_order}]" return string class audio_lines: def __init__(self, stream): # 1. Basic ID self.id = stream.get("index") # 2. Name (usually in tags as 'title') self.name = stream.get("tags", {}).get("title", "") # 3. Language (usually in tags) raw_lang = stream.get("tags", {}).get("language", "und").lower() self.language = normalize_language(raw_lang) # 4. Duration (fallback to file duration if stream duration is missing) stream_dur = stream.get("duration") if stream_dur: self.duration = seconds_to_hms(float(stream_dur)) else: self.duration = None # 5. Codec self.codec = stream.get("codec_name", "") # 6. Sample Rate (converted to kHz for readability, e.g., 48000 -> 48.0) sr = stream.get("sample_rate") self.sample_rate = f"{int(sr) / 1000} kHz" if sr else "" # 7. Channels self.channels = stream.get("channels", "") # 8. Bit Depth # PCM uses bits_per_sample; lossy like AAC/MP3 might use bits_per_raw_sample depth = stream.get("bits_per_sample") or stream.get("bits_per_raw_sample") self.bit_depth = f"{depth}-bit" if depth else "" # 9. Bitrate (converted to kbps) br = stream.get("bit_rate") self.bitrate = f"{int(br) // 1000} kbps" if br else "" def __str__(self): string = "dub" if self.id is not None: string += f" {self.id:02d}: " else: string += ": " string += f"{short_codec_name(self.codec)}" if self.language: string += f" [{self.language}]" if len(self.language) >= 14: pass elif len(self.language) >= 8: string += "\t" else: string += "\t\t" if self.duration: string += f" {self.duration}s" # Grouping audio specs: Channels, Sample Rate, and Bit Depth specs = [] if self.channels: specs.append(f"{self.channels}ch") if self.sample_rate: specs.append(self.sample_rate) if self.bit_depth: specs.append(self.bit_depth) if specs: string += f" ({', '.join(specs)})" if self.bitrate: string += f" @{self.bitrate}" if self.name: string += f" [{self.name}]" return string class subtitles: def __init__(self, stream): self.id = stream.get("index") self.name = stream.get("tags", {}).get("title", "") # Language translation raw_lang = stream.get("tags", {}).get("language", "und") self.language = normalize_language(raw_lang) # Duration logic stream_dur = stream.get("duration") self.duration = seconds_to_hms(float(stream_dur)) if stream_dur else None # Codec (e.g., srt, ass, subrip) self.codec = stream.get("codec_name", "") # Disposition (Extra helpful info for subs) dispo = stream.get("disposition", {}) self.is_forced = dispo.get("forced") == 1 self.is_default = dispo.get("default") == 1 def __str__(self): parts = [f"sub {self.id:02d}:" if self.id is not None else "Subtitle:"] if self.codec: parts.append(short_codec_name(self.codec)) if self.language: parts.append(f"[{self.language}]") if len(self.language) >= 14: pass elif len(self.language) >= 7: parts.append("\t") else: parts.append("\t\t") if self.duration: parts.append(f"{self.duration}s") # Add flags for Forced/Default flags = [] if self.is_forced: flags.append("FORCED") if self.is_default: flags.append("Default") if flags: parts.append(f"({'/'.join(flags)})") if self.name: parts.append(f"[{self.name}]") return " ".join(parts) # def get_media_info(file): # cmd = [ # "ffprobe", # "-v", "error", # "-show_entries", # ( # "format=duration:" # "stream=index,codec_type,codec_name," # "width,height,r_frame_rate,bit_rate,duration,nb_frames," # "pix_fmt,field_order,time_base,display_aspect_ratio," # "color_space,color_transfer,color_primaries,bits_per_raw_sample," # "sample_rate,channels,bits_per_sample," # "stream_disposition=forced,default:" # "stream_tags=language,title" # ), # "-of", "json", # file # ] # try: # result = subprocess.run(cmd, capture_output=True, text=True, check=True) # info = json.loads(result.stdout) # except subprocess.CalledProcessError as e: # print(f"Error running ffprobe: {e.stderr}") # return None, [], [], [] # # Container / file duration (string seconds, per ffprobe convention) # duration = None # if "format" in info: # duration = info["format"].get("duration") # video_streams = [] # audio_streams = [] # subtitle_streams = [] # for stream in info.get("streams", []): # stream_type = stream.get("codec_type") # if stream_type == "video": # video_streams.append(stream) # elif stream_type == "audio": # audio_streams.append(stream) # elif stream_type == "subtitle": # subtitle_streams.append(stream) # if duration is not None: # duration = float(duration) # else: # duration = float('nan') # return duration, video_streams, audio_streams, subtitle_streams def _side_data_hints(side_data_list): """Return set of hints from ffprobe side_data_list (case-insensitive).""" hints = set() for sd in side_data_list or []: try: t = str(sd.get("side_data_type", "")).lower() except Exception: continue if "dovi" in t or "dolby" in t or "rpu" in t: hints.add("DV") if "2094-40" in t or "hdr dynamic" in t or "hdr10+" in t: hints.add("HDR10+") if "mastering" in t or "content light" in t or "content_light" in t or "cll" in t: hints.add("HDR-static") return hints def _classify_fast(video): """Tier 0: pure-python, zero extra subprocess. Returns (label, needs_deep). Final labels need no further probing. UNKNOWN with needs_deep=True means Tier 1/2 should run. Uses SDR? for highly-probable-but-untagged SDR, HDR? for probable HDR. """ transfer = str(video.get("color_transfer") or "").lower() primaries = str(video.get("color_primaries") or "").lower() space = str(video.get("color_space") or "").lower() pix_fmt = str(video.get("pix_fmt") or "").lower() codec = str(video.get("codec_name") or "").lower() profile = str(video.get("profile") or "").lower() bits = video.get("bits_per_raw_sample") try: bits = int(bits) if bits is not None else 0 except Exception: bits = 0 hints = _side_data_hints(video.get("side_data_list")) # Cover art / thumbnails are not real video tracks — never deep-probe them. dispo = video.get("disposition") or {} try: if dispo.get("attached_pic") == 1 or codec in ("mjpeg", "png", "gif", "bmp", "tiff"): return ("SDR", False) except Exception: pass # 1. Definitive transfer function if transfer == "smpte2084": if "DV" in hints: return ("DV", False) if "HDR10+" in hints: return ("HDR10+", False) return ("HDR10", False) if transfer == "arib-std-b67": return ("HLG", False) sdr_transfers = { "bt709", "bt470m", "bt470bg", "smpte170m", "smpte240m", "gamma22", "gamma28", "iec61966-2-1", "iec61966-2-4", "srgb", } if transfer in sdr_transfers: return ("SDR", False) # 2. Primaries / matrix when transfer was stripped is_bt2020 = primaries == "bt2020" or space in ("bt2020nc", "bt2020c", "bt2020_ncl", "bt2020_cl") is_bt709 = primaries == "bt709" or space == "bt709" if is_bt2020: if "DV" in hints: return ("DV", False) if "HDR10+" in hints: return ("HDR10+", False) # Transfer missing so HDR10 vs HLG unknown — probable HDR. return ("HDR?", False) if is_bt709: return ("SDR", False) if "HDR-static" in hints or "DV" in hints or "HDR10+" in hints: if "DV" in hints: return ("DV", False) if "HDR10+" in hints: return ("HDR10+", False) return ("HDR?", False) # 3. All color tags empty — fall back to bit-depth / codec heuristics. is_10bit_plus = ("p10" in pix_fmt or "p12" in pix_fmt or "p14" in pix_fmt or "p16" in pix_fmt or bits >= 10 or "10" in profile or "12" in profile) if not transfer and not primaries and not space: if is_10bit_plus: # 10-bit SDR (anime etc.) and stripped HDR look identical here. return ("UNKNOWN", True) # 8-bit + common SDR codecs = highly probable untagged SDR. return ("SDR?", False) # Partial tags, unknown values — probable SDR, no expensive probe. if not is_10bit_plus: return ("SDR?", False) return ("UNKNOWN", True) def _classify_colors(transfer, primaries, space, hints): """Shared helper for Tier 1/2 dicts. Returns label or None.""" transfer = str(transfer or "").lower() primaries = str(primaries or "").lower() space = str(space or "").lower() if transfer == "smpte2084": if "DV" in hints: return "DV" if "HDR10+" in hints: return "HDR10+" return "HDR10" if transfer == "arib-std-b67": return "HLG" if transfer in ("bt709", "smpte170m", "smpte240m", "gamma22", "gamma28", "iec61966-2-1", "iec61966-2-4", "srgb", "bt470m", "bt470bg"): return "SDR" if primaries == "bt2020" or space in ("bt2020nc", "bt2020c", "bt2020_ncl", "bt2020_cl"): if "DV" in hints: return "DV" if "HDR10+" in hints: return "HDR10+" return "HDR?" if primaries == "bt709" or space == "bt709": return "SDR" if "DV" in hints: return "DV" if "HDR10+" in hints: return "HDR10+" if "HDR-static" in hints: return "HDR?" return None def _probe_frame_tier1(path, v_idx): """Tier 1 (cheap): 1-frame + full stream dump for one v stream. No decode.""" try: cmd = [ "ffprobe", "-v", "error", "-select_streams", f"v:{v_idx}", "-show_streams", "-show_frames", "-read_intervals", "%+#1", "-show_entries", "frame=color_space,color_transfer,color_primaries,pix_fmt,side_data_list", "-of", "json", path, ] r = subprocess.run(cmd, capture_output=True, text=True, timeout=10) if r.returncode != 0: return None info = json.loads(r.stdout or "{}") streams = info.get("streams") or [] frames = info.get("frames") or [] s0 = streams[0] if streams else {} f0 = frames[0] if frames else {} hints = _side_data_hints(s0.get("side_data_list")) | _side_data_hints(f0.get("side_data_list")) # Prefer stream tags, fall back to frame tags per field. label = _classify_colors( s0.get("color_transfer") or f0.get("color_transfer"), s0.get("color_primaries") or f0.get("color_primaries"), s0.get("color_space") or f0.get("color_space"), hints, ) return label except Exception: return None def _probe_trace_tier2(path): """Tier 2 (expensive): bitstream VUI/SEI via trace_headers, 1 frame, no decode.""" try: import shutil if shutil.which("ffmpeg") is None: return None cmd = [ "ffmpeg", "-hide_banner", "-v", "info", "-i", path, "-c:v", "copy", "-bsf:v", "trace_headers", "-frames:v", "1", "-f", "null", "-", ] r = subprocess.run(cmd, capture_output=True, text=True, timeout=15) out = ((r.stderr or "") + "\n" + (r.stdout or "")).lower() if not out.strip(): return None has_dv = "dovi" in out or "rpu" in out has_hdr10plus = "2094-40" in out or "hdr10+" in out or "dhdr10" in out has_static = "mastering" in out or "content_light" in out or "max_cll" in out # VUI transfer_characteristics: 16=PQ(ST2084), 18=HLG; primaries 9=BT.2020 import re m = re.search(r"transfer_characteristics\s*[:=]\s*(\d+)", out) trc = m.group(1) if m else "" m2 = re.search(r"colour_primaries\s*[:=]\s*(\d+)|color_primaries\s*[:=]\s*(\d+)", out) prim = (m2.group(1) or m2.group(2)) if m2 else "" if has_dv: return "DV" if has_hdr10plus: return "HDR10+" if trc == "16": return "HDR10" if trc == "18": return "HLG" if prim == "9" or has_static: return "HDR?" if trc in ("1", "6", "7"): return "SDR" return None except Exception: return None def get_media_info(file): cmd = [ "ffprobe", "-v", "error", "-show_entries", ( "format=duration:format_tags=title:" "stream=index,codec_type,codec_name,profile," "width,height,r_frame_rate,bit_rate,duration,nb_frames," "pix_fmt,field_order,time_base,display_aspect_ratio," "color_space,color_transfer,color_primaries,bits_per_raw_sample," "sample_rate,channels,bits_per_sample," "stream_disposition=forced,default,attached_pic:" "stream_tags=language,title" ), "-of", "json", file ] try: result = subprocess.run(cmd, capture_output=True, text=True, check=True) info = json.loads(result.stdout) except subprocess.CalledProcessError as e: print(f"Error running ffprobe: {e.stderr}") return None, [], [], [], None, [] # Duration duration = None if "format" in info: duration = info["format"].get("duration") # ✅ Extract container title title = None if "format" in info: title = info["format"].get("tags", {}).get("title") video_streams = [] audio_streams = [] subtitle_streams = [] for stream in info.get("streams", []): stream_type = stream.get("codec_type") if stream_type == "video": video_streams.append(stream) elif stream_type == "audio": audio_streams.append(stream) elif stream_type == "subtitle": subtitle_streams.append(stream) # Tiered HDR/SDR detection: Tier 0 fast path first, expensive only on UNKNOWN. hdr_types = [] if video_streams: trace_cache = {} # file-level Tier 2 result, run at most once per file for v_pos, video in enumerate(video_streams): label, needs_deep = _classify_fast(video) if not needs_deep: hdr_types.append(label) continue # Tier 1: 1-frame + stream dump (cheap, no decode) deep = _probe_frame_tier1(file, v_pos) if deep: hdr_types.append(deep) continue # Tier 2: bitstream trace_headers (expensive, no decode) if "trace" not in trace_cache: trace_cache["trace"] = _probe_trace_tier2(file) if trace_cache["trace"]: hdr_types.append(trace_cache["trace"]) continue hdr_types.append("UNKNOWN") if duration is not None: duration = float(duration) else: duration = float('nan') return ( duration, video_streams, audio_streams, subtitle_streams, title, hdr_types ) def seconds_to_hms(seconds): if type(seconds) is float: h = int(seconds // 3600) m = int((seconds % 3600) // 60) s = int(seconds % 60) return f"{h:02}:{m:02}:{s:02}" else: return "ERROR" def get_stream_bitrate(file_size, duration): if type(duration) is float: if duration != float('nan'): return round(float((file_size * 8)/duration/1000000), 2) if duration > 0 else 0 return float('nan') class video_file: def __init__(self, path, base_tab=""): self.base_tab = base_tab # \t self.path = path # folder/25.mkv self.name = os.path.basename(path) # 25.mkv self.size = os.path.getsize(path) self.duration, videos, audios, subs, title, hdr_types = get_media_info(path) self.sort_video_info(videos, hdr_types) self.sort_audio_info(audios) self.sort_subs_info(subs) self.bitrate = get_stream_bitrate(self.size, self.duration) self.size = human_readable_size(self.size) # 198MB self.duration = seconds_to_hms(self.duration) self.title = title def sort_video_info(self, videos, hdr_types): self.videos = [] if videos: for vl, hdr_t in zip(videos, hdr_types): video_line = video_lines(vl, hdr_t) self.videos.append(video_line) def sort_audio_info(self, audios): self.audios = [] if audios: for al in audios: self.audios.append(audio_lines(al)) def sort_subs_info(self, subs): self.subtitles = [] if subs: for st in subs: subtitle = subtitles(st) self.subtitles.append(subtitle) def print(self): if not mini: if self.base_tab == "\t": np(f"{self.base_tab}{self.name} ({self.size}, {self.duration}, {self.bitrate} MB/s):", NORMAL_STYLE) else: np(f"{os.path.dirname(self.path)}/", INFO_STYLE) np(f"\t{self.name} ({self.size}, {self.duration}, {self.bitrate} MB/s):", NORMAL_STYLE) if self.title: np(f"\t\tTitle: \"{self.title}\"", NORMAL_STYLE) if self.videos: np(f"\t Videos:", INFO_STYLE) for video in self.videos: np(f"\t\t{video}", NORMAL_STYLE) if self.audios: np(f"\t Audios:", INFO_STYLE) for audio in self.audios: np(f"\t\t{audio}", NORMAL_STYLE) if self.subtitles: np(f"\t Subtitles:", INFO_STYLE) for subtitle in self.subtitles: np(f"\t\t{subtitle}", NORMAL_STYLE) print() else: if self.base_tab == "\t": np(f"{self.base_tab}{self.name} ({self.size}, {self.duration}, {self.bitrate} MB/s):", NORMAL_STYLE) else: np(f"{os.path.dirname(self.path)}/", INFO_STYLE) np(f"\t{self.name} ({self.size}, {self.duration}, {self.bitrate} MB/s):", NORMAL_STYLE) if self.title: np(f"\t\tTitle: \"{self.title}\"", NORMAL_STYLE) print() def get_folder_info(files): np(f"Videos in {os.path.dirname(files[0])}/", INFO_STYLE) for file in files: file = video_file(file, "\t") file.print() def get_file_info(file, file_name): file = video_file(file, "") file.print() def handle_files(files, all_files): if(files != []): files.sort() grouped = [] current_dir = None current_group = [] for f in files: dir_path = os.path.dirname(f) if dir_path != current_dir: if current_group: if len(current_group) == 1: grouped.append(current_group[0]) # singleton as string else: grouped.append(current_group) # multiple files as list current_dir = dir_path current_group = [f] else: current_group.append(f) # Add the last group if current_group: if len(current_group) == 1: grouped.append(current_group[0]) else: grouped.append(current_group) all_files.extend(grouped) def handle_folders(dirs, all_files): if(dirs != []): dirs.sort(key=lambda f: os.path.dirname(f)) for dir in dirs: dir_files = [] for file in os.scandir(dir): if file.is_file(): file = file.path if check_video_ext(os.path.splitext(file)[1]): dir_files.append(file) # else: # np(f"{file} is not a compatabile Video file", WARN_STYLE) if(dir_files != []): dir_files.sort() all_files.append(dir_files) if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("-m", "--mini", action="store_true", help="Enable mini mode") parser.add_argument("paths", nargs="*", help="Files or directories to process") args = parser.parse_args() mini = args.mini # print(args) subprocess.run(["python", "/home/honney/.bin/tracker.py", "add", "ff"]) file_dir_array = [] # No paths were provided: use the current directory if not args.paths: current_dir = os.getcwd() handle_folders([os.path.abspath(current_dir)], file_dir_array) else: files = [] dirs = [] for argv in args.paths: if os.path.isfile(argv): if check_video_ext(os.path.splitext(argv)[1]): files.append(os.path.abspath(argv)) # else: # np( # f"{os.path.abspath(argv)} is not a compatible Video file", # WARN_STYLE # ) elif os.path.isdir(argv): dirs.append(os.path.abspath(argv)) else: np(f"This is not a file or directory: {argv}\nNow canceling!",ERROR_STYLE) sys.exit() handle_folders(dirs, file_dir_array) handle_files(files, file_dir_array) for element in file_dir_array: if type(element) == list: get_folder_info(element) else: file = element get_file_info(file, file)