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

"""Acceptance checks against this actual, frozen Wang Yong collection."""
import argparse
import json
import statistics
import time
from pathlib import Path
import voc_db as v


def verify(db):
    before = v.sha(Path(db).read_bytes())
    with v.readonly(db) as c:
        counts = v.table_counts(c)
        assert {k:counts[k] for k in ('products','reviews','social_posts','policy_documents','policy_clauses')} == dict(products=4639,reviews=2360,social_posts=3814,policy_documents=25,policy_clauses=31)
        assert c.execute('PRAGMA integrity_check').fetchone()[0] == 'ok'
        assert not c.execute('PRAGMA foreign_key_check').fetchall()
        assert c.execute("SELECT count(*) FROM products WHERE coverage_type='recovered_review_reference_partial_catalog'").fetchone()[0] == 4
        assert c.execute("SELECT count(*) FROM reviews WHERE product_id IS NULL").fetchone()[0] == 0
        assert c.execute("SELECT count(*) FROM source_links WHERE resolution!='matched'").fetchone()[0] == 0
        assert c.execute("SELECT count(*) FROM policy_documents WHERE attachment_hash_state='matched'").fetchone()[0] == 25
        assert c.execute('SELECT sum(is_default=0),sum(is_default=1) FROM reviews').fetchone()[:] == (789,1571)
        examples, timings = [], []
        for market in ('jd','tmall'):
            for entity in ('products','reviews','prices'):
                request = dict(entity=entity,marketplace=market,brand='ksf',category='banmian',limit=20)
                result = v.query(c,request,True)
                assert result['scopeCombination']=='AND'
                assert any('idx_products_scope' in p for p in result['queryPlan'])
                if entity=='reviews':assert any('idx_reviews_product' in p for p in result['queryPlan'])
                if entity=='prices':assert any('idx_prices_product_time' in p for p in result['queryPlan'])
                cursor, seen = None, set()
                while True:
                    page = v.query(c,{**request, **({'cursor':cursor} if cursor else {})})
                    for row in page['records']:
                        assert (row['marketplace'],row['brand_code'],row['category_code']) == (market,'ksf','banmian')
                        assert row['id'] not in seen
                        assert row['citation']['archiveSha256']
                        seen.add(row['id'])
                        if entity=='reviews':
                            assert row['is_default']==0 and row['identity_state']=='matched' and row['category_state']=='consistent'
                            assert row['product_id'].startswith('jd:' if market=='jd' else 'taobao:')
                    cursor=page['nextCursor']
                    if cursor is None:break
                assert len(seen)==result['totalMatched']
                for _ in range(8):
                    started=time.perf_counter();v.query(c,request);timings.append((time.perf_counter()-started)*1000)
                examples.append({'filters':request,'totalMatched':result['totalMatched'],'queryPlan':result['queryPlan'],'paginationComplete':True})
        # All registered brands remain isolated at the SQL layer.
        for market in ('jd','tmall'):
            for brand in c.execute('SELECT code FROM brands'):
                result=v.query(c,dict(entity='products',marketplace=market,brand=brand[0],category='banmian',limit=100))
                assert all(r['marketplace']==market and r['brand_code']==brand[0] and r['category_code']=='banmian' for r in result['records'])
        stats={
            'classification':dict(c.execute('SELECT classification_state,count(*) FROM products GROUP BY 1')),
            'marketplaces':[dict(r) for r in c.execute('SELECT marketplace,count(*) AS count FROM products GROUP BY 1')],
            'reviews':[dict(r) for r in c.execute('SELECT marketplace_claim,identity_state,category_state,is_default,count(*) AS count FROM reviews GROUP BY 1,2,3,4')],
            'quality':dict(c.execute('SELECT code,count(*) FROM quality_issues GROUP BY code')),
            'validNumericPrices':c.execute('SELECT count(*) FROM prices WHERE amount_minor IS NOT NULL').fetchone()[0],
            'verifiedUnitPrices':c.execute("SELECT count(*) FROM prices WHERE unit_price_verification='verified'").fetchone()[0],
            'titleSpecCandidates':c.execute("SELECT count(*) FROM specs WHERE verification='title_candidate'").fetchone()[0],
            'socialWithoutFulltext':c.execute('SELECT count(*) FROM social_posts WHERE fulltext_available=0').fetchone()[0]
        }
        generation=c.execute("SELECT value FROM meta WHERE key='generation'").fetchone()[0]
    assert before==v.sha(Path(db).read_bytes())
    return dict(ok=True,schema=v.VERSION,generation=generation,databaseSha256=before,counts=counts,statistics=stats,examples=examples,
                queryBenchmark=dict(samples=len(timings),medianMs=round(statistics.median(timings),3),maxMs=round(max(timings),3),scope='本机已有连接，查询20条并附来源；不含生产网络/Agent启动耗时'),
                checks=['integrity','foreign_keys','primary_counts','4_recovered_references','no_orphan_reviews','all_source_paths_matched','25_policy_hashes_with_explicit_basis','default_review_counts','exact_scope','review_sku_identity','all_brand_isolation','complete_keyset_pagination','compound_indexes','readonly_hash_unchanged'],
                productionDatabaseChanged=False,agentInstalled=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({k:val for k,val in result.items() if k not in ('examples','statistics')},ensure_ascii=False,indent=2))