kangshifu-board/scripts/verify-industry-install.py

"""Read-only native-runtime preflight; never installs modules or changes a library pointer."""
import argparse
import hashlib
import json
from pathlib import Path
import platform
import sys

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / 'server/voc-library'))


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    sources = parser.add_mutually_exclusive_group(required=True)
    sources.add_argument('--snapshot')
    sources.add_argument('--registry')
    sources.add_argument('--db')
    parser.add_argument('--voc-runtime')
    parser.add_argument('--monitoring-store', dest='monitoringStore')
    parser.add_argument('--expected-generation', required=True)
    args = parser.parse_args()
    if not args.snapshot:
        if not args.voc_runtime or not (Path(args.voc_runtime) / 'board_library.py').is_file():
            raise ValueError('EXISTING_VOC_RUNTIME_REQUIRED')
        sys.path.insert(0, str(Path(args.voc_runtime).resolve()))
    from industry_monitor import load, Monitor, run
    config = {k: getattr(args, k) for k in ('snapshot', 'registry', 'db', 'monitoringStore') if getattr(args, k)}
    data = load(config)
    if data['generation'] != args.expected_generation:
        raise ValueError('LIBRARY_VERSION_CHANGED')
    monitor = Monitor(data)
    baseline = monitor.review_coverage({'platform': 'jd', 'shopId': '1000084104'})
    corpus = monitor.review_research({}, {'mode': 'analysis'})
    probes = {}
    for op in ('contract', 'listings', 'reviews', 'policies', 'rankings', 'plan', 'insights'):
        result = run(config, {'op': op, 'generation': data['generation']})
        probes[op] = result.get('ok') is True and result.get('databaseWrites') == 0
    # Reload the binding after all reads to reject a concurrent publication.
    if load(config)['generation'] != args.expected_generation:
        raise ValueError('LIBRARY_VERSION_CHANGED')
    fingerprints = {}
    if args.voc_runtime:
        for filename in ('board_library.py', 'voc_db.py', 'retrieval.py', 'schema.sql'):
            path = Path(args.voc_runtime) / filename
            if path.is_file():
                fingerprints[filename] = hashlib.sha256(path.read_bytes().replace(b'\r\n', b'\n')).hexdigest()
    print(json.dumps({'ok': all(probes.values()), 'readOnly': True, 'databaseWrites': 0, 'paidCalls': 0,
                      'runtime': platform.python_version(), 'nativePython39Verified': sys.version_info[:2] == (3, 9),
                      'generation': data['generation'], 'counts': data['summary']['counts'],
                      'primaryStore': {k: baseline[k] for k in ('products', 'reviewRows', 'uniqueReadableTexts')},
                      'analysisRowsRecomputed': corpus['total'], 'probes': probes,
                      'existingRuntimeFingerprints': fingerprints}, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    try:
        main()
    except Exception as error:
        print(json.dumps({'ok': False, 'error': str(error) if isinstance(error, ValueError) else type(error).__name__,
                          'databaseWrites': 0, 'paidCalls': 0}))
        sys.exit(1)