kangshifu-board/scripts/package-unified-database.py

"""Freeze an explicit runtime whitelist; exclude databases, secrets and business binaries."""
import argparse
import ast
import hashlib
import json
from pathlib import Path
import zipfile

ROOT = Path(__file__).resolve().parents[1]
NAME = 'ksf-unified-database-migration-20261109-r3'
MODULES = '''agent_batch board_library industry_decisions industry_innovation industry_monitor
industry_policy industry_research industry_workshop intake_contract library_audit library_channel
library_receiver monitoring_details monitoring_dispatch monitoring_evidence monitoring_inbox
monitoring_ledger monitoring_pipeline monitoring_policy monitoring_store monitoring_tasks
policy_publications policy_usage_review product_insights query_voc ranking_discovery retrieval
review_correction schema_catalog structured_batch unified_store unified_worker update_library voc_db'''.split()
COMMANDS = '''migrate-unified-database verify-unified-parity install-unified-skills
manage-industry-pipeline manage-industry-tasks manage-industry-dispatch register-industry-evidence
register-industry-details register-industry-policy register-industry-ranking-identities
check-monitoring-evidence compare-industry-rankings discover-industry-rankings
prepare-industry-inbox export-industry-snapshot'''.split()


def sha(raw):
    return hashlib.sha256(raw).hexdigest()


