"""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)