kangshifu-board/scripts/verify-industry-recovered-data.py

"""Exercise recovered real evidence through frozen read APIs, without network or dispatch."""
import argparse
from contextlib import closing
import hashlib
import json
from pathlib import Path
import sqlite3
import sys

ROOT=Path(__file__).resolve().parents[1]
sys.path.insert(0,str(ROOT/'server/voc-library'))
from industry_monitor import run
from monitoring_dispatch import batch_preflight
from monitoring_ledger import read


def main():
    p=argparse.ArgumentParser(description=__doc__);p.add_argument('--snapshot',type=Path,required=True)
    p.add_argument('--out',type=Path,required=True);a=p.parse_args()
    store=ROOT/'data/monitoring-registry';snapshot=a.snapshot.resolve()
    sha=lambda path:hashlib.sha256(path.read_bytes()).hexdigest()
    before=sha(snapshot);database_before=sha(store/'monitoring.sqlite')
    generation=json.loads(snapshot.read_bytes())['generation']
    config={'snapshot':str(snapshot),'monitoringStore':str(store)}
    with closing(sqlite3.connect((store/'monitoring.sqlite').as_uri()+'?mode=ro',uri=True)) as c:
        c.row_factory=sqlite3.Row;_,rows=read(c)
    bound=[r for r in rows if r['params']]
    assert len(rows)==154 and len(bound)==139
    requests=[]
    for row in bound:
        modes=['general','feedbackdate'] if row['endpointId'].startswith('taobao.') else [None]
        for mode in modes:
            params=dict(row['params'])
            if mode:params['orderType']=mode
            requests.append({'endpointId':row['endpointId'],'params':params,'generation':generation,'window':'independent-recovery-audit'})
    requests.append(requests[0])
    audited=batch_preflight(store,requests,generation)
    assert len(audited['rows'])==143
    assert all(not r['dispatchAllowed'] and 'REQUEST_ALREADY_CHARGED_NO_AUTOMATIC_REPLAY' in r['blockingReasons'] for r in audited['rows'])
    assert all(len(r['alreadyCharged'])==sum(x['endpointId']==r['specification']['endpointId'] and x['params']=={k:v for k,v in r['specification']['params'].items() if k!='orderType'} for x in bound) for r in audited['rows'])
    assert 'DUPLICATE_REQUEST_IN_BATCH' in audited['rows'][-1]['blockingReasons']
    result=run(config,{'op':'listings','generation':generation,'filters':{'platform':'jd','shopId':'1000084104'},
                       'detailQuery':{'state':'TEXT_AVAILABLE','keyword':'久煮不烂'}})
    detail=result['detailEvidence']
    assert detail['coverage']['withResponse']==41 and detail['coverage']['withDetailText']==12
    found=[r for r in detail['rows'] if r['productId']=='jd:100005232115']
    assert len(found)==1 and any(f['text']=='麻辣爽口|久煮不烂|醇厚口感' for f in found[0]['fields'])
    assert not found[0]['rawAvailableLocally'] and found[0]['projectionVerified']
    assert before==sha(snapshot) and database_before==sha(store/'monitoring.sqlite')
    scope=json.loads((ROOT/'handoff/collector-scope.json').read_bytes())
    assert scope['chargedHistory']['identifiedRows']==154 and scope['chargedHistory']['parameterBoundRows']==139
    assert sum(len(r['settledRequests']) for r in scope['primaryProducts'])>76
    receipt={'ok':True,'generation':generation,'snapshotSha256':before,'identifiedRows':154,'parameterBoundRows':139,
             'unboundRows':15,'preflightCases':143,'allPaidPagesBlocked':True,'taobaoBothModesBlocked':True,
             'duplicateInBatchRejected':True,'primaryDetailsWithResponse':41,'primaryDetailsWithText':12,
             'recoveredLiteralTextVerified':True,'snapshotUnchanged':True,'auxiliaryUnchanged':True,
             'chargedMinor':4620,'paidCalls':0,'productionWrites':0}
    a.out.parent.mkdir(parents=True,exist_ok=True);a.out.write_text(json.dumps(receipt,ensure_ascii=False,indent=2)+'\n','utf-8')
    print(json.dumps(receipt,ensure_ascii=False))


if __name__=='__main__':main()