pymp4dump/pymp4dump.py

from __future__ import annotations
import argparse
import base64
import binascii
import json
import os
import struct
import sys
from dataclasses import dataclass, field
from typing import Any, Dict, Iterable, List, Optional, Tuple

VERSION = "1.0"
BANNER = "MP4 File Dumper - Python Version %s" % VERSION
MAX_SCAN_BOXES = 1_000_000
MAX_PREVIEW_BYTES = 64
CONTAINER_BOXES = {
    "moov", "trak", "mdia", "minf", "dinf", "stbl", "edts", "udta", "meta", "ilst",
    "moof", "traf", "mfra", "skip", "wide", "ipro", "sinf", "schi", "tref", "iref",
    "meco", "mere", "strk", "stri", "strd", "sbgp", "sgpd", "hnti", "hinf", "avcn",
    "avss", "wave", "gmhd", "clip", "matt", "rmra", "rmda", "rmdb", "cmov", "cmvd",
    "dref", "paen", "font", "vttc", "vtte", "vtta", "vttC", "stpp", "uuid",
}
FULL_BOX_TYPES = {
    "mvhd", "tkhd", "mdhd", "hdlr", "vmhd", "smhd", "hmhd", "nmhd", "dref", "stsd", "stts", "ctts",
    "cslg", "stsc", "stsz", "stz2", "stco", "co64", "stss", "stsh", "stdp", "sdtp", "stdp", "padb",
    "elst", "mehd", "trex", "mfhd", "tfhd", "trun", "tfdt", "mfra", "tfra", "mfro", "pdin", "ftyp",
    "mvex", "url ", "urn ", "meta", "pitm", "iloc", "iinf", "infe", "iprp", "ipco", "ipma", "ipro",
    "sidx", "ssix", "pssh", "tenc", "schm", "frma", "saiz", "saio", "senc", "sgpd", "sbgp", "prft",
}
HANDLER_NAMES = {
    "vide": "Video Track", "soun": "Audio Track", "hint": "Hint Track", "meta": "Metadata Track",
    "text": "Text Track", "subt": "Subtitle Track", "sbtl": "Subtitle Track", "clcp": "Closed Caption",
    "mdir": "Metadata", "tmcd": "Timecode Track", "auxv": "Auxiliary Video", "pict": "Picture Track",
}
PROTECTION_SYSTEM_IDS = {
    "edef8ba979d64acea3c827dcd51d21ed": "Widevine",
    "9a04f07998404286ab92e65be0885f95": "PlayReady",
    "94ce86fb07ff4f43adb893d2fa968ca2": "FairPlay",
    "1077efecc0b24d02ace33c1e52e2fb4b": "Common PSSH",
}

class Mp4DumpError(Exception):
    pass

class Reader:
    def __init__(self, data: bytes, base_offset: int = 0) -> None:
        self.data = data
        self.pos = 0
        self.base_offset = base_offset

    def remaining(self) -> int:
        return len(self.data) - self.pos

    def tell(self) -> int:
        return self.pos

    def seek(self, pos: int) -> None:
        if pos < 0 or pos > len(self.data):
            raise Mp4DumpError("seek outside buffer")
        self.pos = pos

    def skip(self, size: int) -> None:
        self.seek(self.pos + size)

    def read(self, size: int) -> bytes:
        if size < 0 or self.pos + size > len(self.data):
            raise Mp4DumpError("unexpected end of data")
        out = self.data[self.pos:self.pos + size]
        self.pos += size
        return out

    def u8(self) -> int:
        return self.read(1)[0]

    def i8(self) -> int:
        return struct.unpack(">b", self.read(1))[0]

    def u16(self) -> int:
        return struct.unpack(">H", self.read(2))[0]

    def i16(self) -> int:
        return struct.unpack(">h", self.read(2))[0]

    def u24(self) -> int:
        b = self.read(3)
        return (b[0] << 16) | (b[1] << 8) | b[2]

    def u32(self) -> int:
        return struct.unpack(">I", self.read(4))[0]

    def i32(self) -> int:
        return struct.unpack(">i", self.read(4))[0]

    def u64(self) -> int:
        return struct.unpack(">Q", self.read(8))[0]

    def i64(self) -> int:
        return struct.unpack(">q", self.read(8))[0]

    def fixed16_16(self) -> float:
        return self.i32() / 65536.0

    def fixed8_8(self) -> float:
        return self.i16() / 256.0

    def cstring(self) -> str:
        start = self.pos
        while self.pos < len(self.data) and self.data[self.pos] != 0:
            self.pos += 1
        raw = self.data[start:self.pos]
        if self.pos < len(self.data):
            self.pos += 1
        return decode_text(raw)


def fourcc(raw: bytes) -> str:
    return "".join(chr(b) if 32 <= b <= 126 else "." for b in raw)


def decode_text(raw: bytes) -> str:
    for enc in ("utf-8", "utf-16-be", "latin-1"):
        try:
            return raw.decode(enc).rstrip("\x00")
        except UnicodeDecodeError:
            pass
    return raw.hex()


def hex_preview(raw: bytes, limit: int = MAX_PREVIEW_BYTES) -> str:
    if len(raw) <= limit:
        return raw.hex()
    return raw[:limit].hex() + "..."


def language_from_code(code: int) -> str:
    chars = []
    for shift in (10, 5, 0):
        chars.append(chr(((code >> shift) & 0x1F) + 0x60))
    return "".join(chars)


def qt_time(seconds_since_1904: int) -> str:
    # Keep numeric output deterministic; avoid timezone/date dependencies.
    return str(seconds_since_1904)

