kangshifu-board/scripts/bootstrap-industry-evidence.py

"""Recheck existing eight-shop pages and delivered review originals into the auxiliary registry."""
import argparse
import csv
import hashlib
import json
from pathlib import Path
import sys

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / 'server/voc-library'))
from monitoring_store import import_run, projection, read_state, need


def load(path):
    return json.loads(path.read_text('utf-8-sig'))


def run(store, historical, reviews, generation):
    imported = []
    groups = {}
    for path in sorted([*(historical / 'originals').glob('*.json'), *(historical / 'competitors/originals').glob('*.json')]):
        doc = load(path)
        shop = str(doc['params']['shopId'])
        groups.setdefault(shop, []).append(path)
    need(len(groups) == 8, 'EIGHT_STORE_BASELINE_REQUIRED')
    for shop, paths in groups.items():
        pages = []
        for path in paths:
            doc = load(path)
            body = doc['result']['data']
            pages.append({'page': int(doc['params']['page']), 'requestKey': body['requestId'], 'status': 'success',
                          'rawFile': path.relative_to(historical).as_posix(), 'rawSha256': hashlib.sha256(path.read_bytes()).hexdigest(),
                          'mapping': {'itemsPointer': '/result/data/data/ItemList', 'idPointer': '/itemId', 'titlePointer': '/title',
                                      'pagePointer': '/result/data/requestParams/query/page', 'requestIdPointer': '/result/data/requestId',
                                      'scopePointers': {'shopId': '/result/data/data/shopId'},
                                      'observedAtPointer': '/result/data/recordTime', 'totalPointer': '/result/data/data/totalResults',
                                      'pageCountPointer': '/result/data/data/TotalPage',
                                      'successChecks': [{'pointer': '/httpStatus', 'equals': 200}, {'pointer': '/status', 'equals': 'ok'},
                                                        {'pointer': '/result/code', 'equals': 200}, {'pointer': '/result/data/code', 'equals': 0}]}})
        manifest = {'schema': 'ksf.monitoring-run.v1', 'kind': 'listings', 'platform': 'jd',
                    'scope': {'shopId': shop}, 'scopeName': body['data']['shopName'], 'window': 'historical-20261104-selected-pass',
                    'endpointId': 'jd.get_shop_item_list_v1', 'generation': generation, 'termination': 'exhausted', 'pages': sorted(pages, key=lambda p: p['page'])}
        imported.append(import_run(store, manifest, historical))
    with (reviews / 'manifest.csv').open(encoding='utf-8-sig', newline='') as stream:
        for item in csv.DictReader(stream):
            product = item['productId']
            path = reviews / 'originals' / (product.split(':')[1] + '.json')
            need(hashlib.sha256(path.read_bytes()).hexdigest() == item['sha256'], 'DELIVERED_REVIEW_SHA_MISMATCH')
            body = load(path)['data']
            page = {'page': int(item['page']), 'requestKey': body['requestId'], 'status': 'success',
                    'rawFile': path.relative_to(reviews).as_posix(), 'rawSha256': item['sha256'],
                    'mapping': {'itemsPointer': '/data/data/data', 'idPointer': '/commentId',
                                'pagePointer': '/data/requestParams/query/page', 'requestIdPointer': '/data/requestId',
                                'scopePointers': {'productId': '/data/requestParams/query/itemId'},
                                'rowScopePointers': {'productId': '/productId'},
                                'totalPointer': '/data/data/pageInfo/count', 'pageCountPointer': '/data/data/pageInfo/maxPage',
                                'observedAtPointer': '/data/recordTime',
                                'successChecks': [{'pointer': '/code', 'equals': 200}, {'pointer': '/data/code', 'equals': 0}]}}
            manifest = {'schema': 'ksf.monitoring-run.v1', 'kind': 'reviews', 'platform': 'jd', 'scope': {'productId': product, 'mode': 'interface_default'},
                        'scopeName': product, 'window': 'g2g3-20261106', 'generation': generation, 'endpointId': 'jd.get_item_comments_v1',
                        'termination': 'stopped', 'pages': [page]}
            imported.append(import_run(store, manifest, reviews))
    return {'ok': True, 'registered': len(imported), 'newRegistrations': sum(not r['alreadyRegistered'] for r in imported),
            'collectionCalls': 0, 'productionWrites': 0, **projection(read_state(store), {}, set())}


if __name__ == '__main__':
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument('--store', required=True)
    p.add_argument('--historical', type=Path, default=ROOT / 'work/jd-historical-baseline-20261106-r2')
    p.add_argument('--reviews', type=Path, default=ROOT / 'work/voc-g2g3-final-readback-20261107-r1/originals-package')
    p.add_argument('--generation', required=True)
    args = p.parse_args()
    try:
        print(json.dumps(run(args.store, args.historical, args.reviews, args.generation), ensure_ascii=False))
    except Exception as error:
        print(json.dumps({'ok': False, 'error': str(error)}, ensure_ascii=False))
        sys.exit(1)