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

"""Package verified sources, staging databases and an offline preparation tool."""
import hashlib
import json
from pathlib import Path
import shutil
import subprocess
import sys
import zipfile

ROOT = Path(__file__).resolve().parents[1]
WORK = ROOT / 'work/product-completion-20261108-r1'
DOCS = ROOT / 'docs/product-completion-20261108'
OUT = ROOT / 'artifacts/ksf-product-completion-20261108-r1'


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


def read(path):
    return json.loads(path.read_bytes())


def copy(source, destination):
    destination.parent.mkdir(parents=True, exist_ok=True)
    shutil.copyfile(source, destination)


def main():
    assert not OUT.exists(), 'NEW_DELIVERY_DIRECTORY_REQUIRED'
    OUT.mkdir(parents=True)
    package = OUT / 'package'
    checked = read(DOCS / 'staging-readback.json')
    assert checked['ok'] and checked['productionImported'] is False
    plan = read(WORK / 'batch-plan.json')
    release = {'schema': 'ksf.product-completion-delivery.v1', 'baselineGeneration': plan['generation'],
               'productionImported': False, 'productionPublishedByThisMachine': False,
               'reason': 'CHANNEL_NOT_CONFIGURED', 'stagingVerified': True,
               'paidCalls': 0, 'productionWrites': 0, 'newSpecRecords': 648, 'newReviewRecords': 21,
               'contextTitleSpecsAlreadyExisting': 650,
               'baselineCounts': read(DOCS / 'baseline.json')['counts'],
               'expectedAfterCounts': {'products': 4674, 'reviews': 3475, 'prices': 4172, 'specs': 5422},
               'productsWithReviewsBeforeAndExpectedAfter': 166,
               'fullReviewCoverageClaimed': False,
               'sourceAttributeSpecProductCoverageBefore': 2,
               'sourceAttributeSpecProductCoverageExpectedAfter': 650,
               'batches': checked['batches'],
               'verifiedSpecRecordIds': checked['verifiedSpecRecordIds'],
               'verifiedReviewRecords': checked['verifiedReviewRecords']}
    for b, source_part in zip(release['batches'], plan['batches']):
        b['expectedProductionAddedSpecs'] = source_part['specs']
        b['expectedProductionAddedReviews'] = source_part.get('reviews', 0)
        b['expectedProductionAddedProducts'] = 0
        b['expectedProductionAddedPrices'] = 0
        directory = WORK / source_part['directory']
        for name in ['batch.json', 'staging.sqlite', 'staging.source.zip', 'preflight.stdout.json',
                     'preflight.stderr.txt', 'preflight.exit.json', 'staging-build.stdout.json',
                     'staging-build.stderr.txt', 'staging-build.exit.json', 'readback.json']:
            copy(directory / name, package / 'batches' / b['batch'] / name)
    for name in ['baseline.json', 'public-probe.json', 'public-probe-corrected.json', 'review-recovery-audit.json',
                 'spec-recovery-plan.json', 'generated-title-spec-duplicates.json', 'staging-readback.json']:
        copy(DOCS / name, package / 'audit' / name)
    copy(WORK / 'batch-plan.json', package / 'audit/batch-plan.json')
    for path in sorted((WORK / 'current-spec-details').glob('*.json')):
        copy(path, package / 'audit/current-spec-details' / path.name)
    for name in ['channel-dry.stdout.json', 'channel-dry.stderr.json', 'channel-dry.exit.json']:
        copy(WORK / name, package / 'audit' / name)
    copy(DOCS / '入库执行单.md', package / '入库执行单.md')
    copy(ROOT / 'scripts/prepare-product-completion.py', package / 'tools/prepare-product-completion.py')
    runtime = ['agent_batch.py', 'structured_batch.py', 'intake_contract.py', 'voc_db.py', 'query_voc.py',
               'retrieval.py', 'schema.sql', 'taxonomy.json', 'schema_catalog.py', 'product_insights.py',
               'update_library.py', 'library_channel.py', 'review_correction.py']
    for name in runtime:
        copy(ROOT / 'server/voc-library' / name, package / 'tools/voc-library' / name)
    test = subprocess.run([sys.executable, '-X', 'utf8', '-B', '-m', 'unittest', 'discover', '-s',
                           str(ROOT / 'scripts'), '-p', 'test_product_completion.py', '-v'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    assert test.returncode == 0
    (package / 'audit/extraction-tests.stdout.txt').write_bytes(test.stdout)
    (package / 'audit/extraction-tests.stderr.txt').write_bytes(test.stderr)
    (package / 'release.json').write_text(json.dumps(release, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    files = sorted(p for p in package.rglob('*') if p.is_file())
    (package / 'FILES.sha256').write_text(''.join(sha(p) + '  ' + p.relative_to(package).as_posix() + '\n' for p in files), encoding='utf-8')
    archive = OUT / 'ksf-product-completion-20261108-r1.zip'
    with zipfile.ZipFile(archive, 'x') as z:
        for p in sorted(package.rglob('*')):
            if not p.is_file():
                continue
            rel = p.relative_to(package).as_posix()
            info = zipfile.ZipInfo(rel, (2026, 11, 8, 0, 0, 0))
            info.external_attr = 0o100600 << 16
            z.writestr(info, p.read_bytes(), compress_type=zipfile.ZIP_STORED if p.suffix == '.zip' else zipfile.ZIP_DEFLATED)
    # Reopen the delivered bytes and verify every file against the manifest.
    with zipfile.ZipFile(archive) as z:
        manifest = z.read('FILES.sha256').decode('utf8')
        assert len(z.namelist()) == len(set(z.namelist())) == len(files) + 1
        for line in manifest.splitlines():
            hashed, rel = line.split('  ', 1)
            assert hashlib.sha256(z.read(rel)).hexdigest() == hashed
    receipt = {'ok': True, 'archive': str(archive), 'bytes': archive.stat().st_size, 'sha256': sha(archive),
               'filesVerified': len(files), 'stagingVerified': True, 'productionImported': False,
               'specsPrepared': 648, 'reviewsPrepared': 21, 'paidCalls': 0, 'productionWrites': 0}
    (OUT / 'delivery.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()