kangshifu-board/scripts/reassemble-voc-database.py

"""Reassemble the two private transport ZIPs; never install or replace a database."""
import argparse
import hashlib
import json
import os
from pathlib import Path
import sys
import tempfile
import zipfile

sys.dont_write_bytecode = True
FILENAME = 'ksf-voc-database-20261106-r1.zip'
SIZE = 25913256
SHA256 = '1ba1ac2fa6a8b774f8945d88e014c07dd5e03332792af298a41f015ba6946f9f'
PART_SIZES = [14000000, 11913256]

def require(value, code):
    if not value:
        raise ValueError(code)

def digest(p):
    with Path(p).open('rb') as f:
        result = hashlib.sha256()
        for block in iter(lambda: f.read(1024 * 1024), b''):
            result.update(block)
        return result.hexdigest()

def regular_path(p):
    p = Path(p)
    require(p.is_absolute(), 'ABSOLUTE_PATH_REQUIRED')
    for parent in (p, *p.parents):
        require(not parent.is_symlink(), 'SYMLINK_PATH_REFUSED')
    return p

def reassemble(parts, output):
    require(len(parts) == 2, 'EXACTLY_TWO_PARTS_REQUIRED')
    paths = [regular_path(p) for p in parts]
    require(len(set(paths)) == 2 and all(p.is_file() for p in paths), 'PART_MISSING_OR_DUPLICATED')
    output = regular_path(output)
    require(output.name == FILENAME and output.parent.is_dir(), 'INVALID_OUTPUT_PATH')
    ordered = {}
    common = None
    for p in paths:
        with zipfile.ZipFile(p) as z:
            require(z.getinfo('TRANSPORT.json').file_size < 16000, 'INVALID_TRANSPORT_MANIFEST')
            meta = json.loads(z.read('TRANSPORT.json'))
            require(meta.get('filename') == FILENAME and meta.get('bytes') == SIZE and
                    meta.get('sha256') == SHA256 and len(meta.get('parts', [])) == 2,
                    'WRONG_SOURCE_PACKAGE')
            require(common is None or common == meta, 'TRANSPORT_MANIFEST_MISMATCH')
            common = meta
            found = [i for i in (1, 2) if FILENAME + f'.part{i:02}' in z.namelist()]
            require(len(found) == 1 and found[0] not in ordered, 'PART_MISSING_OR_DUPLICATED')
            index = found[0]
            entry = meta['parts'][index - 1]
            name = FILENAME + f'.part{index:02}'
            require(entry.get('chunk') == name and entry.get('bytes') == PART_SIZES[index - 1] and
                    isinstance(entry.get('sha256'), str) and len(entry['sha256']) == 64 and
                    z.getinfo(name).file_size == PART_SIZES[index - 1], 'INVALID_PART_SIZE_OR_NAME')
            require(len(z.namelist()) == len(set(z.namelist())) == 4 and
                    set(z.namelist()) == {name, 'TRANSPORT.json', 'reassemble-voc-database.py', 'README.md'},
                    'UNEXPECTED_TRANSPORT_CONTENTS')
            ordered[index] = (p, entry)
    require(set(ordered) == {1, 2}, 'PART_MISSING_OR_DUPLICATED')
    if output.exists():
        require(output.is_file() and output.stat().st_size == SIZE and digest(output) == SHA256,
                'DIFFERENT_OUTPUT_ALREADY_EXISTS')
        return {'ok': True, 'operation': 'unchanged', 'output': str(output), 'bytes': SIZE,
                'sha256': SHA256, 'fileCreated': False, 'installed': False, 'databaseWrites': 0}
    fd, temporary = tempfile.mkstemp(prefix='.ksf-voc-reassemble-', dir=output.parent)
    try:
        total = 0
        whole = hashlib.sha256()
        with os.fdopen(fd, 'wb') as dest:
            for index in (1, 2):
                p, entry = ordered[index]
                chunk_hash = hashlib.sha256()
                length = 0
                with zipfile.ZipFile(p) as z, z.open(entry['chunk']) as source:
                    while data := source.read(1024 * 1024):
                        length += len(data)
                        require(length <= entry['bytes'], 'PART_TOO_LARGE')
                        chunk_hash.update(data)
                        whole.update(data)
                        dest.write(data)
                require(length == entry['bytes'] and chunk_hash.hexdigest() == entry['sha256'],
                        'PART_HASH_MISMATCH')
                total += length
            require(total == SIZE and whole.hexdigest() == SHA256, 'REASSEMBLED_HASH_MISMATCH')
            dest.flush()
            os.fsync(dest.fileno())
        # A same-filesystem hard link publishes the verified file without overwriting a race winner.
        os.link(temporary, output)
        return {'ok': True, 'operation': 'reassembled', 'output': str(output), 'bytes': total,
                'sha256': whole.hexdigest(), 'fileCreated': True, 'installed': False, 'databaseWrites': 0}
    finally:
        os.unlink(temporary)

def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--part', action='append', required=True, help='Actual absolute path to each transport ZIP; repeat twice')
    parser.add_argument('--output', required=True, help='Absolute path of the original ZIP under an existing private directory')
    args = parser.parse_args()
    try:
        print(json.dumps(reassemble(args.part, args.output), ensure_ascii=False, indent=2))
    except Exception as exc:
        code = str(exc) if isinstance(exc, ValueError) else 'REASSEMBLY_STOPPED'
        print(json.dumps({'ok': False, 'error': {'code': code}, 'installed': False, 'databaseWrites': 0}))
        return 1
    return 0

if __name__ == '__main__':
    raise SystemExit(main())