kangshifu-board/scripts/split-voc-database-delivery.py

"""Produce two small ordinary ZIP attachments, then verify exact reconstruction."""
import hashlib
import importlib.util
import json
from pathlib import Path
import shutil
import zipfile

ROOT = Path(__file__).resolve().parents[1]
SOURCE = Path('F:/文档/康师傅skill/ksf-voc-database-20261106-r1.zip')
OUT = ROOT / 'artifacts/研究数据库分包投递-20261106'
WORK = ROOT / 'work/voc-database-transport-20261106'
sha = lambda b: hashlib.sha256(b).hexdigest()

def main():
    source = SOURCE.read_bytes()
    expected = '1ba1ac2fa6a8b774f8945d88e014c07dd5e03332792af298a41f015ba6946f9f'
    if len(source) != 25913256 or sha(source) != expected:
        raise ValueError('SOURCE_HASH_MISMATCH')
    with zipfile.ZipFile(SOURCE) as z:
        assert len(z.namelist()) == 17 and z.testzip() is None
        manifest = z.read('FILES.sha256').decode('utf-8').splitlines()
        assert len(manifest) == 16
        for line in manifest:
            h, name = line.split('  ', 1)
            assert sha(z.read(name)) == h
    if OUT.exists() or WORK.exists():
        raise ValueError('TRANSPORT_OUTPUT_ALREADY_EXISTS')
    OUT.mkdir(parents=True)
    WORK.mkdir(parents=True)
    chunks = [source[:14000000], source[14000000:]]
    transport = {'schema': 'ksf.voc-database.transport.v1', 'filename': SOURCE.name,
                 'bytes': len(source), 'sha256': sha(source), 'parts': [
                     {'index': i, 'chunk': SOURCE.name + f'.part{i:02}', 'bytes': len(data), 'sha256': sha(data)}
                     for i, data in enumerate(chunks, 1)]}
    common = {'TRANSPORT.json': (json.dumps(transport, ensure_ascii=False, indent=2) + '\n').encode(),
              'reassemble-voc-database.py': (ROOT / 'scripts/reassemble-voc-database.py').read_bytes(),
              'README.md': (ROOT / 'docs/voc-database-delivery-20261106/发给运维001-数据库分包投递.md').read_bytes()}
    parts = []
    for i, data in enumerate(chunks, 1):
        p = OUT / (SOURCE.stem + f'-part{i:02}.zip')
        with zipfile.ZipFile(p, 'x') as z:
            for name, value in {**common, transport['parts'][i-1]['chunk']: data}.items():
                info = zipfile.ZipInfo(name, (2026, 11, 6, 0, 0, 0))
                info.external_attr = 0o100600 << 16
                z.writestr(info, value, compress_type=zipfile.ZIP_STORED if name.endswith(f'.part{i:02}') else zipfile.ZIP_DEFLATED)
        with zipfile.ZipFile(p) as z:
            assert z.testzip() is None and len(z.namelist()) == 4
        assert p.stat().st_size < 20000000
        parts.append({'filename': p.name, 'bytes': p.stat().st_size, 'sha256': sha(p.read_bytes())})
    spec = importlib.util.spec_from_file_location('reassemble', ROOT / 'scripts/reassemble-voc-database.py')
    reassembler = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(reassembler)
    inputs = [OUT / p['filename'] for p in parts]
    target = WORK / SOURCE.name
    result = reassembler.reassemble(list(reversed(inputs)), target)
    assert result['ok'] and result['fileCreated'] and target.read_bytes() == source
    stamp = target.stat().st_mtime_ns
    repeat = reassembler.reassemble(inputs, target)
    assert repeat['operation'] == 'unchanged' and target.stat().st_mtime_ns == stamp
    for bad, code in ((inputs[:1], 'EXACTLY_TWO_PARTS_REQUIRED'), ([inputs[0], inputs[0]], 'PART_MISSING_OR_DUPLICATED')):
        try:
            reassembler.reassemble(bad, WORK / SOURCE.name)
        except ValueError as exc:
            assert str(exc) == code
        else:
            raise AssertionError('BAD_INPUT_WAS_ACCEPTED')
    # A same-size altered chunk must fail and must not leave a final or temporary file.
    corrupt = WORK / 'corrupt-part01.zip'
    with zipfile.ZipFile(inputs[0]) as original, zipfile.ZipFile(corrupt, 'x') as changed:
        for name in original.namelist():
            data = original.read(name)
            if name.endswith('.part01'):
                data = bytes([data[0] ^ 1]) + data[1:]
            changed.writestr(name, data)
    bad_dir = WORK / 'bad-output'
    bad_dir.mkdir()
    try:
        reassembler.reassemble([corrupt, inputs[1]], bad_dir / SOURCE.name)
    except ValueError as exc:
        assert str(exc) == 'PART_HASH_MISMATCH'
    else:
        raise AssertionError('CORRUPTED_PART_WAS_ACCEPTED')
    assert not list(bad_dir.iterdir())
    receipt = {'ok': True, 'source': {'filename': SOURCE.name, 'bytes': len(source), 'sha256': sha(source)},
               'attachments': parts, 'sourceManifestVerified': '16/16', 'transportZipEntriesEach': 4,
               'allAttachmentsUnder20000000Bytes': True,
               'validation': {'byteIdenticalReassembly': True, 'reversedInputs': True, 'repeatNoOpMtimePreserved': True,
                              'missingAndDuplicatedInputsRejected': True, 'tamperedPartRejectedNoFinalFile': True},
               'installed': False, 'productionWrites': 0, 'originalPackageModified': False}
    (OUT / 'transport-receipt.json').write_text(json.dumps(receipt, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    shutil.copyfile(ROOT / 'docs/voc-database-delivery-20261106/发给运维001-数据库分包投递.md', OUT / '发给运维001-数据库分包投递.md')
    shutil.copyfile(ROOT / 'scripts/reassemble-voc-database.py', OUT / 'reassemble-voc-database.py')
    print(json.dumps(receipt, ensure_ascii=False, indent=2))

if __name__ == '__main__':
    main()