"""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(" 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("= 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(), }