armiger-ultra / zero_cue_binary.py
spencernorton's picture
Publish Armiger v3.0.0 (Rule-3 Gold mit3): weights, calibration, cards and receipts-synced docs
788374c verified
Raw History Blame Contribute Delete
23.5 kB
"""Zero-Cue Binary Catalog (`ZCB3`) Packer & Zero-Deserialization `mmap` Reader.
Implements Session 3's `ZCB3` sorted 9-byte binary table format for `validated_zero_cue_catalog.json`
(`929,408` BLAKE2b-64 entries across `anime_map`, `title_year_map`, and `title_only_map`):
- Header (16 bytes): `4B magic b"ZCB3"` + `3 x uint32 (<3I)` counts `(n_anime, n_title_year, n_title_only)`
- Body (`9 * N` bytes): 3 contiguous sections in order (`anime_map`, `title_year_map`, `title_only_map`),
each sorted ascending by 8-byte big-endian BLAKE2b-64 key + 1-byte `uint8` category code.
- Size reduction:
* Raw JSON (`25,990,893 B` = `25.99 MB`)
* Uncompressed `ZCB3` (`8,364,688 B` = `8.36 MB`, `-67.8%`, supports `O(log N)` zero-deserialization `mmap`)
* `ZCB3` + `zlib-9` (`7,425,374 B` = `7.43 MB`, `-71.4%`)
* `ZCB3` + `lzma-6` (`6,451,624 B` = `6.45 MB`, `-75.2%`)
"""
from __future__ import annotations
from collections.abc import Iterator, Mapping
import hashlib
import json
import lzma
import mmap
from pathlib import Path
import struct
from typing import Any
import zlib
ZCB3_MAGIC = b"ZCB3"
ZCB3_VERSION = 3
ZCB3_EXT_FLAG = 0x5A434233
ZCB3_HEADER_STRUCT = struct.Struct("<4s3I")
ZCB3_HEADER_SIZE = ZCB3_HEADER_STRUCT.size # 16 bytes
ZCB3_EXT_HEADER_STRUCT = struct.Struct("<4sIIIQQ32s")
ZCB3_EXT_HEADER_SIZE = ZCB3_EXT_HEADER_STRUCT.size # 64 bytes
ZCB3_EXT_RECORD_STRUCT = struct.Struct("<QIIBB6s")
ZCB3_EXT_RECORD_SIZE = ZCB3_EXT_RECORD_STRUCT.size # 24 bytes
ZCB3_RECORD_SIZE = 9 # 8-byte BLAKE2b-64 key + 1-byte category code
ZCB3_SECTION_NAMES: tuple[str, str, str] = ("anime_map", "title_year_map", "title_only_map")
MAX_ZCB3_UNCOMPRESSED_BYTES = 64 * 1024 * 1024 # 64 MiB hard bound against decompression bombs
MAX_ZCB3_LZMA_MEMLIMIT_BYTES = 64 * 1024 * 1024 # 64 MiB LZMA decoder memory bound (supports preset 6)
MAX_EAGER_VALIDATE_RECORDS = 100_000
CAT_TO_CODE: dict[str, int] = {
"movie": 0,
"tv_show": 1,
"ambiguous": 2,
"xxx": 3,
"music": 4,
"comic": 5,
"game": 6,
"audiobook": 7,
"software": 8,
"ebook": 9,
"course": 10,
"unknown": 11,
"books": 12,
}
CODE_TO_CAT: dict[int, str] = {v: k for k, v in CAT_TO_CODE.items()}
def _hash_catalog_key_bytes(key: str) -> bytes:
"""Compute 8-byte keyed BLAKE2b digest for an unhashed lookup key."""
return hashlib.blake2b(key.encode("utf-8"), digest_size=8, person=b"armiger-zc-v2").digest()
def _normalize_lookup_key(key: str | bytes) -> bytes | None:
"""Convert a 16-char hex digest, 8-byte raw digest, or plaintext lookup string into 8 raw bytes."""
if isinstance(key, (bytes, bytearray, memoryview)):
b = bytes(key)
return b if len(b) == 8 else None
if not isinstance(key, str):
return None
if len(key) == 16:
try:
return bytes.fromhex(key)
except ValueError:
pass
return _hash_catalog_key_bytes(key)
def pack_zcb3_bytes(catalog: Mapping[str, Any]) -> bytes:
"""Pack `anime_map`, `title_year_map`, and `title_only_map` into deterministic `ZCB3` bytes."""
packed_sections: list[bytes] = []
counts: list[int] = []
for section_name in ZCB3_SECTION_NAMES:
raw_map = catalog.get(section_name) or {}
if not isinstance(raw_map, Mapping):
raise TypeError(f"Expected mapping for {section_name}, got {type(raw_map)!r}")
parsed_items: list[tuple[bytes, int]] = []
for k_raw, val_str in raw_map.items():
if isinstance(k_raw, bytes) and len(k_raw) == 8:
k_bytes = k_raw
elif isinstance(k_raw, str) and len(k_raw) == 16:
try:
k_bytes = bytes.fromhex(k_raw)
except ValueError:
k_bytes = _hash_catalog_key_bytes(k_raw)
elif isinstance(k_raw, str):
k_bytes = _hash_catalog_key_bytes(k_raw)
else:
raise ValueError(f"Invalid key in {section_name}: {k_raw!r}")
code = CAT_TO_CODE.get(str(val_str), 3)
parsed_items.append((k_bytes, code))
parsed_items.sort(key=lambda item: item[0])
counts.append(len(parsed_items))
buf = bytearray(len(parsed_items) * ZCB3_RECORD_SIZE)
offset = 0
for k_bytes, code in parsed_items:
buf[offset : offset + 8] = k_bytes
buf[offset + 8] = code
offset += ZCB3_RECORD_SIZE
packed_sections.append(bytes(buf))
header = ZCB3_HEADER_STRUCT.pack(ZCB3_MAGIC, counts[0], counts[1], counts[2])
return header + b"".join(packed_sections)
def pack_zcb3_extended_bytes(
records: list[tuple[int, str, int]],
*,
version: int = ZCB3_VERSION,
) -> bytes:
"""Pack `(prefix64, title_str, category_id)` tuples into the 64-byte extended `ZCB3` header
+ 24-byte index records + UTF-8 string table, embedding `sha256(payload)` in `catalog_sha256`.
"""
sorted_recs = sorted(records, key=lambda r: int(r[0]))
index_offset = ZCB3_EXT_HEADER_SIZE
string_table_offset = index_offset + len(sorted_recs) * ZCB3_EXT_RECORD_SIZE
index_buf = bytearray(len(sorted_recs) * ZCB3_EXT_RECORD_SIZE)
str_chunks: list[bytes] = []
cur_str_off = string_table_offset
for idx, (prefix64, title_str, cat_id) in enumerate(sorted_recs):
t_bytes = str(title_str).encode("utf-8")
str_chunks.append(t_bytes)
ZCB3_EXT_RECORD_STRUCT.pack_into(
index_buf,
idx * ZCB3_EXT_RECORD_SIZE,
int(prefix64) & 0xFFFFFFFFFFFFFFFF,
cur_str_off,
len(t_bytes),
int(cat_id) & 0xFF,
0,
b"\x00" * 6,
)
cur_str_off += len(t_bytes)
payload = bytes(index_buf) + b"".join(str_chunks)
digest = hashlib.sha256(payload).digest()
header = ZCB3_EXT_HEADER_STRUCT.pack(
ZCB3_MAGIC,
int(version),
ZCB3_EXT_FLAG,
len(sorted_recs),
index_offset,
string_table_offset,
digest,
)
return header + payload
def compress_zcb3_zlib9(zcb3_bytes: bytes) -> bytes:
"""Compress packed `ZCB3` bytes with `zlib` level 9 (~`7.43 MB` for full catalog)."""
return zlib.compress(zcb3_bytes, level=9)
def compress_zcb3_lzma6(zcb3_bytes: bytes) -> bytes:
"""Compress packed `ZCB3` bytes with `lzma` preset 6 (~`6.45 MB` for full catalog)."""
return lzma.compress(zcb3_bytes, preset=6)
def decompress_zcb3(
payload: bytes,
*,
max_bytes: int = MAX_ZCB3_UNCOMPRESSED_BYTES,
lzma_memlimit: int = MAX_ZCB3_LZMA_MEMLIMIT_BYTES,
) -> bytes:
"""Safely decompress raw, `zlib`, or `lzma`-compressed `ZCB3` bytes with bounded output length
and bounded LZMA decoder memory.
"""
if not isinstance(payload, (bytes, bytearray, memoryview)) or not payload:
raise ValueError("Empty ZCB3 payload")
max_b = int(max_bytes)
mem_b = int(lzma_memlimit)
if max_b <= 0 or mem_b <= 0:
raise ValueError("max_bytes and lzma_memlimit must be positive")
raw = bytes(payload)
if raw.startswith(ZCB3_MAGIC):
if len(raw) > max_b:
raise ValueError(f"ZCB3 payload exceeds max_bytes ({len(raw)} > {max_b})")
return raw
if raw.startswith(b"\xfd7zXZ\x00"):
try:
dec = lzma.LZMADecompressor(memlimit=mem_b)
out = dec.decompress(raw, max_length=max_b + 1)
except lzma.LZMAError as exc:
raise ValueError(f"zcb3_lzma_decompression_error:{exc}") from exc
if len(out) > max_b or not dec.eof:
raise ValueError("zcb3_lzma_decompression_limit_exceeded")
if dec.unused_data:
raise ValueError("zcb3_lzma_trailing_garbage_bytes")
if not out.startswith(ZCB3_MAGIC):
raise ValueError("Decompressed LZMA payload does not start with ZCB3 magic")
return out
try:
zdec = zlib.decompressobj()
out = zdec.decompress(raw, max_length=max_b + 1)
if len(out) > max_b or zdec.unconsumed_tail:
raise ValueError("zcb3_zlib_decompression_limit_exceeded")
rem = max_b + 1 - len(out)
out += zdec.flush(max(1, rem))
except zlib.error as exc:
raise ValueError(f"Invalid ZCB3 header or compressed stream: {exc}") from exc
if len(out) > max_b or zdec.unconsumed_tail:
raise ValueError("zcb3_zlib_decompression_limit_exceeded")
if not zdec.eof:
raise ValueError("zcb3_zlib_incomplete_stream")
if zdec.unused_data:
raise ValueError("zcb3_zlib_trailing_garbage_bytes")
if not out.startswith(ZCB3_MAGIC):
raise ValueError("Decompressed Zlib payload does not start with ZCB3 magic")
return out
class ZCB3MapSectionView(Mapping[str, str]):
"""Read-only `O(log N)` binary-search mapping view over a single `ZCB3` section (`mmap` or `bytes`)."""
def __init__(self, buf: mmap.mmap | bytes, base_offset: int, count: int) -> None:
self._buf = buf
self._base = int(base_offset)
self._count = int(count)
def __len__(self) -> int:
return self._count
def get(self, key: Any, default: Any = None) -> Any:
target8 = _normalize_lookup_key(key)
if target8 is None or self._count <= 0:
return default
lo = 0
hi = self._count - 1
base = self._base
buf = self._buf
while lo <= hi:
mid = (lo + hi) >> 1
off = base + mid * ZCB3_RECORD_SIZE
k8 = buf[off : off + 8]
if mid > 0 and buf[off - ZCB3_RECORD_SIZE : off - 1] >= k8:
raise ValueError(f"ZCB3 keys out of ascending order at index {mid}")
if mid + 1 < self._count and k8 >= buf[off + ZCB3_RECORD_SIZE : off + ZCB3_RECORD_SIZE + 8]:
raise ValueError(f"ZCB3 keys out of ascending order at index {mid}")
code = buf[off + 8]
if code not in CODE_TO_CAT:
raise ValueError(f"Invalid ZCB3 category_id enum index: {code}")
if k8 == target8:
return CODE_TO_CAT[code]
if k8 < target8:
lo = mid + 1
else:
hi = mid - 1
return default
def __contains__(self, key: object) -> bool:
if not isinstance(key, (str, bytes, bytearray, memoryview)):
return False
return self.get(key) is not None
def __getitem__(self, key: str | bytes) -> str:
val = self.get(key)
if val is None:
raise KeyError(key)
return str(val)
def __iter__(self) -> Iterator[str]:
base = self._base
buf = self._buf
for i in range(self._count):
off = base + i * ZCB3_RECORD_SIZE
yield bytes(buf[off : off + 8]).hex()
def iter_records(self) -> Iterator[tuple[str, str]]:
"""Yield `(key_hex, category_str)` in sorted order without building a dict in memory."""
base = self._base
buf = self._buf
prev_k8: bytes | None = None
for i in range(self._count):
off = base + i * ZCB3_RECORD_SIZE
k8 = bytes(buf[off : off + 8])
if prev_k8 is not None and prev_k8 >= k8:
raise ValueError(f"ZCB3 keys out of ascending order at index {i}")
prev_k8 = k8
code = buf[off + 8]
if code not in CODE_TO_CAT:
raise ValueError(f"Invalid ZCB3 category_id enum index: {code}")
yield k8.hex(), CODE_TO_CAT[code]
class ZCB3MmapCatalog(Mapping[str, Any]):
"""Zero-deserialization `ZCB3` catalog backed by an OS `mmap` or in-memory `bytes` buffer."""
def __init__(
self,
path_or_bytes: str | Path | bytes | bytearray | memoryview,
*,
expected_sha256: str | None = None,
) -> None:
self._file_obj: Any = None
self._mmap_obj: mmap.mmap | None = None
self._raw_bytes: bytes | None = None
self.source_path: Path | None = None
self.storage_mode: str = "memory_zcb3"
self.extended_records: list[dict[str, Any]] | None = None
if isinstance(path_or_bytes, (bytes, bytearray, memoryview)):
self._raw_bytes = decompress_zcb3(bytes(path_or_bytes))
buf: mmap.mmap | bytes = self._raw_bytes
else:
p = Path(path_or_bytes)
if not p.exists():
raise FileNotFoundError(f"ZCB3 catalog file not found: {p}")
self.source_path = p
file_len = p.stat().st_size
if file_len == 0:
raise ValueError("Empty ZCB3 file")
with p.open("rb") as f_peek:
head4 = f_peek.read(4)
if head4 == ZCB3_MAGIC:
self._file_obj = p.open("rb")
self._mmap_obj = mmap.mmap(self._file_obj.fileno(), 0, access=mmap.ACCESS_READ)
buf = self._mmap_obj
self.storage_mode = "mmap_zcb3"
else:
self._raw_bytes = decompress_zcb3(p.read_bytes())
buf = self._raw_bytes
self.storage_mode = "compressed_zcb3"
if len(buf) < ZCB3_HEADER_SIZE:
self.close()
raise ValueError(f"ZCB3 buffer too short for header: {len(buf)} bytes")
if expected_sha256 is not None:
actual_sha = hashlib.sha256(buf).hexdigest().lower()
if actual_sha != str(expected_sha256).strip().lower():
self.close()
raise ValueError(
f"ZCB3 catalog_sha256 mismatch: expected {expected_sha256}, got {actual_sha}"
)
# Check if this is an extended 64-byte ZCB3 indexed header (flags == ZCB3_EXT_FLAG)
maybe_flags = struct.unpack_from("<I", buf, 8)[0] if len(buf) >= 12 else 0
if maybe_flags == ZCB3_EXT_FLAG:
self._init_extended_header(buf)
return
magic, n_anime, n_title_year, n_title_only = ZCB3_HEADER_STRUCT.unpack(buf[:ZCB3_HEADER_SIZE])
if magic != ZCB3_MAGIC:
self.close()
raise ValueError(f"Invalid ZCB3 magic bytes: {magic!r}")
total_records = n_anime + n_title_year + n_title_only
expected_size = ZCB3_HEADER_SIZE + total_records * ZCB3_RECORD_SIZE
if len(buf) != expected_size:
self.close()
raise ValueError(
f"ZCB3 buffer size mismatch: expected {expected_size} bytes for {total_records} records, got {len(buf)}"
)
self.counts: dict[str, int] = {
"anime_map": int(n_anime),
"title_year_map": int(n_title_year),
"title_only_map": int(n_title_only),
}
self.total_entries: int = int(total_records)
self.size_bytes: int = len(buf)
off_anime = ZCB3_HEADER_SIZE
off_title_year = off_anime + n_anime * ZCB3_RECORD_SIZE
off_title_only = off_title_year + n_title_year * ZCB3_RECORD_SIZE
self._sections: dict[str, ZCB3MapSectionView] = {
"anime_map": ZCB3MapSectionView(buf, off_anime, n_anime),
"title_year_map": ZCB3MapSectionView(buf, off_title_year, n_title_year),
"title_only_map": ZCB3MapSectionView(buf, off_title_only, n_title_only),
}
if total_records <= MAX_EAGER_VALIDATE_RECORDS:
try:
for view in self._sections.values():
for _ in view.iter_records():
pass
except Exception:
self.close()
raise
def _init_extended_header(self, buf: mmap.mmap | bytes) -> None:
if len(buf) < ZCB3_EXT_HEADER_SIZE:
self.close()
raise ValueError(f"ZCB3 extended header truncated: {len(buf)} < {ZCB3_EXT_HEADER_SIZE} bytes")
magic, version, flags, record_count, index_offset, string_table_offset, header_sha256 = (
ZCB3_EXT_HEADER_STRUCT.unpack(buf[:ZCB3_EXT_HEADER_SIZE])
)
if magic != ZCB3_MAGIC:
self.close()
raise ValueError(f"Invalid ZCB3 magic bytes: {magic!r}")
if version != ZCB3_VERSION:
self.close()
raise ValueError(f"Unsupported ZCB3 version: {version}")
if index_offset != ZCB3_EXT_HEADER_SIZE:
self.close()
raise ValueError(f"Invalid ZCB3 index_offset: {index_offset} != {ZCB3_EXT_HEADER_SIZE}")
index_end = index_offset + record_count * ZCB3_EXT_RECORD_SIZE
if (
record_count > 10_000_000
or index_end > string_table_offset
or string_table_offset > len(buf)
or index_end > len(buf)
):
self.close()
raise ValueError(
f"Out-of-bounds ZCB3 header offsets: record_count={record_count}, "
f"index_offset={index_offset}, string_table_offset={string_table_offset}, size={len(buf)}"
)
actual_payload_sha = hashlib.sha256(buf[ZCB3_EXT_HEADER_SIZE:]).digest()
if actual_payload_sha != header_sha256:
self.close()
raise ValueError("ZCB3 embedded catalog_sha256 header digest mismatch against payload")
records_out: list[dict[str, Any]] = []
prev_prefix: int | None = None
for i in range(record_count):
rec_off = index_offset + i * ZCB3_EXT_RECORD_SIZE
prefix64, title_offset, title_len, category_id, rec_flags, _ = ZCB3_EXT_RECORD_STRUCT.unpack_from(
buf, rec_off
)
if prev_prefix is not None and prefix64 <= prev_prefix:
self.close()
raise ValueError(f"Out-of-order prefix64 key at index {i}: {prefix64} <= {prev_prefix}")
prev_prefix = prefix64
if category_id not in CODE_TO_CAT:
self.close()
raise ValueError(f"Invalid ZCB3 category_id enum index at {i}: {category_id}")
if title_offset < string_table_offset or title_offset + title_len > len(buf):
self.close()
raise ValueError(
f"ZCB3 record {i} title slice out of bounds: {title_offset}+{title_len} > {len(buf)}"
)
raw_title = bytes(buf[title_offset : title_offset + title_len])
try:
title_str = raw_title.decode("utf-8", errors="strict")
except UnicodeDecodeError as exc:
self.close()
raise ValueError(f"Invalid UTF-8 sequence in ZCB3 string table at record {i}") from exc
records_out.append(
{
"prefix64": prefix64,
"title": title_str,
"category": CODE_TO_CAT[category_id],
"category_id": category_id,
"flags": rec_flags,
}
)
self.extended_records = records_out
self.counts = {"anime_map": 0, "title_year_map": 0, "title_only_map": int(record_count)}
self.total_entries = int(record_count)
self.size_bytes = len(buf)
self._sections = {
"anime_map": ZCB3MapSectionView(b"", 0, 0),
"title_year_map": ZCB3MapSectionView(b"", 0, 0),
"title_only_map": ZCB3MapSectionView(b"", 0, 0),
}
@classmethod
def from_json_catalog(
cls,
json_path_or_dict: str | Path | Mapping[str, Any],
*,
out_zcb3_path: str | Path | None = None,
) -> ZCB3MmapCatalog:
"""Pack a JSON catalog dict or file into `ZCB3` (optionally writing to disk for `mmap`) and return reader."""
if isinstance(json_path_or_dict, Mapping):
doc = json_path_or_dict
else:
doc = json.loads(Path(json_path_or_dict).read_text(encoding="utf-8"))
packed = pack_zcb3_bytes(doc)
if out_zcb3_path is not None:
out_p = Path(out_zcb3_path)
out_p.write_bytes(packed)
return cls(out_p)
return cls(packed)
def get(self, key: str, default: Any = None) -> Any:
if key == "format":
return "zcb3_blake2b64"
return self._sections.get(key, default)
def __getitem__(self, key: str) -> Any:
if key == "format":
return "zcb3_blake2b64"
if key in self._sections:
return self._sections[key]
raise KeyError(key)
def __contains__(self, key: object) -> bool:
return key == "format" or key in self._sections
def __iter__(self) -> Iterator[str]:
yield from self._sections.keys()
def __len__(self) -> int:
return len(self._sections)
def to_dict(self) -> dict[str, dict[str, str]]:
"""Materialize all 3 sections back into standard Python dicts for round-trip verification."""
return {
name: dict(view.iter_records())
for name, view in self._sections.items()
}
def close(self) -> None:
if self._mmap_obj is not None:
try:
self._mmap_obj.close()
except Exception:
pass
self._mmap_obj = None
if self._file_obj is not None:
try:
self._file_obj.close()
except Exception:
pass
self._file_obj = None
def __enter__(self) -> ZCB3MmapCatalog:
return self
def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
self.close()
def write_zcb3_file(
catalog_or_json_path: str | Path | Mapping[str, Any],
out_path: str | Path,
*,
compression: str = "none",
) -> dict[str, Any]:
"""Pack a JSON catalog into `ZCB3` and write raw, `zlib-9`, or `lzma-6` output file."""
if isinstance(catalog_or_json_path, Mapping):
doc = catalog_or_json_path
raw_json_bytes = len(json.dumps(doc).encode("utf-8"))
else:
src_p = Path(catalog_or_json_path)
raw_bytes = src_p.read_bytes()
raw_json_bytes = len(raw_bytes)
doc = json.loads(raw_bytes.decode("utf-8"))
zcb3_raw = pack_zcb3_bytes(doc)
comp_mode = compression.strip().lower()
if comp_mode in ("none", "raw", "zcb3"):
out_bytes = zcb3_raw
elif comp_mode in ("zlib", "zlib9", "zlib-9"):
out_bytes = compress_zcb3_zlib9(zcb3_raw)
elif comp_mode in ("lzma", "lzma6", "lzma-6", "xz"):
out_bytes = compress_zcb3_lzma6(zcb3_raw)
else:
raise ValueError(f"Unsupported ZCB3 compression mode: {compression!r}")
dest = Path(out_path)
dest.parent.mkdir(parents=True, exist_ok=True)
dest.write_bytes(out_bytes)
_, n_anime, n_title_year, n_title_only = ZCB3_HEADER_STRUCT.unpack(zcb3_raw[:ZCB3_HEADER_SIZE])
return {
"out_path": str(dest),
"compression": comp_mode,
"raw_json_bytes": raw_json_bytes,
"zcb3_uncompressed_bytes": len(zcb3_raw),
"written_bytes": len(out_bytes),
"reduction_vs_json_pct": round((1.0 - len(out_bytes) / max(raw_json_bytes, 1)) * 100.0, 2),
"total_entries": int(n_anime + n_title_year + n_title_only),
"map_counts": {
"anime_map": int(n_anime),
"title_year_map": int(n_title_year),
"title_only_map": int(n_title_only),
},
"zcb3_sha256": hashlib.sha256(zcb3_raw).hexdigest(),
"written_sha256": hashlib.sha256(out_bytes).hexdigest(),
}