kangshifu-board/scripts/package-product-overview.py

"""Freeze the committed product overview 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]
BASE = 'f4bdaa70a1f9ceac7c44f79a32d7454143875fd5'
NAME = 'ksf-product-overview-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')]
DEPS = ['server/voc-library/' + n for n in ('voc_db.py', '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)))
    old = zipfile.ZipFile(io.BytesIO(subprocess.check_output(['git', 'archive', '--format=zip', BASE], cwd=ROOT)))
    out = ROOT / 'artifacts/商品评价概况与入库更新-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 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 name in ('ksf-workshop-runtime', 'ksf-voc-library')}
    js('baselines.json', baselines)
    def entry(name):
        prior = old.read(name) if name in old.namelist() else None
        return dict(path=name, sha256=sha(frozen.read(name)), priorNormalizedSha256=[norm(prior)] if prior else [], allowMissing=prior is None)
    manifest = dict(version='ksf.product-overview.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-product-overview')
    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/product-overview-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)
    with tempfile.TemporaryDirectory(prefix='installer-check-', dir=package.parent) as td:
        data = Path(td)
        for name in baselines:
            for rel in baselines[name][0]:
                target = data / 'skills' / name / rel; target.parent.mkdir(parents=True, exist_ok=True)
                target.write_bytes(old.read('skills/' + name + '/' + rel))
        plan = installer.install(package, data)
        assert plan['filesChecked'] == 25 and len(plan['changed']) == 6
        assert all(n.startswith('ksf-voc-library/') for n in plan['changed'])
        assert installer.install(package, data, True)['filesystemVerified']
        assert not installer.install(package, data, True)['applied']
    # 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, baselineCommit=BASE, skills=skills,
        staticFiles=2, backendFiles=4, databaseIncluded=False, databaseMigrationRequired=False,
        credentialsIncluded=False, 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')
    print(json.dumps(receipt, ensure_ascii=False, indent=2))

if __name__ == '__main__': main()