mirror of
https://github.com/DarkflameUniverse/DarkflameServer.git
synced 2026-10-02 02:43:44 +00:00
docs: how the client reads brickdb.zip, and a tool to rebuild it
The client mounts brickdb.zip with LDD's own zip reader. It is lenient about compression, order and extra entries, but loses the whole brick database when the files sit under a wrapping folder or "./", use backslashes, carry ZIP64 end records, or set the data descriptor flag with sizes still in the local header. tools/brickdb/repack.py rebuilds the shipped layout (a round trip gives the identical file), adds or replaces entries from an overlay folder, and checks a zip against the reader's rules. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Aaron Kimbrell
229
tools/brickdb/repack.py
Executable file
229
tools/brickdb/repack.py
Executable file
@@ -0,0 +1,229 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Rebuild or check the client's res/brickdb.zip in the layout the LEGO Universe client (1.10.64) accepts.
|
||||
|
||||
The client mounts brickdb.zip with LEGO Digital Designer's own zip reader (see docs/BrickDb.md). It is lenient
|
||||
about compression and entry order, but a few things other zip tools commonly do make it drop the whole brick
|
||||
database: a wrapping folder or "./" prefix, backslash separators, ZIP64 end records, and data descriptors whose
|
||||
local header still carries the sizes. This script always writes the same layout the shipped file uses.
|
||||
|
||||
repack.py build SRC OUT.zip [--overlay DIR]... SRC is an extracted brickdb folder or a brickdb.zip; each
|
||||
--overlay folder (same layout) adds or replaces entries
|
||||
repack.py check FILE.zip [--res RES_DIR] report everything the client would reject; with --res also
|
||||
look for each primitive's geometry in brickprimitives/
|
||||
|
||||
Repacking the shipped brickdb.zip with no overlay gives back the identical file.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import struct
|
||||
import sys
|
||||
import time
|
||||
import zipfile
|
||||
import zlib
|
||||
import xml.etree.ElementTree as ElementTree
|
||||
|
||||
SIG_LOCAL = 0x04034B50
|
||||
SIG_CENTRAL = 0x02014B50
|
||||
SIG_END = 0x06054B50
|
||||
SIG_END64 = 0x06064B50
|
||||
SIG_END64_LOCATOR = 0x07064B50
|
||||
|
||||
# Where each file lives in the database and what the client does with it
|
||||
ROOT_FILES = ("info.xml", "Materials.xml")
|
||||
NUMBERED_DIRS = {"Primitives": ".xml", "Assemblies": ".lxfml"}
|
||||
|
||||
|
||||
def entry_order(name):
|
||||
"""The shipped order: info.xml, Materials.xml, then primitives and assemblies together by design ID, then the rest."""
|
||||
if name in ROOT_FILES:
|
||||
return (0, ROOT_FILES.index(name), 0, name)
|
||||
folder, _, file = name.partition("/")
|
||||
stem, ext = os.path.splitext(file)
|
||||
if folder in NUMBERED_DIRS and ext.lower() == NUMBERED_DIRS[folder] and stem.isdigit():
|
||||
return (1, int(stem), 0 if folder == "Primitives" else 1, name)
|
||||
return (2, 0, 0, name.lower())
|
||||
|
||||
|
||||
def read_source(src):
|
||||
"""{name: (bytes, date_time)} from an extracted folder or a zip."""
|
||||
entries = {}
|
||||
if os.path.isdir(src):
|
||||
for root, dirs, files in os.walk(src):
|
||||
dirs.sort()
|
||||
for file in files:
|
||||
path = os.path.join(root, file)
|
||||
name = os.path.relpath(path, src).replace(os.sep, "/")
|
||||
with open(path, "rb") as f:
|
||||
entries[name] = (f.read(), time.localtime(os.path.getmtime(path))[:6])
|
||||
else:
|
||||
with zipfile.ZipFile(src) as z:
|
||||
for info in z.infolist():
|
||||
if info.is_dir():
|
||||
continue
|
||||
entries[info.filename.replace("\\", "/")] = (z.read(info), info.date_time)
|
||||
# drop a "./" prefix or a wrapping folder if the source has one
|
||||
entries = {(n[2:] if n.startswith("./") else n): v for n, v in entries.items()}
|
||||
tops = {n.split("/", 1)[0] for n in entries}
|
||||
if "info.xml" not in entries and len(tops) == 1 and tops.pop() + "/info.xml" in entries:
|
||||
top = next(iter(entries)).split("/", 1)[0] + "/"
|
||||
entries = {n[len(top):]: v for n, v in entries.items()}
|
||||
return entries
|
||||
|
||||
|
||||
def build(src, out, overlays):
|
||||
entries = read_source(src)
|
||||
for overlay in overlays:
|
||||
for name, value in read_source(overlay).items():
|
||||
if name.startswith("brickprimitives/"):
|
||||
continue
|
||||
entries[name] = value
|
||||
missing = [n for n in ("info.xml",) if n not in entries]
|
||||
if missing or not any(n.startswith("Primitives/") for n in entries):
|
||||
sys.exit(f"{src}: no info.xml or Primitives/ at the top of the database")
|
||||
tmp = out + ".tmp"
|
||||
# Plain deflate at zlib's default level, no directory entries, no extra fields, no data descriptors, no ZIP64
|
||||
with zipfile.ZipFile(tmp, "w", zipfile.ZIP_DEFLATED, allowZip64=False) as z:
|
||||
for name in sorted(entries, key=entry_order):
|
||||
data, date_time = entries[name]
|
||||
info = zipfile.ZipInfo(name, date_time=max(tuple(date_time), (1980, 1, 1, 0, 0, 0)))
|
||||
info.compress_type = zipfile.ZIP_DEFLATED
|
||||
info.create_system = 3
|
||||
info.external_attr = 0o600 << 16
|
||||
z.writestr(info, data)
|
||||
os.replace(tmp, out)
|
||||
report(check(out, None), "warning: ")
|
||||
print(f"wrote {out}: {len(entries)} entries")
|
||||
|
||||
|
||||
def check(path, res):
|
||||
"""Everything the client's zip reader and brick database would reject, as a list of messages."""
|
||||
problems = []
|
||||
data = open(path, "rb").read()
|
||||
end = data.rfind(struct.pack("<I", SIG_END), max(0, len(data) - 0xFFFF - 22))
|
||||
if end < 0:
|
||||
return ["no end of central directory record"]
|
||||
_, disk, cd_disk, count_disk, count, cd_size, cd_offset, _ = struct.unpack_from("<IHHHHIIH", data, end)
|
||||
if disk or cd_disk or count_disk != count:
|
||||
problems.append("multi-disk archive (disk numbers must be 0)")
|
||||
if cd_offset + cd_size != end:
|
||||
gap = data[cd_offset + cd_size:end]
|
||||
kind = "ZIP64 end records" if gap[:4] == struct.pack("<I", SIG_END64) else f"{len(gap)} unexpected bytes"
|
||||
problems.append(f"{kind} between the central directory and its end record (the client needs them adjacent)")
|
||||
if cd_offset + cd_size > end or cd_offset > len(data):
|
||||
return problems + ["central directory offset is outside the file (data before the zip?)"]
|
||||
|
||||
names = set()
|
||||
owners = {} # design ID or alias -> primitives that claim it
|
||||
pos = cd_offset
|
||||
for _ in range(count):
|
||||
if struct.unpack_from("<I", data, pos)[0] != SIG_CENTRAL:
|
||||
problems.append(f"central directory entry at {pos} has a bad signature")
|
||||
break
|
||||
(_, _, _, flags, method, _, _, crc, csize, usize, name_len, extra_len, comment_len,
|
||||
_, _, _, local) = struct.unpack_from("<IHHHHHHIIIHHHHHII", data, pos)
|
||||
raw = data[pos + 46:pos + 46 + name_len]
|
||||
name = raw.decode("utf-8" if flags & 0x800 else "cp437")
|
||||
pos += 46 + name_len + extra_len + comment_len
|
||||
where = f"{name!r}:"
|
||||
if "\\" in name:
|
||||
problems.append(f"{where} backslash in the path (the client only splits on '/')")
|
||||
parts = name.rstrip("/").split("/")
|
||||
if name.startswith("/") or "" in parts or "." in parts or ".." in parts:
|
||||
problems.append(f"{where} empty, '.' or '..' path component")
|
||||
if flags & 1:
|
||||
problems.append(f"{where} encrypted")
|
||||
if method not in (0, 8):
|
||||
problems.append(f"{where} compression method {method} (only stored and deflate are read)")
|
||||
if struct.unpack_from("<I", data, local)[0] != SIG_LOCAL:
|
||||
problems.append(f"{where} local header missing")
|
||||
continue
|
||||
_, _, lflags, lmethod, _, _, lcrc, lcsize, lusize, lname, lextra = struct.unpack_from("<IHHHHHIIIHH", data, local)
|
||||
if lflags != flags or lmethod != method:
|
||||
problems.append(f"{where} local header flags/method differ from the central directory")
|
||||
if flags & 8:
|
||||
if lcrc or lcsize or lusize:
|
||||
problems.append(f"{where} data descriptor flag set but the local header has sizes (the client needs zeros)")
|
||||
elif (lcrc, lcsize, lusize) != (crc, csize, usize):
|
||||
problems.append(f"{where} local header CRC/sizes differ from the central directory")
|
||||
if not name.endswith("/"):
|
||||
start = local + 30 + lname + lextra
|
||||
body = data[start:start + csize]
|
||||
try:
|
||||
content = body if method == 0 else zlib.decompressobj(-15).decompress(body)
|
||||
if zlib.crc32(content) & 0xFFFFFFFF != crc:
|
||||
problems.append(f"{where} CRC mismatch (the client does not check it, other tools will)")
|
||||
except zlib.error as e:
|
||||
problems.append(f"{where} does not inflate ({e})")
|
||||
content = None
|
||||
names.add(name)
|
||||
folder, _, file = name.partition("/")
|
||||
if folder in NUMBERED_DIRS and "/" not in file and file.lower().endswith(NUMBERED_DIRS[folder]):
|
||||
stem = os.path.splitext(file)[0]
|
||||
if not stem.isdigit():
|
||||
problems.append(f"{where} name is not a design ID; the client parses the file name as the ID")
|
||||
if content is not None:
|
||||
try:
|
||||
root = ElementTree.fromstring(content)
|
||||
except ElementTree.ParseError as e:
|
||||
problems.append(f"{where} XML does not parse ({e}); this brick will be skipped")
|
||||
root = None
|
||||
if root is not None and folder == "Primitives":
|
||||
ids = {stem}
|
||||
for annotation in root.iter("Annotation"):
|
||||
ids.update(a for a in annotation.get("aliases", "").split(";") if a)
|
||||
for i in ids:
|
||||
owners.setdefault(i, []).append(name)
|
||||
|
||||
for i, claimed in sorted(owners.items()):
|
||||
if len(claimed) > 1:
|
||||
problems.append(f"design ID/alias {i} is claimed by {', '.join(claimed)}")
|
||||
lower = {n.lower() for n in names}
|
||||
if "info.xml" not in lower:
|
||||
problems.append("no info.xml at the top level")
|
||||
primitives = sorted(n for n in names if n.lower().startswith("primitives/") and n.lower().endswith(".xml"))
|
||||
if not primitives:
|
||||
problems.append("no Primitives/*.xml at the top level (wrapping folder?); the client drops the whole database")
|
||||
if res:
|
||||
for n in primitives:
|
||||
stem = os.path.splitext(n.split("/")[-1])[0]
|
||||
if stem.isdigit() and not os.path.exists(os.path.join(res, "brickprimitives", "lod0", stem + ".g")):
|
||||
problems.append(f"{n}: no brickprimitives/lod0/{stem}.g")
|
||||
return problems
|
||||
|
||||
|
||||
def report(problems, prefix=""):
|
||||
"""Prints the problems, at most three entries for each kind."""
|
||||
kinds = {}
|
||||
for p in problems:
|
||||
kind = p.split(": ", 1)[1] if p.startswith("'") and ": " in p else p
|
||||
kinds.setdefault(kind, []).append(p)
|
||||
for kind, items in kinds.items():
|
||||
for p in items[:3]:
|
||||
print(prefix + p)
|
||||
if len(items) > 3:
|
||||
print(f"{prefix}... and {len(items) - 3} more entries: {kind}")
|
||||
print(f"{len(problems)} problem(s)")
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
sub = parser.add_subparsers(dest="command", required=True)
|
||||
b = sub.add_parser("build", help="write a brickdb.zip the client accepts")
|
||||
b.add_argument("src")
|
||||
b.add_argument("out")
|
||||
b.add_argument("--overlay", action="append", default=[])
|
||||
c = sub.add_parser("check", help="list what the client would reject")
|
||||
c.add_argument("zip")
|
||||
c.add_argument("--res", help="the client's res folder, to look for geometry")
|
||||
args = parser.parse_args()
|
||||
if args.command == "build":
|
||||
build(args.src, args.out, args.overlay)
|
||||
else:
|
||||
problems = check(args.zip, args.res)
|
||||
report(problems)
|
||||
sys.exit(1 if problems else 0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user