kangshifu-board/scripts/verify-handoff-merge.py

"""Exercise the real handoff against a local baseline; publish only to a NEW test registry."""
import argparse
import json
from pathlib import Path
import sys

ROOT=Path(__file__).resolve().parents[1]
sys.path.insert(0,str(ROOT/'server/voc-library'))
import retrieval
import schema_catalog
import update_library as u
import voc_db as v


def verify(baseline,incoming,out):
    v.require(not out.exists() and not out.is_symlink(),'NEW_TEST_DIRECTORY_REQUIRED')
    fingerprints={str(p):u.digest(p) for p in (baseline,incoming)}
    out.mkdir(parents=True,mode=0o700)
    plan=u.merge_plan(baseline,incoming)
    registry=out/'isolated-registry'
    first=u.publish(baseline,registry,True,label='仅本地验收基线')
    actual=u.publish(incoming,registry,True,label='仅本地真实样本合并',expected_generation=first['generation'])
    current=v.registry_database(registry)
    checks={'planMatchesApply':plan['delta']==actual['delta'] and plan['plannedGeneration']==actual['generation'],
            'inputsUnchanged':all(u.digest(p)==h for p,h in fingerprints.items())}
    with v.readonly(baseline) as old,v.readonly(incoming) as inc,v.readonly(current) as new:
        checks['foreignKeysValid']=not bool(new.execute('PRAGMA foreign_key_check').fetchall())
        checks['schemaComplete']=schema_catalog.catalog(new)['complete']
        lost=[];preserved=0
        for table in ('products','reviews'):
            for row in inc.execute('SELECT * FROM '+table):
                prior=old.execute('SELECT * FROM '+table+' WHERE id=?',(row['id'],)).fetchone()
                if not prior:continue
                selected=new.execute('SELECT * FROM '+table+' WHERE id=?',(row['id'],)).fetchone()
                policy,reason=u.choose_record(table,dict(prior),dict(row))
                if policy=='kept_existing':
                    preserved+=1
                    if dict(selected)!=dict(prior):lost.append({'table':table,'id':row['id']})
                item=v.get_record(new,table,row['id'])['record']
                checks['historyReadable']=bool(item.get('mergeHistory')) and checks.get('historyReadable',True)
                for decision in item['mergeHistory']:
                    for field in ('prior_source_record_id','incoming_source_record_id','selected_source_record_id'):
                        v.get_source(new,decision[field])
        checks['retainedRowsMatchOriginalBytes']=not bool(lost)
        stats={}
        for label,c in [('before',old),('after',new)]:
            stats[label]={'counts':v.table_counts(c)}
            for quality in ('analysis','audit'):
                result=retrieval.search(c,{'entities':['reviews'],'quality':quality,'limit':1})
                stats[label][quality+'Reviews']=result['totalMatched']
            for entity in ('products','reviews'):
                r=retrieval.search(c,{'entities':[entity],'filters':{'platform':'jd','brand':'ksf','category':'banmian'},'quality':'analysis','limit':1})
                stats[label]['strictJdKsfBanmian'+entity.title()]=r['totalMatched']
        checks['existingAnalysisCoveragePreserved']=stats['before']['analysisReviews']==stats['after']['analysisReviews']
        checks['strictReviewBaselinePreserved']=stats['before']['strictJdKsfBanmianReviews']==stats['after']['strictJdKsfBanmianReviews']
        counts=[dict(r) for r in new.execute('SELECT decision,count(*) AS count FROM merge_decisions GROUP BY decision')]
        conflicts=[dict(r) for r in new.execute("SELECT entity,record_id,reason_codes_json FROM merge_decisions WHERE reason_codes_json LIKE '%IDENTITY_OR_FACT_CONFLICT%' ORDER BY entity,record_id")]
        for r in conflicts:r['reasons']=json.loads(r.pop('reason_codes_json'))
        for source in new.execute("SELECT id FROM source_records WHERE id IN (SELECT incoming_source_record_id FROM merge_decisions)"):
            citation=v.get_source(new,source['id'])['citation']
            v.require(bool(citation['archiveSha256']) and all(x['resolution']=='matched' for x in citation['upstreamSources']),'SOURCE_READBACK_FAILED')
    pointer=(registry/'current.json').read_bytes()
    replay=u.publish(incoming,registry,True)
    checks['replayDoesNotReimport']=not replay['applied'] and pointer==(registry/'current.json').read_bytes()
    checks['finalInputsUnchanged']=all(u.digest(p)==h for p,h in fingerprints.items())
    v.require(all(checks.values()),'REAL_HANDOFF_MERGE_CHECK_FAILED:'+str(checks))
    report={'ok':True,'scope':'local-preview-and-isolated-registry-only','productionImported':False,
        'networkRequests':0,'collectionCostMinor':0,'baseline':{'path':str(baseline),'sha256':fingerprints[str(baseline)]},
        'incoming':{'path':str(incoming),'sha256':fingerprints[str(incoming)]},'mergePolicy':u.MERGE_POLICY,
        'checks':checks,'statistics':stats,'preservedExistingRows':preserved,'decisions':counts,
        'conflictsForReview':conflicts,'plan':plan,'isolatedPublication':actual,'replay':replay}
    with (out/'verification.json').open('x',encoding='utf-8') as f:json.dump(report,f,ensure_ascii=False,indent=2);f.write('\n')
    return report


if __name__=='__main__':
    p=argparse.ArgumentParser(description=__doc__);p.add_argument('--baseline-db',type=Path,required=True)
    p.add_argument('--incoming',type=Path,required=True);p.add_argument('--out',type=Path,required=True);a=p.parse_args()
    report=verify(a.baseline_db.absolute(),a.incoming.absolute(),a.out.absolute())
    print(json.dumps({k:report[k] for k in ('ok','checks','statistics','decisions','mergePolicy','productionImported')},ensure_ascii=False,indent=2))