kangshifu-board/scripts/prepare-product-completion.py

"""Verify a completion delivery and reconstruct fresh batches from source archives.

Runs offline. Never installs tools, writes a library registry or publishes data.
"""
import argparse
import hashlib
import json
from pathlib import Path, PurePosixPath
import sys
import zipfile

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE / 'voc-library'))
if not (HERE / 'voc-library').is_dir():
    sys.path.insert(0, str(HERE.parent / 'server/voc-library'))
import library_channel
import structured_batch


def sha(path):
    h = hashlib.sha256()
    with Path(path).open('rb') as stream:
        for block in iter(lambda: stream.read(1024 * 1024), b''):
            h.update(block)
    return h.hexdigest()


def prepare(package, output):
    package, output = Path(package).resolve(), Path(output).absolute()
    expected_files = []
    for line in (package / 'FILES.sha256').read_text('utf-8').splitlines():
        fingerprint, rel = line.split('  ', 1)
        assert not PurePosixPath(rel).is_absolute() and '..' not in PurePosixPath(rel).parts and '\\' not in rel
        p = package / rel
        assert p.resolve().is_relative_to(package) and not p.is_symlink() and sha(p) == fingerprint, 'PACKAGE_HASH_MISMATCH:' + rel
        expected_files.append(rel)
    actual = {p.relative_to(package).as_posix() for p in package.rglob('*') if p.is_file()}
    assert actual == set(expected_files) | {'FILES.sha256'}, 'PACKAGE_FILE_SET_MISMATCH'
    release = json.loads((package / 'release.json').read_bytes())
    assert not output.exists() and not any(p.is_symlink() for p in (output, *output.parents)), 'NEW_OUTPUT_DIRECTORY_REQUIRED'
    output.mkdir(parents=True, mode=0o700)
    reports = []
    for batch in release['batches']:
        name = batch['batch']
        archive = package / 'batches' / name / 'staging.source.zip'
        assert sha(archive) == batch['sourceArchiveSha256']
        directory = output / name
        directory.mkdir(mode=0o700)
        with zipfile.ZipFile(archive) as z:
            assert len(z.namelist()) == len(set(z.namelist())), 'DUPLICATE_ARCHIVE_MEMBER'
            assert sum(i.file_size for i in z.infolist()) <= library_channel.MAX_EXPANDED
            manifest_raw = z.read('agent-batch/batch.json')
            assert hashlib.sha256(manifest_raw).hexdigest() == batch['batchManifestSha256']
            manifest = json.loads(manifest_raw)
            expected = {s['path']: s['sha256'] for s in manifest['sources']}
            expected.update({a['path']: a['sha256'] for a in manifest.get('attachments', [])})
            (directory / 'batch.json').write_bytes(manifest_raw)
            for rel, hashed in expected.items():
                assert rel.startswith('raw/') and ':' not in rel and '\\' not in rel and '..' not in PurePosixPath(rel).parts
                raw = z.read('agent-batch/02_采集档案/' + rel)
                assert hashlib.sha256(raw).hexdigest() == hashed
                path = directory / rel
                path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
                path.write_bytes(raw)
        _, _, _, rows, assets = structured_batch.load(directory / 'batch.json')
        packed, fingerprint = library_channel.pack(directory / 'batch.json')
        assert fingerprint == batch['channelArchiveSha256'], 'CHANNEL_RECONSTRUCTION_MISMATCH'
        report = {'batch': name, 'ok': True, 'sourcesVerified': len(assets),
                  'counts': {k: len(v) for k, v in rows.items()}, 'channelArchiveSha256': fingerprint,
                  'channelArchiveBytes': len(packed), 'batchPath': str(directory / 'batch.json')}
        reports.append(report)
    result = {'ok': True, 'packageFilesVerified': len(expected_files), 'batches': reports,
              'networkRequests': 0, 'productionWrites': 0, 'paidCalls': 0}
    (output / 'prepared.json').write_text(json.dumps(result, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    return result


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--package', required=True)
    parser.add_argument('--output', required=True)
    args = parser.parse_args()
    try:
        print(json.dumps(prepare(args.package, args.output), ensure_ascii=False, indent=2))
    except Exception as error:
        print(json.dumps({'ok': False, 'error': str(error), 'productionWrites': 0}, ensure_ascii=False), file=sys.stderr)
        sys.exit(1)