kangshifu-board/scripts/verify-unified-parity.py

"""Compare the actual old library and new PostgreSQL binding before cutover."""
from contextlib import closing
import json
import os
from pathlib import Path
import sys
import argparse
sys.dont_write_bytecode = True
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'server/voc-library'))
import board_library as board
import industry_monitor as industry
import unified_store as store
import voc_db
import retrieval


def main():
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument('--config', required=True)
    s = p.add_mutually_exclusive_group(required=True)
    s.add_argument('--registry'); s.add_argument('--db')
    p.add_argument('--monitoring-store', required=True)
    p.add_argument('--policy-usage-review');p.add_argument('--policy-usage-sha256')
    p.add_argument('--out', required=True)
    a = p.parse_args()
    old_env = os.environ.pop('KSF_UNIFIED_DB_CONFIG', None)
    original = {'registry': a.registry, 'db': a.db, 'monitoringStore': a.monitoring_store,
                'policyUsageReview': a.policy_usage_review, 'policyUsageSha256': a.policy_usage_sha256}
    new = {**original, 'unifiedConfig': a.config}
    checks = []
    try:
        summary = board.run(original, {'op': 'summary'})
        requests = [{'op': 'summary'}, {'op': 'version'}]
        for entity in board.LABELS:
            for quality in ('analysis', 'audit'):
                requests.append({'op': 'browse', 'entity': entity, 'quality': quality, 'limit': 3})
            r = board.run(original, {'op': 'browse', 'entity': entity, 'quality': 'audit', 'limit': 1})
            if r['records']:
                requests.append({'op': 'detail', 'entity': entity, 'id': r['records'][0]['record_id']})
        for sort in ('reviews_asc', 'reviews_desc', 'price_asc', 'price_desc'):
            requests.append({'op': 'browse', 'entity': 'products', 'sort': sort, 'page': 2, 'limit': 24})
        for entity in ('products', 'reviews'):
            requests.append({'op': 'browse', 'entity': entity, 'keyword': '味道' if entity == 'reviews' else '拌面',
                             'filters': {'brand': 'ksf', 'category': 'banmian'}})
        for req in requests:
            before = board.run(original, req)
            after = board.run(new, req)
            store.need(before == after, 'LIBRARY_PARITY_FAILED')
            checks.append({'channel': 'library', 'op': req['op'], 'entity': req.get('entity'), 'equal': True})
        # The Python industry path also reads monitoring via the unified binding.
        for op in ('contract', 'overview', 'stores', 'listings', 'reviews', 'policies', 'rankings', 'plan', 'insights', 'workshop'):
            req = {'op': op}
            before = industry.run(original, req)
            os.environ['KSF_UNIFIED_DB_CONFIG'] = a.config
            try:
                after = industry.run(new, req)
            finally:
                os.environ.pop('KSF_UNIFIED_DB_CONFIG', None)
            store.need(before == after, 'INDUSTRY_PARITY_FAILED')
            checks.append({'channel': 'industry', 'op': op, 'equal': True})
        # Exact original source payloads, not merely the browser's redacted citation.
        db, _ = board.active_database(original)
        with voc_db.readonly(db) as source, store.read_research(store.research_reference(a.config)) as target:
            ids = [r[0] for r in source.execute('SELECT id FROM source_records ORDER BY id LIMIT 10')]
            for rid in ids:
                store.need(voc_db.get_source(source, rid) == voc_db.get_source(target, rid), 'SOURCE_PARITY_FAILED')
        result = {'ok': True, 'generation': summary['generation'], 'checks': checks, 'originalSourcesCompared': len(ids),
                  'readOnly': True, 'databaseWrites': 0, 'businessRowsChanged': 0}
        out = Path(a.out);out.parent.mkdir(parents=True, exist_ok=True)
        with out.open('x', encoding='utf-8') as f:
            f.write(json.dumps(result, ensure_ascii=False, indent=2) + '\n')
        out.chmod(0o600)
        print(json.dumps({'ok': True, 'checks': len(checks), 'originalSourcesCompared': len(ids), 'receipt': str(out)}))
        return 0
    except Exception as e:
        code = str(e) if isinstance(e, store.StoreError) else 'PARITY_FAILED'
        print(json.dumps({'ok': False, 'error': {'code': code}, 'readOnly': True, 'databaseWrites': 0}))
        return 1
    finally:
        if old_env is not None:
            os.environ['KSF_UNIFIED_DB_CONFIG'] = old_env


if __name__ == '__main__':
    sys.exit(main())