studio/rbxl-importer/src/rbxl_parser_v0.py

"""
rbxl_parser_v0.py — голый парсер Roblox Binary Level (.rbxl) v0.

Эта первая итерация — только структура файла:
  - проверка signature (<roblox!\x89\xff\r\n\x1a\n)
  - чтение header (version, class count, instance count)
  - чтение chunks: name (4 байт), compressed_len, uncompressed_len, reserved, payload
  - LZ4-декомпрессия payload
  - выдача списка chunks с распакованным содержимым

Здесь НЕ ДЕЛАЕМ декодирование INST/PROP per-type — только видим что есть в файле.
Это нужно для первого smoke-test'а на твоей Easy Obby.

Полная спецификация: https://dom.rojo.space/binary
"""
import struct
import io
import lz4.block
from dataclasses import dataclass, field
from typing import List, Optional


RBXL_SIGNATURE = b'<roblox!\x89\xff\r\n\x1a\n'


@dataclass
class RbxlChunk:
    name: str           # 'META', 'SSTR', 'INST', 'PROP', 'PRNT', 'SIGN', 'END '
    compressed_len: int
    uncompressed_len: int
    payload: bytes      # уже декомпрессированный


@dataclass
class RbxlHeader:
    version: int        # обычно 0
    class_count: int    # сколько разных классов (Part, Script, ...)
    instance_count: int # сколько объектов всего


@dataclass
class RbxlFile:
    header: RbxlHeader
    chunks: List[RbxlChunk] = field(default_factory=list)


class RbxlParseError(Exception):
    pass


def parse(blob: bytes) -> RbxlFile:
    """Парсит .rbxl байты в RbxlFile. Бросает RbxlParseError на ошибке."""
    if not blob.startswith(RBXL_SIGNATURE):
        raise RbxlParseError(
            f"signature mismatch: got {blob[:14].hex()}, "
            f"expected {RBXL_SIGNATURE.hex()}"
        )

    stream = io.BytesIO(blob[len(RBXL_SIGNATURE):])

    # Header: version (uint16), class_count (uint32), instance_count (uint32), reserved (8 bytes)
    version, class_count, instance_count = struct.unpack('<HII', stream.read(10))
    stream.read(8)  # reserved
    header = RbxlHeader(version, class_count, instance_count)

    chunks = []
    while True:
        # Chunk header: name (4 bytes), compressed_len (uint32), uncompressed_len (uint32), reserved (uint32)
        header_bytes = stream.read(16)
        if len(header_bytes) < 16:
            break

        name_raw, compressed_len, uncompressed_len, _ = struct.unpack('<4sIII', header_bytes)
        name = name_raw.decode('ascii', errors='replace').rstrip('\x00').rstrip()

        # Payload
        payload_compressed = stream.read(compressed_len if compressed_len > 0 else uncompressed_len)

        if compressed_len == 0:
            # Не сжато
            payload = payload_compressed
        else:
            # LZ4 raw block decompress
            try:
                payload = lz4.block.decompress(payload_compressed, uncompressed_size=uncompressed_len)
            except Exception as e:
                raise RbxlParseError(
                    f"LZ4 decompress failed for chunk {name!r}: {e}"
                )

        chunks.append(RbxlChunk(name, compressed_len, uncompressed_len, payload))

        if name.startswith('END'):
            break

    return RbxlFile(header=header, chunks=chunks)


def summarize(rbxl: RbxlFile) -> str:
    """Печатает компактный отчёт о структуре файла для smoke-test."""
    lines = []
    lines.append(f"=== rbxl file ===")
    lines.append(f"  version: {rbxl.header.version}")
    lines.append(f"  classes: {rbxl.header.class_count}")
    lines.append(f"  instances: {rbxl.header.instance_count}")
    lines.append(f"  chunks: {len(rbxl.chunks)}")
    lines.append("")
    lines.append(f"=== chunks ===")
    for c in rbxl.chunks:
        ratio = (c.compressed_len / c.uncompressed_len * 100) if c.uncompressed_len > 0 else 0
        lines.append(
            f"  {c.name:6s}  compressed={c.compressed_len:>8d}  uncompressed={c.uncompressed_len:>10d}  "
            f"({ratio:5.1f}%)  first 16 bytes: {c.payload[:16].hex()}"
        )
    return "\n".join(lines)


if __name__ == '__main__':
    import sys
    path = sys.argv[1] if len(sys.argv) > 1 else '/opt/rbxl-importer/reference_files/easy_obby.rbxl'
    with open(path, 'rb') as f:
        blob = f.read()
    print(f"file size: {len(blob)} bytes")
    rbxl = parse(blob)
    print(summarize(rbxl))