kangshifu-board/server/voc-library/verify_retrieval.py

"""Acceptance for distinct research intents using the actual Wang Yong corpus."""
import argparse
import json
from pathlib import Path
import statistics
import time
import voc_db as v
import retrieval as r


CASES = [
 ('某品牌的全部商品',dict(entities=['products'],filters={'brand':'康师傅'},limit=20)),
 ('跨品牌的拌面',dict(entities=['products'],filters={'category':'拌面'},limit=20)),
 ('京东某品牌拌面',dict(entities=['products'],filters={'platform':'京东','brand':'康师傅','category':'拌面'},limit=20)),
 ('评价中的份量主题',dict(entities=['reviews'],text={'concepts':['portion_size']},limit=20)),
 ('品牌份量相关的评价和社媒',dict(entities=['reviews','social'],filters={'brand':'康师傅'},text={'concepts':['portion_size']},limit=20)),
 ('按商品关联规格、价格和评价',dict(entities=['specs','prices','reviews'],filters={'productId':'jd:100005702440'},limit=20)),
 ('价格区间与口径',dict(entities=['prices'],filters={'priceMinor':{'min':2000,'max':4000},'priceBasis':'search_listing_snapshot'},limit=20)),
 ('有标题规格候选的记录',dict(entities=['specs'],filters={'specVerification':'title_candidate','netWeightG':{'min':100,'max':150}},limit=20)),
 ('营养标签政策全文',dict(entities=['documents'],text={'any':['营养标签','营养成分表']},limit=20)),
 ('政策条款与原文',dict(entities=['clauses'],filters={'documentId':'GB17400-2015'},limit=20)),
 ('榜单标识与范围',dict(entities=['rankings'],filters={'rankNumber':{'max':10}},limit=20)),
 ('采集覆盖',dict(entities=['coverage'],filters={'brand':'康师傅'},limit=20)),
 ('品牌未知资料审计',dict(entities=['products'],filters={'brand':{'exists':False}},limit=20)),
]


def verify(db):
    before=v.sha(Path(db).read_bytes());examples=[];timings=[]
    with v.readonly(db) as c:
        assert c.execute('PRAGMA integrity_check').fetchone()[0]=='ok'
        assert not c.execute('PRAGMA foreign_key_check').fetchall()
        assert c.execute('SELECT count(*) FROM retrieval_documents').fetchone()[0]==21108
        assert c.execute('SELECT count(*) FROM policy_documents WHERE length(full_text)>0').fetchone()[0]==25
        for title,q in CASES:
            result=r.search(c,q,True)
            assert result['totalMatched']>0,(title,result)
            for row in result['records']:
                assert row['citation']['archiveSha256']
                if row['entity']=='reviews':assert row['is_default']==0 and row['identity_state']=='matched' and row['category_state']!='conflict'
                if 'platform' in q.get('filters',{}):assert row['platform']=='jd'
                if 'category' in q.get('filters',{}):assert row['category']=='banmian'
                if row['entity']!='social' and 'brand' in q.get('filters',{}) and isinstance(q['filters']['brand'],str):assert row['brand']=='ksf'
                # Detail for every returned entity type must resolve exactly.
                detail=v.get_record(c,row['entity'],row['record_id'])
                assert detail['record']['source_record_id']==row['source_record_id']
            if q.get('text'):
                assert any('VIRTUAL TABLE INDEX' in step for step in result['queryPlan'])
                assert all(row['match']['matchedTerms'] for row in result['records'])
            for _ in range(4):
                start=time.perf_counter();r.search(c,q);timings.append((time.perf_counter()-start)*1000)
            examples.append(dict(question=title,request=q,totalMatched=result['totalMatched'],countsByEntity=result['countsByEntity'],
                distinctTextFingerprintsByEntity=result['distinctTextFingerprintsByEntity'],queryPlan=result['queryPlan'],firstRefs=[dict(entity=x['entity'],id=x['record_id'],sourceId=x['source_record_id']) for x in result['records'][:2]]))
        # Brand-only must not discard a product merely because category is unknown.
        brands=r.search(c,dict(entities=['products'],filters={'brand':'ksf'}))
        expected=c.execute("SELECT count(*) FROM products WHERE brand_code='ksf'").fetchone()[0]
        assert brands['totalMatched']==expected
        assert c.execute("SELECT count(*) FROM products WHERE brand_code='ksf' AND category_code IS NULL").fetchone()[0]>0
        # Entire matching feedback scope pages once; no repeated IDs or silent scope change.
        q=dict(entities=['reviews'],text={'concepts':['portion_size']},limit=15);seen=set();cursor=None
        while True:
            page=r.search(c,{**q,**({'cursor':cursor} if cursor else {})})
            for x in page['records']:
                assert x['record_id'] not in seen;seen.add(x['record_id'])
            cursor=page['nextCursor']
            if not cursor:break
        assert len(seen)==page['totalMatched']
        groups=r.aggregate(c,dict(entities=['reviews'],filters={'brand':'ksf'},groupBy=['platform','category']))
        assert all(x['recordCount']>=x['distinctTextCount'] and x['recordCount']>=x['linkedProductCount'] for x in groups['groups']['platform,category']['rows'])
        facets=r.aggregate(c,dict(entities=['products'],filters={'brand':'ksf'},groupBy=['platform','category']),True)
        generation=c.execute("SELECT value FROM meta WHERE key='generation'").fetchone()[0]
        counts=v.table_counts(c)
    assert v.sha(Path(db).read_bytes())==before
    return dict(ok=True,schema=v.VERSION,generation=generation,databaseSha256=before,counts=counts,scenarios=examples,
        benchmark=dict(samples=len(timings),medianMs=round(statistics.median(timings),3),maxMs=round(max(timings),3),scope='本地SQLite已有连接；含来源/全文匹配统计，不含Agent或网络耗时'),
        assertions=['13种真实研究查询','按实际条件而非强制三维','未知未限定字段不导致漏掉相关记录','中文正文全文索引','有限同义词可见','跨类型不混计','关联回读','25份政策全文','条款/应用提示区分','组合索引','范围绑定分页','统计口径分开','查询前后数据库SHA不变'],
        productionInstalled=False,realMessageVerified=False,publicBoardConnected=False)


if __name__=='__main__':
    p=argparse.ArgumentParser();p.add_argument('--db',required=True);p.add_argument('--out',required=True);a=p.parse_args()
    result=verify(a.db);Path(a.out).write_text(json.dumps(result,ensure_ascii=False,indent=2)+'\n',encoding='utf-8')
    print(json.dumps(dict(ok=result['ok'],scenarios=len(result['scenarios']),benchmark=result['benchmark'],generation=result['generation']),ensure_ascii=False,indent=2))