def main():
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument('--out', required=True, type=Path, help='New delivery directory')
    a = p.parse_args()
    assert not a.out.exists(), 'NEW_DELIVERY_DIRECTORY_REQUIRED'
    evidence = ROOT / 'work/single-database-20261109'
    for name in ('unified-test-result.json', 'regression-result.json', 'http-test-result.json'):
        assert json.loads((evidence / name).read_text('utf-8'))['success'], name
    assert json.loads((evidence / 'cli-full/04-parity.json').read_text('utf-8'))['ok']
    assert json.loads((evidence / 'component-installer-result-r2.json').read_text('utf-8'))['success']
    assert json.loads((evidence / 'r3/validation.json').read_text('utf-8'))['success']
    files = {}

    def copy(source, target=None):
        files[target or source] = (ROOT / source).read_bytes()

    for n in MODULES:
        copy('server/voc-library/' + n + '.py')
    for n in ('schema.sql', 'taxonomy.json'):
        copy('server/voc-library/' + n)
    for n in ('library-api', 'industry-api', 'industry-service', 'unified-worker'):
        copy('server/' + n + '.mjs')
    for n in COMMANDS:
        copy('scripts/' + n + '.py')
    copy('docs/unified-database-migration-20261109/执行单.md', '执行单.md')
    copy('docs/unified-database-migration-20261109/本地验收.md', '本地验收.md')
    copy('docs/unified-database-migration-20261109/unified-db.example.json', 'deploy/unified-db.example.json')
    copy('docs/unified-database-migration-20261109/requirements-unified.txt', 'requirements-unified.txt')
    protected = []
    for skill in ('ksf-voc-library', 'ksf-workshop-runtime'):
        base = ROOT / 'skills' / skill
        component_manifest = {}
        for line in (base / 'FILES.sha256').read_text('utf-8').splitlines():
            digest, name = line.split('  ', 1)
            assert sha((base / name).read_bytes()) == digest, (skill, name)
            if skill == 'ksf-voc-library':
                if name.startswith('scripts/'):
                    copy('skills/' + skill + '/' + name, 'components/research-runtime/' + name)
                    component_manifest[name] = digest
                continue
            key = 'skills/' + skill + '/' + name
            if skill == 'ksf-workshop-runtime' and (name.startswith('scripts/') or name == 'runtime-lock.json'):
                protected.append({'path': name, 'sha256': digest})
                continue
            copy(key)
        if skill == 'ksf-voc-library':
            files['components/research-runtime/FILES.sha256'] = ''.join(h + '  ' + n + '\n' for n, h in sorted(component_manifest.items())).encode('utf-8')
        else:
            copy('skills/' + skill + '/FILES.sha256')
    assert len(protected) == 7
    checks = ['unified-test-result.json', 'regression-result.json', 'http-test-result.json',
              'workshop-test-result.json', 'skill-static-verification-r2.json', 'component-installer-result-r2.json',
              'cli-full/01-plan/migration-receipt.json', 'cli-full/02-import/migration-receipt.json',
              'cli-full/03-verify/migration-receipt.json', 'cli-full/04-parity.json', 'runtime-measurement.json']
    for name in checks:
        copy('work/single-database-20261109/' + name, 'evidence/history-r2/' + name)
    for name in ('validation.json', 'parity.json', 'plan-final/migration-receipt.json', 'apply-final/migration-receipt.json', 'verify-final/migration-receipt.json', 'store-tests.log', 'installer-probe-tests.log', 'http-node-tests.log', 'rehearsal-corrections.json', 'receiver-regression.log', 'receiver-offline-service-tests.log', 'skill-static-verification.json'):
        copy('work/single-database-20261109/r3/' + name, 'evidence/r3/' + name)
    copy('docs/unified-database-migration-20261109/operations-readonly-identity-confirmed-r3.json', 'evidence/r3/operations-readonly-identity-confirmed.json')
    # Provenance hashes do not substitute for the separately shipped operational checks.
    for name, body in files.items():
        assert not name.endswith(('.sqlite', '.zip', '.env', '.pem', '.key'))
        if name.endswith('.py'):
            ast.parse(body.decode('utf-8-sig'), filename=name, feature_version=(3, 9))
    release = {'release': NAME, 'schema': 'ksf.unified-db.v1', 'database': 'FKieRbIoGt_master', 'role': 'FKieRbIoGt_admin',
               'appId': 'FKieRbIoGt_main', 'supersedes': 'ksf-unified-database-migration-20261109-r2', 'strictTargetGuardRetained': True,
               'workshopSkillVersion': 'r9', 'productionDeployed': False, 'sourceDataIncluded': False,
               'connectionCredentialsIncluded': False, 'businessBinariesIncluded': False,
               'protectedInstalledBusinessFiles': protected, 'payloadFiles': len(files),
               'workshopSkillEntrypoints': ['ksf-workshop-runtime'],
               'researchRuntimeComponent': 'components/research-runtime',
               'legacyResearchSkillEntryDistributed': False,
               'localValidationScope': 'isolated PostgreSQL; older local library; production acceptance pending'}
    entries = [n for n in files if n.endswith('/SKILL.md')]
    assert entries == ['skills/ksf-workshop-runtime/SKILL.md'], entries
    files['release.json'] = (json.dumps(release, ensure_ascii=False, indent=2) + '\n').encode('utf-8')
    files['FILES.sha256'] = ''.join(sha(body) + '  ' + name + '\n' for name, body in sorted(files.items())).encode('utf-8')
    a.out.mkdir(parents=True)
    package = a.out / 'package'
    for name, raw in files.items():
        path = package / name;path.parent.mkdir(parents=True, exist_ok=True);path.write_bytes(raw)
    archive = a.out / (NAME + '.zip')
    with zipfile.ZipFile(archive, 'w', zipfile.ZIP_DEFLATED) as z:
        for name, raw in sorted(files.items()):
            z.writestr(name, raw)
    with zipfile.ZipFile(archive) as z:
        assert z.testzip() is None
        assert set(z.namelist()) == set(files)
        for line in z.read('FILES.sha256').decode('utf-8').splitlines():
            digest, name = line.split('  ', 1)
            assert sha(z.read(name)) == digest
    receipt = {**release, 'zip': str(archive.absolute()), 'sha256': sha(archive.read_bytes()),
               'bytes': archive.stat().st_size, 'entriesVerified': len(files)}
    (a.out / 'delivery-receipt.json').write_text(json.dumps(receipt, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    print(json.dumps(receipt, ensure_ascii=False))


if __name__ == '__main__':
    main()