876 lines
28 KiB
Python
Executable File
876 lines
28 KiB
Python
Executable File
#!/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)
|