mirror of
https://github.com/Hugoved/ism-downloader.git
synced 2026-07-15 18:50:03 +02:00
Add files via upload
This commit is contained in:
Hugoved
@@ -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 <c> 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)
|
||||
+944
@@ -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 "<KID>" not in xml_text.upper():
|
||||
xml_text = protection_data.decode("utf-8", errors="ignore")
|
||||
match = re.search(r"<KID>(.*?)</KID>", 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())
|
||||
Reference in New Issue
Block a user