kangshifu-board/scripts/package-voc-product-fields.py

"""Seal the fixed product restoration with verified existing evidence; no network."""
import argparse
import ast
import hashlib
import importlib.util
import json
from pathlib import Path
import shutil
import sys
import zipfile

ROOT = Path(__file__).resolve().parents[1]


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


def main():
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument('--out', type=Path, default=ROOT / 'artifacts/商品历史缺项恢复-20261106-r1')
    a = p.parse_args()
    out = a.out.absolute()
    if out.exists():
        raise ValueError('NEW_PACKAGE_DIRECTORY_REQUIRED')
    work = ROOT / 'work/voc-product-restoration-20261106-r1'
    package = out / 'package'
    shutil.copytree(work / 'prepared', package)
    tool = ROOT / 'scripts/restore-voc-product-fields.py'
    shutil.copyfile(tool, package / tool.name)
    instruction = ROOT / 'docs/industry-voc-monitoring-20261106/30-发给运维001-商品历史缺项定向恢复.md'
    shutil.copyfile(instruction, package / '发给运维001.md')
    shutil.copyfile(instruction, out / '发给运维001.md')
    (package / 'verification').mkdir()
    for name in ('local-simulation.json', 'source-availability.json'):
        shutil.copyfile(work / name, package / 'verification' / name)
    local = json.loads((work / 'local-simulation.json').read_text(encoding='utf-8'))
    assert local['ok'] and all(local['checks'].values())
    assert local['before']['analysis'] == local['after']['analysis'] == 771
    assert local['before']['strict'] == local['after']['strict'] == {'analysis': 18, 'audit': 39}
    release = json.loads((package / 'RELEASE.json').read_text(encoding='utf-8'))
    release.update(pythonMinimum='3.9', python39GrammarVerified=True, nativePython39RuntimeVerified=False,
        productionInstalled=False, productionImported=False, databaseIncluded=False, credentialsIncluded=False,
        skillFiles=0, staticFiles=0, backendModuleReplacements=0, collectionCalls=0,
        expectedGeneration='8857847197722f25c7392c4df85c8ab4427f3329a80a1fd36aad585f123eae93',
        expectedDatabaseSha256='634e7809f26d4bcf6422da03bb4d5742e279ae1c787d770531953245eeaf0754',
        productsChanged=60, fieldCounts=local['fieldCounts'], localSimulationChecks=local['checks'],
        localFixtureIsProductionDatabase=False, targetProductionDryRequired=True,
        currentKnownConflictsOverwritten=0, originalReferencesVerified=152, originalReferencesStillUnavailable=73,
        fixedToolSha256=sha(tool.read_bytes()),
        testCommand='py -3.14 -X utf8 -B scripts/test_restore_voc_product_fields.py -v',
        testResult={'tests': 5, 'passed': 5,
            'coverage': ['raw original identity and exact fields', 'staged restore preserves all other business rows',
                'changed current row rejected', 'dry apply status replay rollback and publication guards',
                'tampered source rejected']})
    (package / 'RELEASE.json').write_text(json.dumps(release, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    ast.parse(tool.read_text(encoding='utf-8'), feature_version=(3, 9))
    spec = importlib.util.spec_from_file_location('sealed_repair', package / tool.name)
    repair = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(repair)
    v, u = repair.module_check(ROOT / 'server/voc-library')
    plan = repair.plan(package, v)
    assert len(plan['products']) == 60 and sum(plan['fields'].values()) == 70
    (package / 'verification/fixed-field-plan.json').write_text(json.dumps(plan, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    files = sorted(f for f in package.rglob('*') if f.is_file())
    assert not any(f.suffix in ('.sqlite', '.db', '.key', '.pyc') or f.name.startswith('.env') for f in files)
    for f in files:
        raw = f.read_bytes()
        import re
        assert not re.search(rb'\bsk-[A-Za-z0-9_-]{20,}|Bearer\s+[A-Za-z0-9._-]{20,}', raw), 'POSSIBLE_CREDENTIAL:' + str(f)
    manifest = ''.join(sha(f.read_bytes()) + '  ' + f.relative_to(package).as_posix() + '\n' for f in files)
    (package / 'FILES.sha256').write_text(manifest, encoding='utf-8')
    archive = out / 'ksf-voc-product-fields-20261106-r1.zip'
    with zipfile.ZipFile(archive, 'x', zipfile.ZIP_DEFLATED, compresslevel=9) as z:
        for f in [*files, package / 'FILES.sha256']:
            z.write(f, f.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():
            expected, name = line.split('  ', 1)
            assert sha(z.read(name)) == expected
    receipt = {'ok': True, 'fileName': archive.name, 'bytes': archive.stat().st_size,
               'sha256': sha(archive.read_bytes()), 'filesVerified': len(files), 'zipEntries': len(files) + 1,
               'productionImported': False, 'nativePython39RuntimeVerified': False,
               'localTestsPassed': 5, 'productsChanged': 60, 'businessMissingFields': 7,
               'historicalCoverageFields': 60, 'supportingClassificationFields': 3, 'collectionCalls': 0}
    (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, indent=2))


if __name__ == '__main__':
    main()