kangshifu-board/scripts/package-full-schema-intake.py

"""Freeze the cumulative full-schema / ingestion patch, without private databases or credentials."""
import argparse
import hashlib
import importlib.util
import io
import json
from pathlib import Path
import shutil
import subprocess
import tempfile
import zipfile

ROOT = Path(__file__).resolve().parents[1]
BASES = ['f4bdaa70a1f9ceac7c44f79a32d7454143875fd5', 'bc5a0092890803d96d790ab782cd9b1069661821']
BASE = BASES[0]
NAME = 'ksf-full-schema-intake-20261106-r1'
FRONT = ['src/core/library.js', 'src/pages/library-product-detail.js']
BACK = ['server/voc-library/' + n for n in ('board_library.py', 'product_insights.py', 'update_library.py', 'agent_batch.py', 'voc_db.py', 'schema_catalog.py', 'intake_contract.py', 'structured_batch.py')]
DEPS = ['server/voc-library/' + n for n in ('retrieval.py', 'schema.sql', 'taxonomy.json')]
sha = lambda b: hashlib.sha256(b).hexdigest()
norm = lambda b: sha(b.replace(b'\r\n', b'\n'))

def main():
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument('--commit', default='HEAD'); p.add_argument('--homepage-file', required=True)
    args = p.parse_args()
    commit = subprocess.check_output(['git', 'rev-parse', args.commit + '^{commit}'], cwd=ROOT).decode().strip()
    subprocess.run(['git', 'merge-base', '--is-ancestor', BASE, commit], cwd=ROOT, check=True)
    frozen = zipfile.ZipFile(io.BytesIO(subprocess.check_output(['git', 'archive', '--format=zip', commit], cwd=ROOT)))
    olds = [zipfile.ZipFile(io.BytesIO(subprocess.check_output(['git', 'archive', '--format=zip', b], cwd=ROOT))) for b in BASES]
    old = olds[0]
    out = ROOT / 'artifacts/全表字段与Agent入库-20261106'
    package = ROOT / 'work' / NAME / 'package'
    if package.exists() or (out / (NAME + '.zip')).exists(): raise ValueError('RELEASE_ALREADY_EXISTS')
    package.mkdir(parents=True)
    def write(name, body):
        dest = package / name; dest.parent.mkdir(parents=True, exist_ok=True); dest.write_bytes(body)
    def js(name, value): write(name, (json.dumps(value, ensure_ascii=False, indent=2) + '\n').encode())
    for name in FRONT + BACK: write('payload/' + name, frozen.read(name))
    for n in ('ksfGetDataDictionary.js','manifest.json'): write('functions/' + n, (ROOT / 'server/dist/data-dictionary' / n).read_bytes())
    for n in ('postgres.mjs','core.mjs','data-dictionary.mjs'):
        assert (ROOT/'server/src'/n).read_bytes().replace(b'\r\n',b'\n') == frozen.read('server/src/'+n).replace(b'\r\n',b'\n'), 'FUNCTION_SOURCE_NOT_FROZEN'
    assert (ROOT/'server/references/business-data-dictionary.json').read_bytes().replace(b'\r\n',b'\n') == frozen.read('server/references/business-data-dictionary.json').replace(b'\r\n',b'\n')
    function_manifest = json.loads((package / 'functions/manifest.json').read_text(encoding='utf-8'))
    for item in function_manifest['functions']:
        assert sha((package / 'functions' / item['file']).read_bytes()) == item['sha256']
    write('发给王刚001.md', frozen.read('docs/full-schema-intake-20261106/发给王刚001.md'))
    for name in frozen.namelist():
        if name.startswith('skills/') and not name.endswith('/'): write(name, frozen.read(name))
    write('install-query-skills.py', frozen.read('scripts/install-query-skills.py'))
    baselines = {name: [{p.split('/', 2)[2]: sha(old.read(p)) for p in old.namelist() if p.startswith('skills/' + name + '/') and not p.endswith('/')} for old in olds]
                 for name in ('ksf-workshop-runtime', 'ksf-voc-library')}
    js('baselines.json', baselines)
    def entry(name):
        prior = [o.read(name) for o in olds if name in o.namelist()]
        return dict(path=name, sha256=sha(frozen.read(name)), priorNormalizedSha256=sorted(set(norm(b) for b in prior)), allowMissing=len(prior)<len(olds))
    manifest = dict(version='ksf.full-schema-intake.publish.v1', release=NAME,
        homepageSha256=json.loads(old.read('deploy/board-files-20261106.json'))['homepageSha256'], files=[entry(n) for n in FRONT])
    js('deploy/board-files.json', manifest); js('deploy/backend-files.json', dict(files=[entry(n) for n in BACK]))
    js('deploy/dependencies.json', dict(files=[dict(path=n, normalizedSha256=norm(frozen.read(n))) for n in DEPS]))
    publisher = frozen.read('scripts/publish-opportunity-ui.mjs').decode()
    start = publisher.index('export const FILES=Object.freeze(['); end = publisher.index(']);', start) + 3
    publisher = publisher[:start] + 'export const FILES=Object.freeze(' + json.dumps(FRONT) + ');' + publisher[end:]
    publisher = publisher.replace('ksf.opportunity-ui.publish.v1', manifest['version']).replace('ksf.opportunity-ui.20261103.v1', NAME)
    publisher = publisher.replace('filesChecked:6', 'filesChecked:FILES.length').replace('ksf-opportunity-ui', 'ksf-full-schema-intake')
    publisher = publisher.replace("sourceDir:root,manifestPath:path.join(root,'deploy/opportunity-ui-files-20261103.json')", "sourceDir:path.join(root,'payload'),manifestPath:path.join(root,'deploy/board-files.json')")
    write('scripts/publish-board.mjs', publisher.encode())
    write('发给运维001.md', frozen.read('docs/full-schema-intake-20261106/发给运维001.md'))
    skills = {}
    for name in baselines:
        folder = package / 'skills' / name
        lines = (folder / 'FILES.sha256').read_text(encoding='utf-8').splitlines()
        for line in lines:
            h, rel = line.split('  ', 1); assert sha((folder / rel).read_bytes()) == h
        skills[name] = dict(entries=len(lines), manifestSha256=sha((folder / 'FILES.sha256').read_bytes()))
    # Exercise the actual delivered installer against both frozen baseline skills.
    spec = importlib.util.spec_from_file_location('overview_skill_installer', package / 'install-query-skills.py')
    installer = importlib.util.module_from_spec(spec); spec.loader.exec_module(installer)
    checks=[]
    for index, old_version in enumerate(olds):
        with tempfile.TemporaryDirectory(prefix='installer-check-', dir=package.parent) as td:
            data=Path(td)
            for name in baselines:
                for rel in baselines[name][index]:
                    target=data/'skills'/name/rel;target.parent.mkdir(parents=True,exist_ok=True)
                    target.write_bytes(old_version.read('skills/'+name+'/'+rel))
            state=data/'ksf-workshop-state/sessions/existing.json';state.parent.mkdir(parents=True);state.write_text('existing-session')
            binding=data/'skills/ksf-workshop-runtime/credential-binding.json';binding.write_text('preserved-existing-binding')
            plan=installer.install(package,data)
            assert plan['filesChecked']==sum(x['entries']+1 for x in skills.values())
            assert installer.install(package,data,True)['filesystemVerified']
            assert not installer.install(package,data,True)['applied']
            assert state.read_text()=='existing-session' and binding.read_text()=='preserved-existing-binding'
            checks.append(dict(baselineCommit=BASES[index],filesChecked=plan['filesChecked'],changed=plan['changed'],changedCount=len(plan['changed']),applyVerified=True,noOpVerified=True))
    # Check the two-file publisher plan/apply/no-op against the actual baseline.
    with tempfile.TemporaryDirectory(prefix='static-check-', dir=package.parent) as td:
        target = Path(td)
        for name in ['index.html', *FRONT]:
            dest = target / name; dest.parent.mkdir(parents=True, exist_ok=True)
            dest.write_bytes(old.read(name))
        # The maintained homepage is external to Git's local preview page.
        index = Path(args.homepage_file).read_bytes()
        assert sha(index) == manifest['homepageSha256'], 'HOMEPAGE_MISMATCH'
        (target / 'index.html').write_bytes(index)
        cli = ['node', str(package / 'scripts/publish-board.mjs'), '--target-dir', str(target)]
        result = json.loads(subprocess.check_output(cli, cwd=ROOT))
        assert result['changes'] == FRONT
        assert json.loads(subprocess.check_output([*cli, '--apply'], cwd=ROOT))['filesystemVerified']
        assert not json.loads(subprocess.check_output([*cli, '--apply'], cwd=ROOT))['changes']
    js('RELEASE.json', dict(release=NAME, sourceCommit=commit, baselineCommits=BASES, skills=skills, installerChecks=checks, cloudFunctions=function_manifest['functions'],
        staticFiles=2, backendFiles=len(BACK), databaseIncluded=False, databaseMigrationRequired=False,
        credentialsIncluded=False, sharedRegistryBindingRequired=True, productionInstalled=False, productionBrowserVerified=False,
        installerCheck=True, publisherCheck=True))
    files = sorted(q for q in package.rglob('*') if q.is_file())
    assert not any(q.suffix in ('.sqlite', '.db', '.key', '.pyc') or q.name.startswith('.env') for q in files)
    write('FILES.sha256', ''.join(sha(q.read_bytes()) + '  ' + q.relative_to(package).as_posix() + '\n' for q in files).encode())
    out.mkdir(parents=True, exist_ok=True); archive = out / (NAME + '.zip')
    with zipfile.ZipFile(archive, 'x', zipfile.ZIP_DEFLATED, compresslevel=9) as z:
        for q in [*files, package / 'FILES.sha256']: z.write(q, q.relative_to(package).as_posix())
    with zipfile.ZipFile(archive) as z:
        assert z.testzip() is None
        for line in z.read('FILES.sha256').decode().splitlines():
            h, name = line.split('  ', 1); assert sha(z.read(name)) == h
    receipt = dict(ok=True, filename=archive.name, bytes=archive.stat().st_size, sha256=sha(archive.read_bytes()),
        entries=len(files) + 1, filesVerified=len(files), sourceCommit=commit, productionInstalled=False)
    (out / 'delivery-receipt.json').write_text(json.dumps(receipt, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    shutil.copyfile(package / '发给运维001.md', out / '发给运维001.md')
    shutil.copyfile(package / '发给王刚001.md', out / '发给王刚001.md')
    print(json.dumps(receipt, ensure_ascii=False, indent=2))

if __name__ == '__main__': main()