Publish Armiger v3.0.0 (Rule-3 Gold mit3): weights, calibration, cards and receipts-synced docs
788374c verified Download zero_cue_binary.py from NorviTech/armiger-ultra: direct link, hf CLI and curl.
- Browser
- Download file 23.5 kB
-
https://huggingface.co/NorviTech/armiger-ultra/resolve/main/zero_cue_binary.py
- Command line
-
hf download hf://NorviTech/armiger-ultra/zero_cue_binary.py
-
curl -L -o zero_cue_binary.py https://huggingface.co/NorviTech/armiger-ultra/resolve/main/zero_cue_binary.py
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), | |
| } | |
| 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(), | |
| } | |