@dataclass
class Box:
    type: str
    size: int
    start: int
    header_size: int
    payload_start: int
    payload_size: int
    uuid: Optional[str] = None
    version: Optional[int] = None
    flags: Optional[int] = None
    fields: Dict[str, Any] = field(default_factory=dict)
    children: List["Box"] = field(default_factory=list)
    error: Optional[str] = None

    @property
    def end(self) -> int:
        return self.start + self.size

    def to_json(self) -> Dict[str, Any]:
        obj: Dict[str, Any] = {
            "name": self.type,
            "type": self.type,
            "offset": self.start,
            "size": self.size,
            "header_size": self.header_size,
        }
        if self.uuid:
            obj["uuid"] = self.uuid
        if self.version is not None:
            obj["version"] = self.version
        if self.flags is not None:
            obj["flags"] = self.flags
        if self.fields:
            obj.update(self.fields)
        if self.error:
            obj["error"] = self.error
        if self.children:
            obj["children"] = [child.to_json() for child in self.children]
        return obj

class Mp4Parser:
    def __init__(self, data: bytes, verbosity: int = 0) -> None:
        self.data = data
        self.verbosity = max(0, min(3, verbosity))
        self.track_context: Dict[int, Dict[str, Any]] = {}
        self.current_track_id: Optional[int] = None
        self.current_handler: Optional[str] = None
        self.current_timescale: Optional[int] = None

    def parse(self) -> List[Box]:
        return self._parse_children(0, len(self.data), None)

    def _parse_children(self, start: int, end: int, parent_type: Optional[str]) -> List[Box]:
        boxes: List[Box] = []
        pos = start
        count = 0
        while pos + 8 <= end and count < MAX_SCAN_BOXES:
            count += 1
            box = self._parse_box(pos, end, parent_type)
            boxes.append(box)
            if box.size <= 0:
                break
            next_pos = box.end
            if next_pos <= pos:
                break
            pos = next_pos
        return boxes

    def _parse_box(self, start: int, parent_end: int, parent_type: Optional[str]) -> Box:
        r = Reader(self.data[start:parent_end], base_offset=start)
        try:
            small_size = r.u32()
            typ = fourcc(r.read(4))
            header_size = 8
            size = small_size
            if small_size == 1:
                size = r.u64()
                header_size = 16
            elif small_size == 0:
                size = parent_end - start
            uuid_value = None
            if typ == "uuid":
                uuid_value = r.read(16).hex()
                header_size += 16
            if size < header_size:
                size = header_size
            if start + size > parent_end:
                size = parent_end - start
            payload_start = start + header_size
            payload_size = max(0, size - header_size)
            box = Box(typ, size, start, header_size, payload_start, payload_size, uuid_value)
            self._decode_box(box, parent_type)
            return box
        except Exception as exc:
            return Box("????", max(8, parent_end - start), start, 8, start + 8, max(0, parent_end - start - 8), error=str(exc))

    def _decode_box(self, box: Box, parent_type: Optional[str]) -> None:
        payload = self.data[box.payload_start:box.payload_start + box.payload_size]
        r = Reader(payload, base_offset=box.payload_start)
        saved_track = self.current_track_id
        saved_handler = self.current_handler
        saved_timescale = self.current_timescale
        try:
            decoder = getattr(self, "decode_" + box.type.strip().replace(" ", "_"), None)
            if box.type == "url ":
                decoder = self.decode_url
            elif box.type == "urn ":
                decoder = self.decode_urn
            elif box.type == "©nam":
                decoder = self.decode_data_container
            if decoder:
                decoder(box, r)
            elif self._looks_like_full_box(box.type, payload):
                self._read_full_box(box, r)
            if self._is_container(box, parent_type):
                child_start = box.payload_start + self._container_payload_skip(box)
                if child_start < box.end:
                    box.children = self._parse_children(child_start, box.end, box.type)
        except Exception as exc:
            box.error = str(exc)
        finally:
            if box.type == "trak":
                self.current_track_id = saved_track
                self.current_handler = saved_handler
                self.current_timescale = saved_timescale

    def _looks_like_full_box(self, typ: str, payload: bytes) -> bool:
        return typ in FULL_BOX_TYPES and len(payload) >= 4

    def _read_full_box(self, box: Box, r: Reader) -> None:
        box.version = r.u8()
        box.flags = r.u24()

    def _is_container(self, box: Box, parent_type: Optional[str]) -> bool:
        if box.type in {"mdat", "free", "ftyp", "styp", "sidx", "ssix", "pssh", "emsg"}:
            return False
        if box.type in CONTAINER_BOXES:
            return True
        if parent_type in {"stsd", "ipco", "ilst", "wave", "sinf", "schi", "dref"}:
            return box.payload_size >= 8
        return False

    def _container_payload_skip(self, box: Box) -> int:
        if box.type == "meta":
            return 4 if box.version is not None else 0
        if box.type in {"stsd", "dref", "ipro"}:
            return 8
        return 0

    def _register(self, key: str, value: Any) -> None:
        if self.current_track_id is None:
            return
        ctx = self.track_context.setdefault(self.current_track_id, {})
        ctx[key] = value

    def decode_ftyp(self, box: Box, r: Reader) -> None:
        box.fields["major_brand"] = fourcc(r.read(4))
        box.fields["minor_version"] = r.u32()
        brands = []
        while r.remaining() >= 4:
            brands.append(fourcc(r.read(4)))
        box.fields["compatible_brands"] = brands

    decode_styp = decode_ftyp

    def decode_mvhd(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if box.version == 1:
            creation, modification, timescale, duration = r.u64(), r.u64(), r.u32(), r.u64()
        else:
            creation, modification, timescale, duration = r.u32(), r.u32(), r.u32(), r.u32()
        box.fields.update({"creation_time": qt_time(creation), "modification_time": qt_time(modification), "timescale": timescale, "duration": duration})
        if r.remaining() >= 8:
            box.fields["duration_seconds"] = duration / timescale if timescale else None
            box.fields["rate"] = r.fixed16_16()
            box.fields["volume"] = r.fixed8_8()
        if r.remaining() >= 10:
            r.skip(10)
        if r.remaining() >= 36:
            box.fields["matrix"] = [r.i32() for _ in range(9)]
        if r.remaining() >= 24:
            r.skip(24)
        if r.remaining() >= 4:
            box.fields["next_track_id"] = r.u32()

    def decode_tkhd(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if box.version == 1:
            creation, modification, track_id, reserved, duration = r.u64(), r.u64(), r.u32(), r.u32(), r.u64()
        else:
            creation, modification, track_id, reserved, duration = r.u32(), r.u32(), r.u32(), r.u32(), r.u32()
        self.current_track_id = track_id
        self.track_context.setdefault(track_id, {"id": track_id})
        box.fields.update({"creation_time": qt_time(creation), "modification_time": qt_time(modification), "track_id": track_id, "duration": duration})
        if r.remaining() >= 8:
            r.skip(8)
        if r.remaining() >= 8:
            box.fields["layer"] = r.i16(); box.fields["alternate_group"] = r.i16(); box.fields["volume"] = r.fixed8_8(); r.skip(2)
        if r.remaining() >= 36:
            box.fields["matrix"] = [r.i32() for _ in range(9)]
        if r.remaining() >= 8:
            box.fields["width"] = r.fixed16_16(); box.fields["height"] = r.fixed16_16()

    def decode_mdhd(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if box.version == 1:
            creation, modification, timescale, duration = r.u64(), r.u64(), r.u32(), r.u64()
        else:
            creation, modification, timescale, duration = r.u32(), r.u32(), r.u32(), r.u32()
        lang = language_from_code(r.u16()) if r.remaining() >= 2 else "und"
        if r.remaining() >= 2:
            r.u16()
        self.current_timescale = timescale
        self._register("timescale", timescale)
        box.fields.update({"creation_time": qt_time(creation), "modification_time": qt_time(modification), "timescale": timescale, "duration": duration, "language": lang})
        box.fields["duration_seconds"] = duration / timescale if timescale else None

    def decode_hdlr(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if r.remaining() >= 4:
            box.fields["pre_defined"] = r.u32()
        handler = fourcc(r.read(4)) if r.remaining() >= 4 else ""
        self.current_handler = handler
        self._register("handler", handler)
        box.fields["handler_type"] = handler
        box.fields["handler_name"] = HANDLER_NAMES.get(handler, "")
        if r.remaining() >= 12:
            r.skip(12)
        if r.remaining() > 0:
            box.fields["name"] = r.cstring()

    def decode_vmhd(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if r.remaining() >= 8:
            box.fields["graphics_mode"] = r.u16(); box.fields["op_color"] = [r.u16(), r.u16(), r.u16()]

    def decode_smhd(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if r.remaining() >= 4:
            box.fields["balance"] = r.fixed8_8(); r.skip(2)

    def decode_hmhd(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if r.remaining() >= 8:
            box.fields.update({"max_pdu_size": r.u16(), "avg_pdu_size": r.u16(), "max_bitrate": r.u32(), "avg_bitrate": r.u32()})

    def decode_nmhd(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)

    def decode_stsd(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if r.remaining() >= 4:
            box.fields["entry_count"] = r.u32()

    def _decode_sample_entry_common(self, box: Box, r: Reader) -> None:
        if r.remaining() >= 6:
            r.skip(6)
        if r.remaining() >= 2:
            box.fields["data_reference_index"] = r.u16()

    def decode_avc1(self, box: Box, r: Reader) -> None: self._decode_visual_sample_entry(box, r)
    def decode_avc3(self, box: Box, r: Reader) -> None: self._decode_visual_sample_entry(box, r)
    def decode_hev1(self, box: Box, r: Reader) -> None: self._decode_visual_sample_entry(box, r)
    def decode_hvc1(self, box: Box, r: Reader) -> None: self._decode_visual_sample_entry(box, r)
    def decode_dvhe(self, box: Box, r: Reader) -> None: self._decode_visual_sample_entry(box, r)
    def decode_dvh1(self, box: Box, r: Reader) -> None: self._decode_visual_sample_entry(box, r)
    def decode_mp4v(self, box: Box, r: Reader) -> None: self._decode_visual_sample_entry(box, r)
    def decode_encv(self, box: Box, r: Reader) -> None: self._decode_visual_sample_entry(box, r)

    def _decode_visual_sample_entry(self, box: Box, r: Reader) -> None:
        self._decode_sample_entry_common(box, r)
        if r.remaining() >= 16:
            r.skip(16)
        if r.remaining() >= 4:
            box.fields["width"] = r.u16(); box.fields["height"] = r.u16()
        if r.remaining() >= 14:
            box.fields["horiz_resolution"] = r.fixed16_16(); box.fields["vert_resolution"] = r.fixed16_16(); r.skip(4); box.fields["frame_count"] = r.u16()
        if r.remaining() >= 32:
            compressor = r.read(32)
            box.fields["compressor_name"] = decode_text(compressor[1:1 + min(compressor[0], 31)]) if compressor else ""
        if r.remaining() >= 4:
            box.fields["depth"] = r.u16(); box.fields["pre_defined_3"] = r.i16()
        consumed = r.tell()
        child_start = box.payload_start + consumed
        if child_start < box.end:
            box.children = self._parse_children(child_start, box.end, box.type)

    def decode_mp4a(self, box: Box, r: Reader) -> None: self._decode_audio_sample_entry(box, r)
    def decode_enca(self, box: Box, r: Reader) -> None: self._decode_audio_sample_entry(box, r)
    def decode_alac(self, box: Box, r: Reader) -> None: self._decode_audio_sample_entry(box, r)
    def decode_ac_3(self, box: Box, r: Reader) -> None: self._decode_audio_sample_entry(box, r)
    def decode_ec_3(self, box: Box, r: Reader) -> None: self._decode_audio_sample_entry(box, r)

    def _decode_audio_sample_entry(self, box: Box, r: Reader) -> None:
        self._decode_sample_entry_common(box, r)
        if r.remaining() >= 8:
            r.skip(8)
        if r.remaining() >= 8:
            box.fields["channel_count"] = r.u16(); box.fields["sample_size"] = r.u16(); r.skip(4)
        if r.remaining() >= 4:
            rate = r.u32()
            box.fields["sample_rate"] = rate >> 16
        consumed = r.tell()
        child_start = box.payload_start + consumed
        if child_start < box.end:
            box.children = self._parse_children(child_start, box.end, box.type)

    def decode_esds(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if r.remaining() > 0:
            box.fields["descriptor_data"] = hex_preview(r.read(r.remaining()))

    def decode_avcC(self, box: Box, r: Reader) -> None:
        if r.remaining() >= 7:
            box.fields["configuration_version"] = r.u8(); box.fields["avc_profile_indication"] = r.u8(); box.fields["profile_compatibility"] = r.u8(); box.fields["avc_level_indication"] = r.u8()
            length_size_minus_one = r.u8() & 3
            box.fields["nal_unit_length_size"] = length_size_minus_one + 1
            sps_count = r.u8() & 31
            sps = []
            for _ in range(sps_count):
                if r.remaining() < 2: break
                n = r.u16(); sps.append(r.read(min(n, r.remaining())).hex())
            pps_count = r.u8() if r.remaining() >= 1 else 0
            pps = []
            for _ in range(pps_count):
                if r.remaining() < 2: break
                n = r.u16(); pps.append(r.read(min(n, r.remaining())).hex())
            box.fields["sequence_parameter_sets"] = sps
            box.fields["picture_parameter_sets"] = pps

    def decode_hvcC(self, box: Box, r: Reader) -> None:
        if r.remaining() >= 23:
            box.fields["configuration_version"] = r.u8()
            byte = r.u8(); box.fields["general_profile_space"] = (byte >> 6) & 3; box.fields["general_tier_flag"] = (byte >> 5) & 1; box.fields["general_profile_idc"] = byte & 31
            box.fields["general_profile_compatibility_flags"] = r.u32()
            box.fields["general_constraint_indicator_flags"] = r.read(6).hex()
            box.fields["general_level_idc"] = r.u8()
            r.skip(12)
            arrays = r.u8() if r.remaining() else 0
            box.fields["num_arrays"] = arrays
            if self.verbosity >= 2:
                parsed = []
                for _ in range(arrays):
                    if r.remaining() < 3: break
                    hdr = r.u8(); count = r.u16(); units = []
                    for _ in range(count):
                        if r.remaining() < 2: break
                        n = r.u16(); units.append(r.read(min(n, r.remaining())).hex())
                    parsed.append({"nal_unit_type": hdr & 0x3F, "nal_units": units})
                box.fields["arrays"] = parsed

    def decode_dvcC(self, box: Box, r: Reader) -> None: self._decode_dovi(box, r)
    def decode_dvvC(self, box: Box, r: Reader) -> None: self._decode_dovi(box, r)

    def _decode_dovi(self, box: Box, r: Reader) -> None:
        if r.remaining() >= 4:
            box.fields["dv_version_major"] = r.u8(); box.fields["dv_version_minor"] = r.u8()
            value = r.u16()
            box.fields["dv_profile"] = (value >> 9) & 0x7F
            box.fields["dv_level"] = (value >> 3) & 0x3F
            box.fields["rpu_present_flag"] = (value >> 2) & 1
            box.fields["el_present_flag"] = (value >> 1) & 1
            box.fields["bl_present_flag"] = value & 1

    def decode_pasp(self, box: Box, r: Reader) -> None:
        if r.remaining() >= 8:
            box.fields["h_spacing"] = r.u32(); box.fields["v_spacing"] = r.u32()

    def decode_clli(self, box: Box, r: Reader) -> None:
        if r.remaining() >= 4:
            box.fields["max_content_light_level"] = r.u16(); box.fields["max_pic_average_light_level"] = r.u16()

    def decode_mdcv(self, box: Box, r: Reader) -> None:
        if r.remaining() >= 24:
            box.fields["display_primaries"] = [[r.u16(), r.u16()] for _ in range(3)]
            box.fields["white_point"] = [r.u16(), r.u16()]
            box.fields["max_display_mastering_luminance"] = r.u32()
            box.fields["min_display_mastering_luminance"] = r.u32()

    def decode_btrt(self, box: Box, r: Reader) -> None:
        if r.remaining() >= 12:
            box.fields["buffer_size_db"] = r.u32(); box.fields["max_bitrate"] = r.u32(); box.fields["avg_bitrate"] = r.u32()

    def decode_colr(self, box: Box, r: Reader) -> None:
        if r.remaining() >= 4:
            colour_type = fourcc(r.read(4)); box.fields["colour_type"] = colour_type
            if colour_type in {"nclx", "nclc"} and r.remaining() >= 6:
                box.fields["colour_primaries"] = r.u16(); box.fields["transfer_characteristics"] = r.u16(); box.fields["matrix_coefficients"] = r.u16()
                if colour_type == "nclx" and r.remaining() >= 1:
                    box.fields["full_range_flag"] = (r.u8() >> 7) & 1
            elif r.remaining() > 0:
                box.fields["payload"] = hex_preview(r.read(r.remaining()))

    def decode_stts(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r); count = r.u32() if r.remaining() >= 4 else 0; entries = []
        for _ in range(count):
            if r.remaining() < 8: break
            entries.append({"sample_count": r.u32(), "sample_delta": r.u32()})
        box.fields["entry_count"] = count
        if self.verbosity >= 1: box.fields["entries"] = entries
        self._register("stts", entries)

    def decode_ctts(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r); count = r.u32() if r.remaining() >= 4 else 0; entries = []
        for _ in range(count):
            if r.remaining() < 8: break
            offset = r.i32() if box.version == 1 else r.u32()
            entries.append({"sample_count": r.u32(), "sample_offset": offset})
        box.fields["entry_count"] = count
        if self.verbosity >= 1: box.fields["entries"] = entries

    def decode_stsc(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r); count = r.u32() if r.remaining() >= 4 else 0; entries = []
        for _ in range(count):
            if r.remaining() < 12: break
            entries.append({"first_chunk": r.u32(), "samples_per_chunk": r.u32(), "sample_description_index": r.u32()})
        box.fields["entry_count"] = count
        if self.verbosity >= 1: box.fields["entries"] = entries
        self._register("stsc", entries)

    def decode_stsz(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        sample_size = r.u32() if r.remaining() >= 4 else 0
        sample_count = r.u32() if r.remaining() >= 4 else 0
        sizes = []
        if sample_size == 0:
            for _ in range(sample_count):
                if r.remaining() < 4: break
                sizes.append(r.u32())
        else:
            sizes = [sample_size] * sample_count
        box.fields["sample_size"] = sample_size; box.fields["sample_count"] = sample_count
        if self.verbosity >= 1: box.fields["entries"] = sizes
        self._register("stsz", sizes)

    def decode_stz2(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if r.remaining() >= 4: r.skip(3); field_size = r.u8()
        else: field_size = 0
        sample_count = r.u32() if r.remaining() >= 4 else 0
        sizes = []
        if field_size == 4:
            for _ in range((sample_count + 1) // 2):
                if r.remaining() < 1: break
                b = r.u8(); sizes.extend([(b >> 4) & 15, b & 15])
            sizes = sizes[:sample_count]
        elif field_size == 8:
            sizes = [r.u8() for _ in range(min(sample_count, r.remaining()))]
        elif field_size == 16:
            for _ in range(sample_count):
                if r.remaining() < 2: break
                sizes.append(r.u16())
        box.fields["field_size"] = field_size; box.fields["sample_count"] = sample_count
        if self.verbosity >= 1: box.fields["entries"] = sizes
        self._register("stsz", sizes)

    def decode_stco(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r); count = r.u32() if r.remaining() >= 4 else 0; entries = []
        for _ in range(count):
            if r.remaining() < 4: break
            entries.append(r.u32())
        box.fields["entry_count"] = count
        if self.verbosity >= 1: box.fields["entries"] = entries
        self._register("chunk_offsets", entries)

    def decode_co64(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r); count = r.u32() if r.remaining() >= 4 else 0; entries = []
        for _ in range(count):
            if r.remaining() < 8: break
            entries.append(r.u64())
        box.fields["entry_count"] = count
        if self.verbosity >= 1: box.fields["entries"] = entries
        self._register("chunk_offsets", entries)

    def decode_stss(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r); count = r.u32() if r.remaining() >= 4 else 0; entries = []
        for _ in range(count):
            if r.remaining() < 4: break
            entries.append(r.u32())
        box.fields["entry_count"] = count
        if self.verbosity >= 1: box.fields["entries"] = entries

    def decode_sdtp(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if self.verbosity >= 2:
            box.fields["sample_dependency_table"] = list(r.read(r.remaining()))
        else:
            box.fields["entry_count"] = r.remaining()

    def decode_elst(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r); count = r.u32() if r.remaining() >= 4 else 0; entries = []
        for _ in range(count):
            if box.version == 1:
                if r.remaining() < 20: break
                duration, media_time = r.u64(), r.i64()
            else:
                if r.remaining() < 12: break
                duration, media_time = r.u32(), r.i32()
            entries.append({"segment_duration": duration, "media_time": media_time, "media_rate": r.fixed16_16()})
        box.fields["entry_count"] = count
        if self.verbosity >= 1: box.fields["entries"] = entries

    def decode_url(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if r.remaining() > 0:
            box.fields["location"] = r.cstring()

    def decode_urn(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        box.fields["name"] = r.cstring() if r.remaining() else ""
        box.fields["location"] = r.cstring() if r.remaining() else ""

    def decode_dref(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        box.fields["entry_count"] = r.u32() if r.remaining() >= 4 else 0

    def decode_mfhd(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if r.remaining() >= 4: box.fields["sequence_number"] = r.u32()

    def decode_tfhd(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        track_id = r.u32() if r.remaining() >= 4 else 0
        self.current_track_id = track_id
        flags = box.flags or 0
        box.fields["track_id"] = track_id
        if flags & 0x000001 and r.remaining() >= 8: box.fields["base_data_offset"] = r.u64()
        if flags & 0x000002 and r.remaining() >= 4: box.fields["sample_description_index"] = r.u32()
        if flags & 0x000008 and r.remaining() >= 4: box.fields["default_sample_duration"] = r.u32()
        if flags & 0x000010 and r.remaining() >= 4: box.fields["default_sample_size"] = r.u32()
        if flags & 0x000020 and r.remaining() >= 4: box.fields["default_sample_flags"] = r.u32()

    def decode_tfdt(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if box.version == 1 and r.remaining() >= 8: box.fields["base_media_decode_time"] = r.u64()
        elif r.remaining() >= 4: box.fields["base_media_decode_time"] = r.u32()

    def decode_trun(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        flags = box.flags or 0
        sample_count = r.u32() if r.remaining() >= 4 else 0
        box.fields["sample_count"] = sample_count
        if flags & 0x000001 and r.remaining() >= 4: box.fields["data_offset"] = r.i32()
        if flags & 0x000004 and r.remaining() >= 4: box.fields["first_sample_flags"] = r.u32()
        entries = []
        for _ in range(sample_count):
            entry: Dict[str, Any] = {}
            if flags & 0x000100 and r.remaining() >= 4: entry["sample_duration"] = r.u32()
            if flags & 0x000200 and r.remaining() >= 4: entry["sample_size"] = r.u32()
            if flags & 0x000400 and r.remaining() >= 4: entry["sample_flags"] = r.u32()
            if flags & 0x000800 and r.remaining() >= 4: entry["sample_composition_time_offset"] = r.i32() if box.version == 1 else r.u32()
            entries.append(entry)
        if self.verbosity >= 1: box.fields["entries"] = entries

    def decode_trex(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if r.remaining() >= 20:
            box.fields.update({"track_id": r.u32(), "default_sample_description_index": r.u32(), "default_sample_duration": r.u32(), "default_sample_size": r.u32(), "default_sample_flags": r.u32()})

    def decode_mehd(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if box.version == 1 and r.remaining() >= 8: box.fields["fragment_duration"] = r.u64()
        elif r.remaining() >= 4: box.fields["fragment_duration"] = r.u32()

    def decode_sidx(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if r.remaining() < 8: return
        box.fields["reference_id"] = r.u32(); box.fields["timescale"] = r.u32()
        if box.version == 1:
            if r.remaining() < 16: return
            box.fields["earliest_presentation_time"] = r.u64(); box.fields["first_offset"] = r.u64()
        else:
            if r.remaining() < 8: return
            box.fields["earliest_presentation_time"] = r.u32(); box.fields["first_offset"] = r.u32()
        if r.remaining() >= 2: r.u16()
        count = r.u16() if r.remaining() >= 2 else 0
        refs = []
        for _ in range(count):
            if r.remaining() < 12: break
            a = r.u32(); b = r.u32(); c = r.u32()
            refs.append({"reference_type": (a >> 31) & 1, "referenced_size": a & 0x7fffffff, "subsegment_duration": b, "starts_with_sap": (c >> 31) & 1, "sap_type": (c >> 28) & 7, "sap_delta_time": c & 0x0fffffff})
        box.fields["reference_count"] = count
        if self.verbosity >= 1: box.fields["references"] = refs

    def decode_pssh(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if r.remaining() < 20: return
        system_id = r.read(16).hex()
        box.fields["system_id"] = system_id
        box.fields["system_name"] = PROTECTION_SYSTEM_IDS.get(system_id, "")
        if box.version and box.version > 0 and r.remaining() >= 4:
            kid_count = r.u32(); kids = []
            for _ in range(kid_count):
                if r.remaining() < 16: break
                kids.append(r.read(16).hex())
            box.fields["kid_count"] = kid_count; box.fields["kids"] = kids
        if r.remaining() >= 4:
            data_size = r.u32(); data = r.read(min(data_size, r.remaining()))
            box.fields["data_size"] = data_size
            if self.verbosity >= 1:
                box.fields["data"] = data.hex()
                box.fields["data_base64"] = base64.b64encode(data).decode("ascii")

    def decode_tenc(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if r.remaining() >= 2:
            box.fields["reserved"] = r.u8()
            second = r.u8()
            if box.version and box.version > 0:
                box.fields["default_crypt_byte_block"] = (second >> 4) & 0x0F
                box.fields["default_skip_byte_block"] = second & 0x0F
            else:
                box.fields["reserved_2"] = second
        if r.remaining() >= 1: box.fields["default_is_protected"] = r.u8()
        if r.remaining() >= 1: box.fields["default_per_sample_iv_size"] = r.u8()
        if r.remaining() >= 16: box.fields["default_kid"] = r.read(16).hex()
        if box.fields.get("default_is_protected") == 1 and box.fields.get("default_per_sample_iv_size") == 0 and r.remaining() >= 1:
            const_iv_size = r.u8(); box.fields["default_constant_iv_size"] = const_iv_size
            if r.remaining() >= const_iv_size: box.fields["default_constant_iv"] = r.read(const_iv_size).hex()

    def decode_frma(self, box: Box, r: Reader) -> None:
        if r.remaining() >= 4: box.fields["data_format"] = fourcc(r.read(4))

    def decode_schm(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if r.remaining() >= 8:
            box.fields["scheme_type"] = fourcc(r.read(4)); box.fields["scheme_version"] = r.u32()
        if (box.flags or 0) & 1 and r.remaining() > 0:
            box.fields["scheme_uri"] = r.cstring()

    def decode_saiz(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if (box.flags or 0) & 1 and r.remaining() >= 8:
            box.fields["aux_info_type"] = fourcc(r.read(4)); box.fields["aux_info_type_parameter"] = r.u32()
        if r.remaining() >= 5:
            default_size = r.u8(); sample_count = r.u32()
            box.fields["default_sample_info_size"] = default_size; box.fields["sample_count"] = sample_count
            if default_size == 0 and self.verbosity >= 1:
                box.fields["sample_info_sizes"] = list(r.read(min(sample_count, r.remaining())))

    def decode_saio(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if (box.flags or 0) & 1 and r.remaining() >= 8:
            box.fields["aux_info_type"] = fourcc(r.read(4)); box.fields["aux_info_type_parameter"] = r.u32()
        count = r.u32() if r.remaining() >= 4 else 0; offsets = []
        for _ in range(count):
            need = 8 if box.version == 1 else 4
            if r.remaining() < need: break
            offsets.append(r.u64() if box.version == 1 else r.u32())
        box.fields["entry_count"] = count
        if self.verbosity >= 1: box.fields["offsets"] = offsets

    def decode_senc(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        count = r.u32() if r.remaining() >= 4 else 0
        box.fields["sample_count"] = count
        if self.verbosity >= 2:
            box.fields["raw"] = hex_preview(r.read(r.remaining()), 256)

    def decode_prft(self, box: Box, r: Reader) -> None:
        self._read_full_box(box, r)
        if r.remaining() >= 12:
            box.fields["reference_track_id"] = r.u32(); box.fields["ntp_timestamp"] = r.u64()
            if box.version == 1 and r.remaining() >= 8: box.fields["media_time"] = r.i64()
            elif r.remaining() >= 4: box.fields["media_time"] = r.i32()

    def decode_uuid(self, box: Box, r: Reader) -> None:
        if self.verbosity >= 2 and r.remaining() > 0:
            box.fields["payload"] = hex_preview(r.read(r.remaining()), 256)

    def decode_data(self, box: Box, r: Reader) -> None:
        if r.remaining() >= 8:
            data_type = r.u32(); locale = r.u32(); payload = r.read(r.remaining())
            box.fields["data_type"] = data_type; box.fields["locale"] = locale
            if data_type in {1, 2}:
                box.fields["value"] = decode_text(payload)
            elif data_type in {13, 14, 27}:
                box.fields["value"] = payload.hex()
            else:
                box.fields["value"] = hex_preview(payload)

    def decode_data_container(self, box: Box, r: Reader) -> None:
        if box.payload_size >= 8:
            box.children = self._parse_children(box.payload_start, box.end, box.type)

    def decode_mdat(self, box: Box, r: Reader) -> None:
        box.fields["data_size"] = box.payload_size
        if self.verbosity >= 3 and box.payload_size:
            box.fields["data_preview"] = hex_preview(r.read(min(box.payload_size, MAX_PREVIEW_BYTES)))

    def decode_free(self, box: Box, r: Reader) -> None:
        box.fields["data_size"] = box.payload_size

    decode_skip = decode_free

class TextPrinter:
    def __init__(self, verbosity: int = 0) -> None:
        self.verbosity = verbosity

    def print(self, boxes: List[Box]) -> None:
        for box in boxes:
            self._print_box(box, 0)

    def _print_box(self, box: Box, depth: int) -> None:
        indent = "  " * depth
        extras = []
        if box.uuid: extras.append("uuid=%s" % box.uuid)
        if box.version is not None: extras.append("version=%d" % box.version)
        if box.flags is not None: extras.append("flags=0x%06x" % box.flags)
        detail = " " + " ".join(extras) if extras else ""
        print(f"{indent}[{box.type}] size={box.size} offset={box.start}{detail}")
        if box.error:
            print(f"{indent}  error = {box.error}")
        for key, value in box.fields.items():
            if self.verbosity == 0 and key in {"entries", "data", "data_base64", "payload", "sample_info_sizes", "offsets", "raw"}:
                continue
            self._print_field(key, value, depth + 1)
        for child in box.children:
            self._print_box(child, depth + 1)

    def _print_field(self, key: str, value: Any, depth: int) -> None:
        indent = "  " * depth
        if isinstance(value, list) and len(value) > 16 and self.verbosity < 2:
            print(f"{indent}{key} = [{len(value)} entries]")
            return
        text = json.dumps(value, ensure_ascii=False) if isinstance(value, (dict, list)) else str(value)
        if len(text) > 240 and self.verbosity < 3:
            text = text[:240] + "..."
        print(f"{indent}{key} = {text}")

class TrackDumper:
    def __init__(self, data: bytes, parser: Mp4Parser) -> None:
        self.data = data
        self.parser = parser

    def dump_track(self, input_filename: str, track_id: int) -> str:
        ctx = self.parser.track_context.get(track_id)
        if not ctx:
            raise Mp4DumpError("track not found: %d" % track_id)
        sizes = ctx.get("stsz") or []
        offsets = ctx.get("chunk_offsets") or []
        stsc = ctx.get("stsc") or []
        if not sizes or not offsets or not stsc:
            raise Mp4DumpError("track %d cannot be dumped because sample tables are incomplete or fragmented-only" % track_id)
        sample_offsets = self._build_sample_offsets(sizes, offsets, stsc)
        if len(sample_offsets) != len(sizes):
            raise Mp4DumpError("track %d sample table mismatch" % track_id)
        output_name = f"{input_filename}.{track_id}"
        with open(output_name, "wb") as out:
            for off, size in zip(sample_offsets, sizes):
                if off < 0 or off + size > len(self.data):
                    raise Mp4DumpError("track %d sample offset outside file" % track_id)
                out.write(struct.pack(">I", size))
                out.write(self.data[off:off + size])
        return output_name

    def _build_sample_offsets(self, sizes: List[int], chunk_offsets: List[int], stsc: List[Dict[str, int]]) -> List[int]:
        sample_offsets: List[int] = []
        sample_index = 0
        entries = sorted(stsc, key=lambda e: e.get("first_chunk", 0))
        for entry_index, entry in enumerate(entries):
            first_chunk = entry.get("first_chunk", 1)
            next_first = entries[entry_index + 1].get("first_chunk", len(chunk_offsets) + 1) if entry_index + 1 < len(entries) else len(chunk_offsets) + 1
            samples_per_chunk = entry.get("samples_per_chunk", 0)
            for chunk_number in range(first_chunk, next_first):
                if chunk_number < 1 or chunk_number > len(chunk_offsets):
                    continue
                offset = chunk_offsets[chunk_number - 1]
                for _ in range(samples_per_chunk):
                    if sample_index >= len(sizes):
                        break
                    sample_offsets.append(offset)
                    offset += sizes[sample_index]
                    sample_index += 1
        return sample_offsets

def parse_track_option(value: str) -> Tuple[int, Optional[str]]:
    if ":" in value:
        tid, key = value.split(":", 1)
    else:
        tid, key = value, None
    try:
        track_id = int(tid, 10)
    except ValueError as exc:
        raise argparse.ArgumentTypeError("invalid track id") from exc
    if key:
        clean = key.replace("-", "").replace(" ", "")
        if len(clean) != 32:
            raise argparse.ArgumentTypeError("track keys must be 128-bit hex values")
        try:
            binascii.unhexlify(clean)
        except binascii.Error as exc:
            raise argparse.ArgumentTypeError("invalid track key hex") from exc
    return track_id, key

def load_file(path: str) -> bytes:
    with open(path, "rb") as f:
        return f.read()

PLAYREADY_SYSTEM_ID = "9a04f07998404286ab92e65be0885f95"
WIDEVINE_SYSTEM_ID = "edef8ba979d64acea3c827dcd51d21ed"

def normalize_system_id(system_id: str) -> str:
    return system_id.lower().replace("-", "").strip()


def walk_boxes(boxes: List[Box]) -> Iterable[Box]:
    for box in boxes:
        yield box
        if box.children:
            yield from walk_boxes(box.children)


def parse_mp4_boxes(input_file: str, verbosity: int = 1) -> List[Box]:
    if not os.path.isfile(input_file):
        raise Mp4DumpError("input file does not exist: %s" % input_file)
    data = load_file(input_file)
    parser = Mp4Parser(data, verbosity=verbosity)
    return parser.parse()


def extract_pssh_boxes_from_mp4(input_file: str, system_id: Optional[str] = None) -> List[Dict[str, Any]]:
    wanted = normalize_system_id(system_id) if system_id else None
    if not os.path.isfile(input_file):
        raise Mp4DumpError("input file does not exist: %s" % input_file)

    data = load_file(input_file)
    parser = Mp4Parser(data, verbosity=1)
    boxes = parser.parse()
    results: List[Dict[str, Any]] = []

    for box in walk_boxes(boxes):
        if box.type != "pssh":
            continue

        found_system_id = str(box.fields.get("system_id", "")).lower()
        if wanted and found_system_id != wanted:
            continue

        full_box = data[box.start:box.end]
        item: Dict[str, Any] = {
            "system_id": found_system_id,
            "system_name": box.fields.get("system_name", ""),
            "offset": box.start,
            "size": box.size,
            "pssh_base64": base64.b64encode(full_box).decode("ascii"),
            "pssh_hex": full_box.hex(),
            "data_size": box.fields.get("data_size"),
            "data_base64": box.fields.get("data_base64"),
            "data_hex": box.fields.get("data"),
        }
        if "kids" in box.fields:
            item["kids"] = box.fields.get("kids")
        if "kid_count" in box.fields:
            item["kid_count"] = box.fields.get("kid_count")
        results.append(item)

    return results


def extract_playready_pssh_from_mp4(input_file: str) -> Optional[str]:
    matches = extract_pssh_boxes_from_mp4(input_file, PLAYREADY_SYSTEM_ID)
    if not matches:
        return None
    value = matches[0].get("pssh_base64")
    return str(value) if value is not None else None


def extract_playready_pssh_payload_from_mp4(input_file: str) -> Optional[str]:
    matches = extract_pssh_boxes_from_mp4(input_file, PLAYREADY_SYSTEM_ID)
    if not matches:
        return None
    value = matches[0].get("data_base64")
    return str(value) if value is not None else None


def extract_playready_pssh_from_init(input_file: str) -> Optional[str]:
    return extract_playready_pssh_from_mp4(input_file)


def extract_playready_pssh_payload_from_init(input_file: str) -> Optional[str]:
    return extract_playready_pssh_payload_from_mp4(input_file)

def build_arg_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        prog="pymp4dump.py",
        description="Python MP4/QuickTime atom dumper inspired by Bento4 mp4dump.",
        formatter_class=argparse.RawTextHelpFormatter,
    )
    p.add_argument("input", help="input MP4/QuickTime/ISO BMFF file")
    p.add_argument("--verbosity", type=int, default=0, choices=[0, 1, 2, 3], help="sets the detail level between 0 and 3")
    p.add_argument("--format", choices=["text", "json"], default="text", help="output format: text or json")
    p.add_argument("--track", action="append", default=[], metavar="TRACK_ID[:KEY]", help="dump a track to <input>.<track_id>; KEY is accepted for Bento4 CLI compatibility but decryption is not performed")
    p.add_argument("--no-banner", action="store_true", help="do not print the banner in text mode")
    p.add_argument("--extract-playready-pssh", action="store_true", help="print the first complete PlayReady PSSH box as base64, matching pypr")
    p.add_argument("--extract-playready-payload", action="store_true", help="print only the first inner PlayReady PSSH data payload as base64")
    return p

def main(argv: Optional[List[str]] = None) -> int:
    args = build_arg_parser().parse_args(argv)
    if not os.path.isfile(args.input):
        print("ERROR: input file does not exist: %s" % args.input, file=sys.stderr)
        return 1
    try:
        track_specs = [parse_track_option(x) for x in args.track]
        if args.extract_playready_pssh:
            pssh = extract_playready_pssh_from_mp4(args.input)
            if pssh is None:
                print("ERROR: no PlayReady PSSH found", file=sys.stderr)
                return 1
            print(pssh)
            return 0
        if args.extract_playready_payload:
            payload = extract_playready_pssh_payload_from_mp4(args.input)
            if payload is None:
                print("ERROR: no PlayReady PSSH payload found", file=sys.stderr)
                return 1
            print(payload)
            return 0
        data = load_file(args.input)
        parser = Mp4Parser(data, verbosity=args.verbosity)
        boxes = parser.parse()
        if args.format == "json":
            print(json.dumps([box.to_json() for box in boxes], indent=2, ensure_ascii=False))
        else:
            if not args.no_banner:
                print(BANNER)
                print()
            TextPrinter(args.verbosity).print(boxes)
        if track_specs:
            dumper = TrackDumper(data, parser)
            for track_id, key in track_specs:
                if key:
                    print("WARNING: decryption keys are accepted for CLI compatibility, but this Python implementation writes stored sample bytes only.", file=sys.stderr)
                output = dumper.dump_track(args.input, track_id)
                print("Dumped track %d to %s" % (track_id, output), file=sys.stderr)
        return 0
    except (OSError, Mp4DumpError, argparse.ArgumentTypeError) as exc:
        print("ERROR: %s" % exc, file=sys.stderr)
        return 1
    except KeyboardInterrupt:
        print("Interrupted", file=sys.stderr)
        return 130

if __name__ == "__main__":
    raise SystemExit(main())