From 22150eca966a4c77d659289723d8d0a97b496c77 Mon Sep 17 00:00:00 2001 From: Hugoved Date: Thu, 23 Apr 2026 14:25:11 +0200 Subject: [PATCH] Add files via upload --- ism_downloader.py | 508 +++++++++++++++++++++++++ ism_parser.py | 944 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 1452 insertions(+) create mode 100644 ism_downloader.py create mode 100644 ism_parser.py diff --git a/ism_downloader.py b/ism_downloader.py new file mode 100644 index 0000000..911e96c --- /dev/null +++ b/ism_downloader.py @@ -0,0 +1,508 @@ +import concurrent.futures +import importlib.machinery +import types +import re +import sys +import time +import uuid +from pathlib import Path +from urllib.parse import urlparse, urlunparse +import xml.etree.ElementTree as ET + +import requests + +try: + import inquirer +except Exception: + inquirer = None + + +DEFAULT_HEADERS = { + "accept": "*/*", + "user-agent": "Mozilla/5.0", +} + +SCRIPT_DIR = Path(__file__).resolve().parent +PARSER_PATH = SCRIPT_DIR / "ism_parser.py" + +TIMEOUT = 30 +CHUNK_SIZE = 1024 * 256 +DOWNLOAD_PROGRESS_WIDTH = 40 +PROBE_WORKERS = 16 +VIDEO_PREPROBE_LIMIT = None + +try: + if not PARSER_PATH.exists(): + raise FileNotFoundError(f"ism_parser.py was not found at: {PARSER_PATH}") + + loader = importlib.machinery.SourceFileLoader("ism_parser_module", str(PARSER_PATH)) + parser_module = types.ModuleType(loader.name) + parser_module.__file__ = str(PARSER_PATH) + loader.exec_module(parser_module) + + if not hasattr(parser_module, "write_piff_header"): + raise RuntimeError("ism_parser.py does not expose write_piff_header") + + url = input("Enter your URL: ").strip() + if not url: + raise ValueError("A manifest URL is required") + + headers = dict(DEFAULT_HEADERS) + + response = requests.get(url, headers=headers, timeout=TIMEOUT) + response.raise_for_status() + manifest_text = response.text + + xml_text = manifest_text.lstrip("\ufeff").strip() + root = ET.fromstring(xml_text) + if root.tag != "SmoothStreamingMedia": + raise ValueError("The response is not a Smooth Streaming manifest") + + protection = root.find("Protection") + pssh = None + if protection is not None: + header = protection.find("ProtectionHeader") + if header is not None and header.text is not None: + value = header.text.strip() + pssh = value or None + + tracks = [] + next_track_id = 1 + manifest_duration = int(root.get("Duration", "0")) + + for stream_index in root.findall("StreamIndex"): + media_type = (stream_index.get("Type") or "").lower() + if media_type not in {"video", "audio"}: + continue + + url_template = stream_index.get("Url") + if not url_template: + continue + + chunk_times = [] + current_time = None + + for chunk in stream_index.findall("c"): + t_attr = chunk.get("t") + d_attr = chunk.get("d") + r_attr = chunk.get("r") + + if d_attr is None: + raise ValueError("Each chunk must contain a duration 'd'") + + duration = int(d_attr) + repeat = int(r_attr) if r_attr is not None else 0 + + if t_attr is not None: + current_time = int(t_attr) + elif current_time is None: + current_time = 0 + + chunk_times.append(current_time) + + for _ in range(repeat): + current_time += duration + chunk_times.append(current_time) + + current_time += duration + + stream_name = stream_index.get("Name") or media_type + language = stream_index.get("Language") or "und" + scheme = stream_index.get("ProtectionScheme") or ("cenc" if pssh else "-") + + for quality in stream_index.findall("QualityLevel"): + codec = ( + quality.get("FourCC") + or quality.get("Codec") + or quality.get("Subtype") + or ("AAC" if media_type == "audio" else "avc1") + ) + + track = { + "track_id": next_track_id, + "media_type": media_type, + "stream_name": stream_name, + "language": language, + "url_template": url_template, + "chunk_times": chunk_times, + "bitrate": int(quality.get("Bitrate", "0") or 0), + "codec": codec, + "codec_private_data": quality.get("CodecPrivateData", "") or "", + "timescale": int(root.get("TimeScale", stream_index.get("TimeScale", "10000000"))), + "duration": manifest_duration, + "pssh": pssh, + "scheme": quality.get("ProtectionScheme") or scheme, + "width": int(quality.get("MaxWidth", stream_index.get("MaxWidth", quality.get("Width", "0")) or 0)), + "height": int(quality.get("MaxHeight", stream_index.get("MaxHeight", quality.get("Height", "0")) or 0)), + "channels": int(quality.get("Channels", "2") or 2), + "bits": int(quality.get("BitsPerSample", "16") or 16), + "sample_rate": int(quality.get("SamplingRate", "48000") or 48000), + "nal_unit_length_field": 4, + "is_drm_protected": pssh is not None, + "estimated_real_bitrate": None, + "estimated_from_fragments": False, + "probe_complete": False, + "probe_error": None, + } + tracks.append(track) + next_track_id += 1 + + video_tracks = [track for track in tracks if track["media_type"] == "video"] + if VIDEO_PREPROBE_LIMIT is not None: + video_tracks = video_tracks[:VIDEO_PREPROBE_LIMIT] + + if video_tracks: + print("[+] Measuring video variants before the menu") + total_videos = len(video_tracks) + last_probe_line_length = 0 + + for index, track in enumerate(video_tracks, start=1): + codec = str(track.get("codec") or "unknown") + if track.get("width") and track.get("height"): + size = f"{track['width']}x{track['height']}" + else: + size = "unknown size" + + probe_line = f"[+] Pre-probing videos: {index}/{total_videos} ({codec} {size})" + padding = " " * max(0, last_probe_line_length - len(probe_line)) + print("\r" + probe_line + padding, end="", flush=True) + last_probe_line_length = len(probe_line) + + chunk_times = track.get("chunk_times") or [] + if not chunk_times: + track["probe_complete"] = True + track["probe_error"] = "No chunks found" + continue + + parsed = list(urlparse(url)) + parsed[4] = "" + parsed[5] = "" + manifest_base = urlunparse(parsed) + if "/" in manifest_base: + manifest_base = manifest_base.rsplit("/", 1)[0] + "/" + + fragment_urls = [] + for start_time_value in chunk_times: + template = track["url_template"] + if "{bitrate}" in template: + template = template.replace("{bitrate}", str(track["bitrate"])) + if "{start time}" in template: + template = template.replace("{start time}", str(start_time_value)) + if "{start_time}" in template: + template = template.replace("{start_time}", str(start_time_value)) + fragment_urls.append(manifest_base + template) + + measured = 0 + total_bytes = 0 + + with requests.Session() as probe_session: + probe_session.headers.update(headers) + + with concurrent.futures.ThreadPoolExecutor(max_workers=PROBE_WORKERS) as executor: + future_map = {} + + for fragment_url in fragment_urls: + future = executor.submit( + lambda session, fragment: ( + (lambda: + ( + (lambda response: + int(response.headers.get("Content-Length")) + if response.headers.get("Content-Length", "").isdigit() + else None + )(session.head(fragment, timeout=TIMEOUT, allow_redirects=True)) + ) + )() + ), + probe_session, + fragment_url, + ) + future_map[future] = fragment_url + + for future in concurrent.futures.as_completed(future_map): + fragment_url = future_map[future] + size = 0 + + try: + value = future.result() + if value: + size = int(value) + except Exception: + size = 0 + + if size <= 0: + try: + range_response = probe_session.get( + fragment_url, + headers={"Range": "bytes=0-0"}, + stream=True, + timeout=TIMEOUT, + allow_redirects=True, + ) + range_response.raise_for_status() + content_range = range_response.headers.get("Content-Range", "") + match = re.search(r"/(\d+)$", content_range) + if match: + size = int(match.group(1)) + elif range_response.headers.get("Content-Length", "").isdigit(): + size = int(range_response.headers.get("Content-Length")) + except Exception: + size = 0 + + if size <= 0: + try: + full_response = probe_session.get( + fragment_url, + stream=True, + timeout=TIMEOUT, + allow_redirects=True, + ) + full_response.raise_for_status() + if full_response.headers.get("Content-Length", "").isdigit(): + size = int(full_response.headers.get("Content-Length")) + else: + streamed_total = 0 + for piece in full_response.iter_content(chunk_size=CHUNK_SIZE): + if piece: + streamed_total += len(piece) + size = streamed_total + except Exception: + size = 0 + + if size > 0: + measured += 1 + total_bytes += size + + track["probe_complete"] = True + + if measured > 0: + if measured == len(fragment_urls): + estimated_total_bytes = total_bytes + else: + estimated_total_bytes = (total_bytes / measured) * len(fragment_urls) + + duration_seconds = 0.0 + if int(track.get("duration") or 0) > 0 and int(track.get("timescale") or 0) > 0: + duration_seconds = int(track["duration"]) / int(track["timescale"]) + + if duration_seconds > 0: + estimated_kbps = int((estimated_total_bytes * 8) / duration_seconds / 1000) + track["estimated_real_bitrate"] = estimated_kbps + track["estimated_from_fragments"] = True + else: + track["probe_error"] = "Invalid duration/timescale" + else: + track["probe_error"] = "Could not measure fragment sizes" + + print("\r" + " " * last_probe_line_length, end="", flush=True) + print("\r[+] Video bitrate pre-probing finished") + + print(f"[+] Found {len(tracks)} downloadable track variants") + + if not tracks: + raise RuntimeError("No audio or video tracks were found in the manifest") + + def_describe_disabled = True # marker only, no defs used + + descriptions = [] + for track in tracks: + if track.get("estimated_from_fragments") and track.get("estimated_real_bitrate"): + bitrate_text = f"real≈{track['estimated_real_bitrate']} kbps" + else: + bitrate_text = f"{track['bitrate'] // 1000 if track['bitrate'] else 0} kbps" + + if track["media_type"] == "video": + size = f"{track['width']}x{track['height']}" if track["width"] and track["height"] else "unknown size" + label = f"VIDEO | {track['codec']} | {size} | {bitrate_text} | chunks={len(track['chunk_times'])}" + else: + label = ( + f"AUDIO | {track['language']} | {track['codec']} | " + f"{track['channels']}ch | {track['sample_rate']} Hz | {bitrate_text} | chunks={len(track['chunk_times'])}" + ) + descriptions.append(label) + + if inquirer is not None: + choices = [(descriptions[index], index) for index in range(len(tracks))] + answer = inquirer.prompt([inquirer.List("track_index", message="Select the track to download", choices=choices)]) + if not answer: + raise KeyboardInterrupt + selected_track = tracks[int(answer["track_index"])] + selected_description = descriptions[int(answer["track_index"])] + else: + print("[?] Select the track to download:") + for index, label in enumerate(descriptions, start=1): + prefix = ">" if index == 1 else " " + print(f" {prefix} {index:03d}. {label}") + + while True: + raw = input("Enter track number [1]: ").strip() or "1" + if raw.isdigit() and 1 <= int(raw) <= len(tracks): + selected_track = tracks[int(raw) - 1] + selected_description = descriptions[int(raw) - 1] + break + print("[!] Invalid selection") + + print(f"[+] Selected: {selected_description}") + + stem_parts = [selected_track["media_type"], str(selected_track.get("codec") or "track")] + if selected_track["media_type"] == "video": + if selected_track.get("width") and selected_track.get("height"): + stem_parts.append(f"{selected_track['width']}x{selected_track['height']}") + extension = ".mp4" + else: + stem_parts.append(selected_track.get("language") or "und") + extension = ".m4a" + + default_name = "_".join(stem_parts) + default_name = re.sub(r"[\\/:*?\"<>|]+", "_", default_name) + default_name = re.sub(r"\s+", "_", default_name).strip("._ ") + if not default_name: + default_name = f"output_{uuid.uuid4().hex[:8]}" + default_name += extension + + output_name = input(f"Output file name [{default_name}]: ").strip() or default_name + output_path = Path(output_name).expanduser().resolve() + output_path.parent.mkdir(parents=True, exist_ok=True) + + manifest_copy_name = re.sub(r"[\\/:*?\"<>|]+", "_", output_path.stem + "_manifest.xml") + manifest_copy_name = re.sub(r"\s+", "_", manifest_copy_name).strip("._ ") + if not manifest_copy_name: + manifest_copy_name = f"manifest_{uuid.uuid4().hex[:8]}.xml" + manifest_copy_path = output_path.with_name(manifest_copy_name) + manifest_copy_path.write_text(manifest_text, encoding="utf-8") + + print("Detected tracks:") + encrypted = "yes" if selected_track.get("is_drm_protected") else "no" + entry = str(selected_track.get("codec") or "unknown").lower() + handler = "vide" if selected_track["media_type"] == "video" else "soun" + scheme = "-" + if selected_track.get("is_drm_protected"): + scheme = str(selected_track.get("scheme") or "cenc") + + print( + f" track={selected_track['track_id']} " + f"handler={handler} " + f"entry={entry} " + f"encrypted={encrypted} " + f"scheme={scheme}" + ) + print(f"[+] Output: {output_path}") + + chunk_times = selected_track.get("chunk_times") or [] + if not chunk_times: + raise RuntimeError("The selected track does not contain chunks") + + parsed = list(urlparse(url)) + parsed[4] = "" + parsed[5] = "" + manifest_base = urlunparse(parsed) + if "/" in manifest_base: + manifest_base = manifest_base.rsplit("/", 1)[0] + "/" + + temp_path = output_path.with_suffix(output_path.suffix + ".part") + download_start = time.time() + + with requests.Session() as download_session: + download_session.headers.update(headers) + + first_template = selected_track["url_template"] + if "{bitrate}" in first_template: + first_template = first_template.replace("{bitrate}", str(selected_track["bitrate"])) + if "{start time}" in first_template: + first_template = first_template.replace("{start time}", str(chunk_times[0])) + if "{start_time}" in first_template: + first_template = first_template.replace("{start_time}", str(chunk_times[0])) + first_url = manifest_base + first_template + + first_response = download_session.get(first_url, timeout=TIMEOUT) + first_response.raise_for_status() + first_segment = first_response.content + + with open(temp_path, "wb") as output_file: + params = { + "track_id": selected_track["track_id"], + "duration": selected_track["duration"], + "kid": None, + "timescale": selected_track["timescale"], + "language": selected_track["language"], + "height": selected_track["height"], + "width": selected_track["width"], + "pssh": selected_track["pssh"], + "media_type": selected_track["media_type"], + "is_drm_protected": selected_track["is_drm_protected"], + "codec": str(selected_track["codec"]).lower(), + "codec_private_data": selected_track["codec_private_data"], + "channels": selected_track["channels"], + "bits": selected_track["bits"], + "sample_rate": selected_track["sample_rate"], + "nal_unit_length_field": selected_track["nal_unit_length_field"], + "bitrate": selected_track["bitrate"], + "first_segment": first_segment, + "stream_name": selected_track["stream_name"], + } + parser_module.write_piff_header(output_file, params) + + output_file.write(first_segment) + + ratio = 1.0 if len(chunk_times) <= 0 else max(0.0, min(1.0, 1 / len(chunk_times))) + filled = int(DOWNLOAD_PROGRESS_WIDTH * ratio) + bar = "■" * filled + " " * (DOWNLOAD_PROGRESS_WIDTH - filled) + elapsed = max(0.0, time.time() - download_start) + remaining = 0.0 if ratio <= 0 else max(0.0, elapsed * (1.0 - ratio) / ratio) + elapsed_s = int(round(elapsed)) + remaining_s = int(round(remaining)) + elapsed_h, elapsed_rem = divmod(elapsed_s, 3600) + elapsed_m, elapsed_sec = divmod(elapsed_rem, 60) + remaining_h, remaining_rem = divmod(remaining_s, 3600) + remaining_m, remaining_sec = divmod(remaining_rem, 60) + print( + f"[{bar}] {ratio * 100:6.2f}% (elapsed: {elapsed_h:02d}:{elapsed_m:02d}:{elapsed_sec:02d}, remaining: {remaining_h:02d}:{remaining_m:02d}:{remaining_sec:02d})", + end="\r" if len(chunk_times) > 1 else "\n", + flush=True, + ) + + for index, chunk_start in enumerate(chunk_times[1:], start=2): + template = selected_track["url_template"] + if "{bitrate}" in template: + template = template.replace("{bitrate}", str(selected_track["bitrate"])) + if "{start time}" in template: + template = template.replace("{start time}", str(chunk_start)) + if "{start_time}" in template: + template = template.replace("{start_time}", str(chunk_start)) + fragment_url = manifest_base + template + + with download_session.get(fragment_url, stream=True, timeout=TIMEOUT) as fragment_response: + fragment_response.raise_for_status() + for piece in fragment_response.iter_content(chunk_size=CHUNK_SIZE): + if piece: + output_file.write(piece) + + ratio = 1.0 if len(chunk_times) <= 0 else max(0.0, min(1.0, index / len(chunk_times))) + filled = int(DOWNLOAD_PROGRESS_WIDTH * ratio) + bar = "■" * filled + " " * (DOWNLOAD_PROGRESS_WIDTH - filled) + elapsed = max(0.0, time.time() - download_start) + remaining = 0.0 if ratio <= 0 else max(0.0, elapsed * (1.0 - ratio) / ratio) + elapsed_s = int(round(elapsed)) + remaining_s = int(round(remaining)) + elapsed_h, elapsed_rem = divmod(elapsed_s, 3600) + elapsed_m, elapsed_sec = divmod(elapsed_rem, 60) + remaining_h, remaining_rem = divmod(remaining_s, 3600) + remaining_m, remaining_sec = divmod(remaining_rem, 60) + print( + f"[{bar}] {ratio * 100:6.2f}% (elapsed: {elapsed_h:02d}:{elapsed_m:02d}:{elapsed_sec:02d}, remaining: {remaining_h:02d}:{remaining_m:02d}:{remaining_sec:02d})", + end="\n" if index >= len(chunk_times) else "\r", + flush=True, + ) + + temp_path.replace(output_path) + + print("[+] Done") + print(f"[+] Saved file: {output_path}") + print(f"[+] Saved manifest copy: {manifest_copy_path}") + +except KeyboardInterrupt: + print("\n[!] Cancelled by user", file=sys.stderr) + raise SystemExit(130) +except Exception as exc: + print(f"[!] ERROR: {exc}", file=sys.stderr) + raise SystemExit(1) \ No newline at end of file diff --git a/ism_parser.py b/ism_parser.py new file mode 100644 index 0000000..4904161 --- /dev/null +++ b/ism_parser.py @@ -0,0 +1,944 @@ +import io +import time +import base64 +import binascii +import re +import inspect +from struct import Struct + +u8 = Struct(">B") +u88 = Struct(">Bx") +u16 = Struct(">H") +u1616 = Struct(">Hxx") +u32 = Struct(">I") +u64 = Struct(">Q") + +s88 = Struct(">bx") +s16 = Struct(">h") +s1616 = Struct(">hxx") +s32 = Struct(">i") + +unity_matrix = (s32.pack(0x10000) + s32.pack(0) * 3) * 2 + s32.pack(0x40000000) + +TRACK_ENABLED = 0x1 +TRACK_IN_MOVIE = 0x2 +TRACK_IN_PREVIEW = 0x4 + +SELF_CONTAINED = 0x1 +PLAYREADY_SYSTEM_ID = "9a04f07998404286ab92e65be0885f95" +WIDEVINE_SYSTEM_ID = "edef8ba979d64acea3c827dcd51d21ed" +START_CODE = b"\x00\x00\x00\x01" + + +class BinaryWriter: + def __init__(self, stream): + self.stream = stream + + def WriteUInt(self, n, offset=0): + if isinstance(n, str): + n = n.encode("ascii") + + n = u32.pack(n) + if offset: + n = n[offset:] + + return self.stream.write(n) + + def Write(self, n): + if isinstance(n, str): + n = n.encode("ascii") + + if not isinstance(n, int): + return self.stream.write(bytes(n)) + + n = u8.pack(n) + + return self.stream.write(n) + + def WriteInt(self, n): + if isinstance(n, str): + n = n.encode("ascii") + + n = s32.pack(n) + + return self.stream.write(n) + + def WriteULong(self, n): + if isinstance(n, str): + n = n.encode("ascii") + + n = u64.pack(n) + + return self.stream.write(n) + + def WriteUShort(self, n, padding=0): + if isinstance(n, str): + n = n.encode("ascii") + + n = (u16 if not padding else u1616).pack(n) + + return self.stream.write(n) + + def WriteShort(self, n, padding=0): + if isinstance(n, str): + n = n.encode("ascii") + + n = (s16 if not padding else s1616).pack(n) + + return self.stream.write(n) + + def WriteByte(self, n, padding=0): + if isinstance(n, str): + n = n.encode("ascii") + + if not isinstance(n, int): + return self.stream.write(bytes(n)) + + n = (u8 if not padding else s88).pack(n) + + return self.stream.write(n) + + +def write_piff_header(stream, params): + track_id = int(params.get("track_id", 1) or 1) + duration = int(params["duration"]) + kid = params.get("kid") + timescale = int(params.get("timescale", 10000000) or 10000000) + language = str(params.get("language", "und") or "und").lower() + height = int(params.get("height", 0) or 0) + width = int(params.get("width", 0) or 0) + pssh = params.get("pssh") + type = params["media_type"] + is_drm_protected = bool(params.get("is_drm_protected")) + + if len(language) != 3: + language = "und" + + first_segment = params.get("first_segment") or params.get("init_segment") + if first_segment: + try: + blob = first_segment if isinstance(first_segment, (bytes, bytearray)) else bytes(first_segment) + moof_data = extract_box_data(blob, [b"moof"]) + traf_data = extract_box_data(moof_data, [b"traf"]) + tfhd_data = extract_box_data(traf_data, [b"tfhd"]) + if tfhd_data and len(tfhd_data) >= 8: + track_id = int.from_bytes(tfhd_data[4:8], byteorder="big") + except Exception: + pass + + if is_drm_protected: + normalized_kid = None + + if kid is not None: + if isinstance(kid, bytes): + if len(kid) == 16: + normalized_kid = kid.hex() + else: + try: + normalized_kid = kid.decode("ascii", errors="ignore") + except Exception: + normalized_kid = kid.hex() + else: + normalized_kid = str(kid).strip() + + normalized_kid = normalized_kid.replace("-", "").replace(" ", "") + if normalized_kid.startswith("0x"): + normalized_kid = normalized_kid[2:] + try: + normalized_kid = bytes.fromhex(normalized_kid).hex() + except Exception: + normalized_kid = None + + if normalized_kid is None and pssh: + protection_data = None + if isinstance(pssh, bytes): + protection_data = bytes(pssh) + else: + text = str(pssh).strip() + try: + protection_data = base64.b64decode(text) + except Exception: + try: + protection_data = bytes.fromhex(text.replace("-", "").replace(" ", "")) + except Exception: + protection_data = None + + if protection_data: + try: + xml_text = protection_data.decode("utf-16-le", errors="ignore") + if "" not in xml_text.upper(): + xml_text = protection_data.decode("utf-8", errors="ignore") + match = re.search(r"(.*?)", xml_text, flags=re.IGNORECASE | re.DOTALL) + if match: + kid_bytes = base64.b64decode(match.group(1)) + if len(kid_bytes) == 16: + kid_bytes = bytearray(kid_bytes) + kid_bytes[0:4] = reversed(kid_bytes[0:4]) + kid_bytes[4:6] = reversed(kid_bytes[4:6]) + kid_bytes[6:8] = reversed(kid_bytes[6:8]) + normalized_kid = bytes(kid_bytes).hex() + except Exception: + pass + + kid = normalized_kid or ("00" * 16) + else: + kid = None + + file_type_box = get_file_type_box() + stream.write(file_type_box) + + moov_payload = get_mvhd_box(timescale, duration) + trak_payload = get_tkhd_box(track_id, duration, width, height) + mdhd_payload = get_mdhd_box(language, timescale, duration) + hdlr_payload = get_hdlr_box(type) + + mdia_payload = mdhd_payload + hdlr_payload + minf_payload = get_minf_box(type) + + stbl_payload = full_box("stts", 0, 0, entry_count(0)) + stbl_payload += full_box("stsc", 0, 0, entry_count(0)) + stbl_payload += full_box("stco", 0, 0, entry_count(0)) + stbl_payload += full_box("stsz", 0, 0, entry_count(0) + entry_count(0)) + + stsd_params = dict(params) + stsd_params["track_id"] = track_id + stsd_params["kid"] = kid + stsd_payload = get_stsd_box(stsd_params, kid) + stbl_payload += full_box("stsd", 0, 0, stsd_payload) + + stbl_box = box("stbl", stbl_payload) + minf_payload += stbl_box + + minf_box = box("minf", minf_payload) + mdia_payload += minf_box + + mdia_box = box("mdia", mdia_payload) + trak_payload += mdia_box + + trak_box = box("trak", trak_payload) + moov_payload += trak_box + + mvex_payload = get_mehd_box(duration) + mvex_payload += get_trex_box(track_id) + moov_payload += box("mvex", mvex_payload) + + if is_drm_protected: + if pssh: + moov_payload += get_playready_pssh_box(pssh) + if kid: + moov_payload += get_widevine_pssh_box(kid) + + stream.write(box("moov", moov_payload)) + return stream + + +def entry_count(byte): + return u32.pack(byte) + + +def get_file_type_box(): + stream = io.BytesIO() + writer = BinaryWriter(stream) + + writer.Write(b"mp41") + writer.WriteUInt(1) + for compatible_brand in ("iso8","isom","mp41","dash","cmfc"): + writer.Write(compatible_brand.encode("ascii")) + + return box("ftyp", stream.getvalue()) + + +def get_mvhd_box(timescale, duration): + stream = io.BytesIO() + writer = BinaryWriter(stream) + + now = int(time.time()) + writer.WriteULong(now) + writer.WriteULong(now) + writer.WriteUInt(timescale) + writer.WriteULong(duration) + writer.WriteUShort(1, padding=2) + writer.WriteByte(1, padding=1) + writer.WriteUShort(0) + + for _ in range(2): + writer.WriteUInt(0) + + writer.Write(unity_matrix) + + for _ in range(6): + writer.WriteUInt(0) + + writer.WriteUInt(0xFFFFFFFF) + + return full_box("mvhd", 1, 0, stream.getvalue()) + + +def get_tkhd_box(track_id, duration, width, height): + stream = io.BytesIO() + writer = BinaryWriter(stream) + + now = int(time.time()) + writer.WriteULong(now) + writer.WriteULong(now) + writer.WriteUInt(track_id) + writer.WriteUInt(0) + writer.WriteULong(duration) + + for _ in range(2): + writer.WriteUInt(0) + + writer.WriteShort(0) + writer.WriteShort(0) + writer.WriteByte(1 if width == 0 and height == 0 else 0, padding=1) + writer.WriteUShort(0) + + writer.Write(unity_matrix) + + writer.WriteUShort(width, padding=2) + writer.WriteUShort(height, padding=2) + + return full_box( + "tkhd", + 1, + TRACK_ENABLED | TRACK_IN_MOVIE | TRACK_IN_PREVIEW, + stream.getvalue(), + ) + + +def get_mdhd_box(language, timescale, duration): + stream = io.BytesIO() + writer = BinaryWriter(stream) + + language = str(language or "und").lower() + if len(language) != 3: + language = "und" + + now = int(time.time()) + writer.WriteULong(now) + writer.WriteULong(now) + writer.WriteUInt(timescale) + writer.WriteULong(duration) + writer.WriteUShort( + ((ord(language[0]) - 0x60) << 10) + | ((ord(language[1]) - 0x60) << 5) + | (ord(language[2]) - 0x60) + ) + writer.WriteUShort(0) + + return full_box("mdhd", 1, 0, stream.getvalue()) + + +def get_hdlr_box(type): + stream = io.BytesIO() + writer = BinaryWriter(stream) + + writer.WriteUInt(0) + if type == "audio": + writer.Write(b"soun") + + for _ in range(3): + writer.WriteUInt(0) + + writer.Write(b"audio\0") + elif type == "video": + writer.Write(b"vide") + + for _ in range(3): + writer.WriteUInt(0) + + writer.Write(b"video\0") + elif type == "text": + writer.Write(b"subt") + + for _ in range(3): + writer.WriteUInt(0) + + writer.Write(b"subtitle\0") + else: + raise NotImplementedError(f"Track Type {type!r} is not supported.") + + return full_box("hdlr", 0, 0, stream.getvalue()) + + +def get_minf_box(type): + if type == "audio": + smhd_box = s88.pack(0) + smhd_box += u16.pack(0) + + minf_payload = full_box("smhd", 0, 0, bytes(smhd_box)) + elif type == "video": + vmhd_box = u16.pack(0) + + for _ in range(3): + vmhd_box += u16.pack(0) + + minf_payload = full_box("vmhd", 0, 1, bytes(vmhd_box)) + elif type == "text": + minf_payload = full_box("sthd", 0, 0, b"") + else: + raise NotImplementedError(f"Track Type {type!r} is not supported.") + + dref_payload = entry_count(1) + dref_payload += full_box("url ", 0, SELF_CONTAINED, b"") + + dinf_payload = full_box("dref", 0, 0, bytes(dref_payload)) + minf_payload += box("dinf", bytes(dinf_payload)) + + return bytes(minf_payload) + + +def get_stsd_box(params, kid): + stream = io.BytesIO() + writer = BinaryWriter(stream) + + writer.WriteUInt(1) + sample_entry_data = get_sample_entry_box(params, kid) + writer.Write(sample_entry_data) + + return stream.getvalue() + + +def get_sample_entry_box(params, kid): + stream = io.BytesIO() + writer = BinaryWriter(stream) + + for _ in range(6): + writer.WriteByte(0) + + writer.WriteUShort(1) + + codec = str(params["codec"]) + codec_lower = codec.lower() + type = params["media_type"] + cpd = params.get("codec_private_data", "") + + if isinstance(cpd, bytes): + codec_private_data = bytes(cpd) + else: + cleaned_cpd = str(cpd or "").strip().replace(" ", "").replace("-", "") + codec_private_data = binascii.unhexlify(cleaned_cpd.encode()) if cleaned_cpd else b"" + + if type == "audio": + for _ in range(2): + writer.WriteUInt(0) + + writer.WriteUShort(int(params.get("channels", 2) or 2)) + writer.WriteUShort(int(params.get("bits", 16) or 16)) + writer.WriteUShort(0) + writer.WriteUShort(0) + writer.WriteUShort(int(params.get("sample_rate", 48000) or 48000), padding=2) + + if "aac" in codec_lower: + if codec_private_data: + esds_box = get_esds_box( + int(params.get("track_id", 1) or 1), + int(params.get("bitrate", 0) or 0), + codec_private_data, + ) + writer.Write(esds_box) + + if params.get("is_drm_protected"): + sinf_box = get_sinf_box(kid, "mp4a") + writer.Write(sinf_box) + return box("enca", stream.getvalue()) + else: + return box("mp4a", stream.getvalue()) + elif codec_lower in ("e-ac3", "ec-3", "ec3"): + dec3_payload = params.get("dec3_payload") or params.get("dec3") + if dec3_payload is not None: + if isinstance(dec3_payload, str): + dec3_payload = bytes.fromhex(dec3_payload.replace(" ", "").replace("-", "")) + writer.Write(full_box("dec3", 0, 0, bytes(dec3_payload))) + else: + atmos = bool( + params.get("is_atmos") + or params.get("joc") + or params.get("dolby_atmos") + or "atmos" in str(params.get("profile", "")).lower() + or "joc" in str(params.get("profile", "")).lower() + ) + writer.Write(get_dec3_box(atmos=atmos)) + + if params.get("is_drm_protected"): + sinf_box = get_sinf_box(kid, "ec-3") + writer.Write(sinf_box) + return box("enca", stream.getvalue()) + else: + return box("ec-3", stream.getvalue()) + else: + raise NotImplementedError(f"Audio Codec {codec!r} is not supported.") + elif type == "video": + writer.WriteUShort(0) + writer.WriteUShort(0) + + for _ in range(3): + writer.WriteUInt(0) + + writer.WriteUShort(int(params.get("width", 0) or 0)) + writer.WriteUShort(int(params.get("height", 0) or 0)) + writer.WriteUShort(0x48, padding=2) + writer.WriteUShort(0x48, padding=2) + writer.WriteUInt(0) + writer.WriteUShort(1) + + for _ in range(32): + writer.WriteByte(0) + + writer.WriteUShort(0x18) + writer.WriteShort(-1) + + nal_units = [n for n in codec_private_data.split(START_CODE) if n] if codec_private_data and START_CODE in codec_private_data else [] + + if codec_lower == "avc1": + sps = next((n for n in nal_units if n and (n[0] & 0x1F) == 7), None) + pps = next((n for n in nal_units if n and (n[0] & 0x1F) == 8), None) + if sps is None or pps is None: + raise ValueError("Missing SPS or PPS in AVC codec private data") + + avcc_box = get_avcc_box( + int(params.get("nal_unit_length_field", 4) or 4), sps, pps + ) + writer.Write(avcc_box) + + if params.get("is_drm_protected"): + frma_codec = str(params.get("frma_codec") or params.get("sinf_codec") or "avc1").lower() + sinf_box = get_sinf_box(kid, frma_codec) + writer.Write(sinf_box) + return box("encv", stream.getvalue()) + else: + return box("avc1", stream.getvalue()) + + elif codec_lower in ("hvc1", "hev1"): + vps = next((n for n in nal_units if n and (((n[0] >> 1) & 0x3F) == 32)), None) + sps = next((n for n in nal_units if n and (((n[0] >> 1) & 0x3F) == 33)), None) + pps = next((n for n in nal_units if n and (((n[0] >> 1) & 0x3F) == 34)), None) + if not (vps and sps and pps): + raise ValueError("Missing VPS, SPS, or PPS in HEVC codec private data") + + hvcc_box = get_hvcc_box( + int(params.get("nal_unit_length_field", 4) or 4), sps, pps, vps + ) + writer.Write(hvcc_box) + + if params.get("is_drm_protected"): + frma_codec = str(params.get("frma_codec") or params.get("sinf_codec") or codec_lower).lower() + sinf_box = get_sinf_box(kid, frma_codec) + writer.Write(sinf_box) + return box("encv", stream.getvalue()) + else: + return box(codec_lower, stream.getvalue()) + + elif codec_lower in ("dvhe", "dvh1"): + vps = next((n for n in nal_units if n and (((n[0] >> 1) & 0x3F) == 32)), None) + sps = next((n for n in nal_units if n and (((n[0] >> 1) & 0x3F) == 33)), None) + pps = next((n for n in nal_units if n and (((n[0] >> 1) & 0x3F) == 34)), None) + + if not (vps and sps and pps): + raise ValueError("Missing VPS, SPS, or PPS in codec private data") + + hvcc_box = get_hvcc_box( + int(params.get("nal_unit_length_field", 4) or 4), sps, pps, vps + ) + dvcc_box = generate_hvcc_dvcc_box() + + writer.Write(hvcc_box) + writer.Write(dvcc_box) + + clli_box = params.get("clli_box") + clli_payload = params.get("clli_payload") + if clli_box is not None: + if isinstance(clli_box, str): + clli_box = bytes.fromhex(clli_box.replace(" ", "").replace("-", "")) + writer.Write(bytes(clli_box)) + elif clli_payload is not None: + if isinstance(clli_payload, str): + clli_payload = bytes.fromhex(clli_payload.replace(" ", "").replace("-", "")) + writer.Write(box("clli", bytes(clli_payload))) + + mdcv_box = params.get("mdcv_box") + mdcv_payload = params.get("mdcv_payload") + if mdcv_box is not None: + if isinstance(mdcv_box, str): + mdcv_box = bytes.fromhex(mdcv_box.replace(" ", "").replace("-", "")) + writer.Write(bytes(mdcv_box)) + elif mdcv_payload is not None: + if isinstance(mdcv_payload, str): + mdcv_payload = bytes.fromhex(mdcv_payload.replace(" ", "").replace("-", "")) + writer.Write(box("mdcv", bytes(mdcv_payload))) + + bitrate = int(params.get("bitrate", 0) or 0) + if bitrate > 0: + btrt_stream = io.BytesIO() + btrt_writer = BinaryWriter(btrt_stream) + btrt_writer.WriteUInt(bitrate) + btrt_writer.WriteUInt(bitrate) + btrt_writer.WriteUInt(bitrate) + writer.Write(box("btrt", btrt_stream.getvalue())) + + if params.get("is_drm_protected"): + default_frma = "hvc1" if codec_lower == "dvhe" else codec_lower + frma_codec = str(params.get("frma_codec") or params.get("sinf_codec") or default_frma).lower() + sinf_box = get_sinf_box(kid, frma_codec) + writer.Write(sinf_box) + return box("encv", stream.getvalue()) + else: + return box(codec_lower, stream.getvalue()) + else: + raise NotImplementedError(f"Video Codec {codec!r} is not supported.") + + elif type == "text": + if codec == "TTML": + writer.Write("http://www.w3.org/ns/ttml\0") + writer.Write("\0") + writer.Write("\0") + + return box("stpp", stream.getvalue()) + else: + raise NotImplementedError(f"Subtitle Codec {codec!r} is not supported.") + else: + raise NotImplementedError(f"Track Type {type!r} is not supported.") + + +def get_mehd_box(duration): + stream = io.BytesIO() + writer = BinaryWriter(stream) + + writer.WriteULong(duration) + + return full_box("mehd", 1, 0, stream.getvalue()) + + +def get_trex_box(track_id): + stream = io.BytesIO() + writer = BinaryWriter(stream) + + writer.WriteUInt(track_id) + writer.WriteUInt(1) + writer.WriteUInt(0) + writer.WriteUInt(0) + writer.WriteUInt(0) + + return full_box("trex", 0, 0, stream.getvalue()) + + +def extract_box_data(data, box_sequence): + data_reader = io.BytesIO(data) + while True: + header = data_reader.read(8) + if len(header) < 8: + raise ValueError(f"Could not find box path: {box_sequence!r}") + + box_size = u32.unpack(header[:4])[0] + box_type = header[4:8] + + if box_size == 1: + largesize = data_reader.read(8) + if len(largesize) < 8: + raise ValueError("Invalid large-size MP4 box") + box_size = int.from_bytes(largesize, byteorder="big") + header_size = 16 + else: + header_size = 8 + + if box_size < header_size: + raise ValueError("Invalid MP4 box size") + + payload_size = box_size - header_size + + if box_type == box_sequence[0]: + box_data = data_reader.read(payload_size) + if len(box_sequence) == 1: + return box_data + return extract_box_data(box_data, box_sequence[1:]) + + data_reader.seek(payload_size, 1) + + +def box(box_type, payload): + stream = io.BytesIO() + writer = BinaryWriter(stream) + + writer.WriteUInt(8 + len(payload)) + writer.Write(box_type.encode("ascii")) + writer.Write(payload) + + return stream.getvalue() + + +def full_box(box_type, version, flags, payload): + stream = io.BytesIO() + writer = BinaryWriter(stream) + + writer.Write(version) + writer.WriteUInt(flags, offset=1) + + return box(box_type, stream.getvalue() + payload) + + +def get_avcc_box(nal_unit_length_field, sps: bytes, pps: bytes): + stream = io.BytesIO() + writer = BinaryWriter(stream) + + writer.WriteByte(1) + writer.Write(sps[1:4]) + writer.WriteByte(0xFC | (nal_unit_length_field - 1)) + writer.WriteByte(1) + writer.WriteUShort(len(sps)) + writer.Write(sps) + writer.WriteByte(1) + writer.WriteUShort(len(pps)) + writer.Write(pps) + + return box("avcC", stream.getvalue()) + +def get_dec3_box(atmos=False): + stream = io.BytesIO() + writer = BinaryWriter(stream) + + payload = b"\x0e\x00\x20\x0f\x00\x01\x10" if atmos else b"\x14\x00\x20\x0f\x00\x00\x00" + writer.Write(payload) + + return full_box("dec3", 0, 0, stream.getvalue()) + +def generate_hvcc_dvcc_box(): + stream = io.BytesIO() + writer = BinaryWriter(stream) + + try: + frame = inspect.currentframe().f_back + params = frame.f_locals.get("params", {}) if frame else {} + except Exception: + params = {} + + raw_box = params.get("dvcc_box") or params.get("dvcC_box") or params.get("dvvC_box") + if raw_box is not None: + if isinstance(raw_box, str): + raw_box = bytes.fromhex(raw_box.replace(" ", "").replace("-", "")) + raw_box = bytes(raw_box) + if len(raw_box) >= 8 and raw_box[4:8] in (b"dvcC", b"dvvC"): + return raw_box + + payload = params.get("dvcc_payload") or params.get("dvcC_payload") or params.get("dvvC_payload") + if payload is not None: + if isinstance(payload, str): + payload = bytes.fromhex(payload.replace(" ", "").replace("-", "")) + payload = bytes(payload) + box_type = str(params.get("dv_box_type") or params.get("dv_config_box_type") or "dvcC") + return box(box_type, payload) + + dv_codec = str(params.get("dv_codec") or params.get("codec_string") or params.get("codec") or "") + match = re.search(r"(?:dvhe|dvh1|dva1|dvav)\.(\d{2})\.(\d{2})", dv_codec, flags=re.IGNORECASE) + + profile = int(params.get("dv_profile", 5) or 5) + level = params.get("dv_level") + if match: + profile = int(match.group(1)) + if level is None: + level = int(match.group(2)) + + width = int(params.get("width", 0) or 0) + height = int(params.get("height", 0) or 0) + + if level is None: + if width > 0 and height > 0: + if width <= 720 and height <= 576: + level = 1 + elif width <= 1280 and height <= 720: + level = 3 + elif width <= 1920 and height <= 1080: + level = 4 + elif width <= 2560 and height <= 1440: + level = 5 + elif width <= 3840 and height <= 2160: + level = 6 + else: + level = 7 + else: + level = 6 + + level = int(level) + + rpu_present = 1 if params.get("dv_rpu_present", 1) else 0 + el_present = 1 if params.get("dv_el_present", 0) else 0 + bl_present = 1 if params.get("dv_bl_present", 1) else 0 + compatibility_id = int(params.get("dv_compatibility_id", 0) or 0) & 0x0F + dv_version_major = int(params.get("dv_version_major", 1) or 1) & 0xFF + dv_version_minor = int(params.get("dv_version_minor", 0) or 0) & 0xFF + + payload = bytearray(24) + payload[0] = dv_version_major + payload[1] = dv_version_minor + payload[2] = ((profile & 0x7F) << 1) | ((level >> 5) & 0x01) + payload[3] = ((level & 0x1F) << 3) | ((rpu_present & 0x01) << 2) | ((el_present & 0x01) << 1) | (bl_present & 0x01) + payload[4] = (compatibility_id & 0x0F) << 4 + + writer.Write(payload) + + default_box_type = "dvvC" if profile >= 8 else "dvcC" + box_type = str(params.get("dv_box_type") or params.get("dv_config_box_type") or default_box_type) + + return box(box_type, stream.getvalue()) + +def get_hvcc_box(nal_unit_length_field, sps: bytes, pps: bytes, vps: bytes): + ori_sps = bytes(sps) + enc_list = bytearray() + + with io.BytesIO(sps) as reader: + while reader.tell() < len(sps): + byte_value = reader.read(1) + if not byte_value: + break + enc_list.extend(byte_value) + if len(enc_list) >= 3 and enc_list[-3:] == bytes([0x00, 0x00, 0x03]): + enc_list.pop() + + sps = bytes(enc_list) + + with io.BytesIO(sps) as reader: + reader.read(2) + first_byte = reader.read(1)[0] + max_sub_layers_minus1 = (first_byte & 0x0E) >> 1 + next_byte = reader.read(1)[0] + general_profile_space = (next_byte & 0xC0) >> 6 + general_tier_flag = (next_byte & 0x20) >> 5 + general_profile_idc = next_byte & 0x1F + general_profile_compatibility_flags = int.from_bytes(reader.read(4), byteorder="big") + constraint_bytes = bytearray(reader.read(6)) + general_level_idc = reader.read(1)[0] + + stream = io.BytesIO() + writer = BinaryWriter(stream) + + writer.WriteByte(1) + writer.WriteByte( + (general_profile_space << 6) + + (0x20 if general_tier_flag == 1 else 0) + + general_profile_idc + ) + writer.WriteUInt( + general_profile_compatibility_flags + ) + writer.Write(constraint_bytes) + writer.WriteByte(general_level_idc) + writer.WriteUShort(0xF000) + writer.WriteByte(0xFC) + writer.WriteByte(0xFC) + writer.WriteByte(0xF8) + writer.WriteByte(0xF8) + writer.WriteUShort(0) + writer.WriteByte( + (0 << 6) | (min(max_sub_layers_minus1, 7) << 3) | (0 << 2) | (nal_unit_length_field - 1) + ) + writer.WriteByte(0x03) + + writer.WriteByte(0x20) + writer.WriteUShort(1) + writer.WriteUShort(len(vps)) + writer.Write(vps) + writer.WriteByte(0x21) + writer.WriteUShort(1) + writer.WriteUShort(len(ori_sps)) + writer.Write(ori_sps) + writer.WriteByte(0x22) + writer.WriteUShort(1) + writer.WriteUShort(len(pps)) + writer.Write(pps) + + return box("hvcC", stream.getvalue()) + +def get_esds_box(track_id, bitrate, codec_private_data: bytes): + stream = io.BytesIO() + writer = BinaryWriter(stream) + + writer.WriteByte(0x03) + writer.WriteByte(20 + len(codec_private_data)) + writer.WriteByte((track_id & 0xFF00) >> 8) + writer.WriteByte(track_id & 0x00FF) + writer.WriteByte(0) + + writer.WriteByte(0x04) + writer.WriteByte(15 + len(codec_private_data)) + writer.WriteByte(0x40) + writer.WriteByte((0x05 << 2) | (0 << 1) | 1) + + for _ in range(3): + writer.WriteByte(0xFF) + + writer.WriteByte((bitrate & 0xFF000000) >> 24) + writer.WriteByte((bitrate & 0x00FF0000) >> 16) + writer.WriteByte((bitrate & 0x0000FF00) >> 8) + writer.WriteByte(bitrate & 0x000000FF) + writer.WriteByte((bitrate & 0xFF000000) >> 24) + writer.WriteByte((bitrate & 0x00FF0000) >> 16) + writer.WriteByte((bitrate & 0x0000FF00) >> 8) + writer.WriteByte(bitrate & 0x000000FF) + + writer.WriteByte(0x05) + writer.WriteByte(len(codec_private_data)) + writer.Write(codec_private_data) + + return full_box("esds", 0, 0, stream.getvalue()) + + +def get_sinf_box(key_id, codec): + key_id = str(key_id or "").replace("-", "").replace(" ", "") + if key_id.startswith("0x"): + key_id = key_id[2:] + key_id = bytes.fromhex(key_id) + frmaBox = box("frma", codec.encode("ascii")) + + sinfPayload = bytearray() + sinfPayload.extend(frmaBox) + + schmPayload = bytearray() + schmPayload.extend(b"cenc") + + schmPayload.extend( + bytes([0, 1, 0, 0]) + ) + schmBox = full_box("schm", 0, 0, bytes(schmPayload)) + + sinfPayload.extend(schmBox) + + tencPayload = bytearray() + tencPayload.extend(bytes([0, 0])) + tencPayload.append(TRACK_ENABLED) + tencPayload.append(0x8) + tencPayload.extend(key_id) + tencBox = full_box("tenc", 0, 0, bytes(tencPayload)) + + schiBox = box("schi", tencBox) + sinfPayload.extend(schiBox) + + return box("sinf", bytes(sinfPayload)) + + +def get_playready_pssh_box(pssh): + stream = io.BytesIO() + writer = BinaryWriter(stream) + + protection_data = bytes.fromhex(base64.b64decode(pssh).hex()) + + sys_id_data = bytes.fromhex(PLAYREADY_SYSTEM_ID) + pssh_data = protection_data + + writer.Write(sys_id_data) + writer.WriteUInt(len(pssh_data)) + writer.Write(pssh_data) + + return full_box("pssh", 0, 0, stream.getvalue()) + + +def get_widevine_pssh_box(key_id): + stream = io.BytesIO() + writer = BinaryWriter(stream) + + sys_id_data = bytes.fromhex(WIDEVINE_SYSTEM_ID) + pssh_data = bytes.fromhex(f"08011210{key_id}1A046E647265220400000000") + + writer.Write(sys_id_data) + writer.WriteUInt(len(pssh_data)) + writer.Write(pssh_data) + + return full_box("pssh", 0, 0, stream.getvalue()) \ No newline at end of file