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

"""Read-only, bounded browser projection using the same retrieval rules as Agents."""
import sys
sys.dont_write_bytecode = True
import json
from pathlib import Path
import voc_db
import retrieval
from product_insights import overview as product_overview

LABELS = {'products':'商品与竞品','reviews':'电商评价','social':'社媒内容','documents':'政策文件',
          'clauses':'条款与应用','prices':'价格快照','specs':'商品规格','rankings':'榜单线索','coverage':'采集覆盖'}

def active_database(config):
    if config.get('registry'):
        root = Path(config['registry']).resolve()
        pointer = json.loads((root/'current.json').read_text(encoding='utf-8'))
        p = (root/pointer['file']).resolve()
        voc_db.require(p.is_relative_to(root) and p.is_file(), 'INVALID_LIBRARY_POINTER')
        return p, pointer
    return Path(config['db']).resolve(), None

def query_request(p):
    entity = p.get('entity','products')
    voc_db.require(entity in LABELS, 'UNKNOWN_ENTITY')
    filters = p.get('filters', {})
    voc_db.require(isinstance(filters,dict), 'INVALID_FILTERS')
    q = {'entities':[entity], 'filters':filters, 'quality':p.get('quality','analysis'), 'limit':p.get('limit',20)}
    if p.get('keyword'): q['text'] = {'all':[p['keyword']]}
    if p.get('cursor'): q['cursor'] = p['cursor']
    if p.get('sort'): q['sort'] = p['sort']
    return q

def summary(c, pointer):
    meta = dict(c.execute('SELECT * FROM meta'))
    counts = voc_db.table_counts(c)
    totals = {}
    for entity in LABELS:
        where,args,_ = retrieval.prepare(c, {'entities':[entity]})
        totals[entity] = c.execute('SELECT count(*) FROM retrieval_documents d WHERE '+where,args).fetchone()[0]
    quality = {
        'defaultReviews':c.execute('SELECT count(*) FROM reviews WHERE is_default=1').fetchone()[0],
        'pendingProducts':c.execute("SELECT count(*) FROM products WHERE classification_state='pending'").fetchone()[0],
        'verifiedSpecs':c.execute("SELECT count(*) FROM specs WHERE verification='source_attribute'").fetchone()[0],
        'datedPrices':c.execute('SELECT count(*) FROM prices WHERE source_calendar_date IS NOT NULL').fetchone()[0],
        'socialFulltext':c.execute('SELECT count(*) FROM social_posts WHERE fulltext_available=1').fetchone()[0],
        'documentFulltext':c.execute("SELECT count(*) FROM policy_documents WHERE length(trim(coalesce(full_text,'')))>0").fetchone()[0],
        'brands':c.execute('SELECT count(DISTINCT brand_code) FROM products').fetchone()[0],
        'shops':c.execute("SELECT count(DISTINCT marketplace||':'||shop_id) FROM products WHERE shop_id IS NOT NULL").fetchone()[0],
    }
    updates=[]
    for item in (pointer.get('history',[]) if pointer else []):
        entry=dict(item);delta=entry.get('delta',{})
        if delta:
            entry['summary']='、'.join(LABELS[e]+'新增'+str(delta.get('added',{}).get(t,0))+'／变更'+str(delta.get('changed',{}).get(t,0)) for e,t in retrieval.ENTITIES.items() if delta.get('added',{}).get(t,0) or delta.get('changed',{}).get(t,0)) or '重复资料，无实体新增'
        updates.append(entry)
    return {'schema':meta['schema'],'generation':meta['generation'], 'counts':counts,'analysisCounts':totals,'quality':quality,
            'taxonomy':json.loads(meta['taxonomy']), 'entities':[{'id':k,'name':v,'count':counts[retrieval.ENTITIES[k]],'analysisCount':totals[k]} for k,v in LABELS.items()],
            'batches':[dict(r) for r in c.execute('SELECT * FROM batches ORDER BY imported_at DESC')],
            'updates':updates,'updatedAt':pointer.get('updatedAt') if pointer else None}

def browse(c,p):
    q = query_request(p)
    result = retrieval.search(c,q)
    # Display prose only, never raw upstream payloads, author IDs, or private filesystem paths.
    for row in result['records']:
        row.pop('citation',None)
        if row['entity']=='reviews':
            raw=c.execute('SELECT content,purchased_sku_text,rating_raw,review_date_raw FROM reviews WHERE id=?',(row['record_id'],)).fetchone()
            row.update(dict(raw));row['preview']=raw['content'][:230]
            body=c.execute("SELECT body FROM retrieval_documents WHERE entity='reviews' AND record_id=?",(row['record_id'],)).fetchone()[0]
            row['append_content']=body[len(raw['content']):].strip() if body.startswith(raw['content']) else ''
        elif row['entity']=='products':
            raw=c.execute('SELECT shop_name FROM products WHERE id=?',(row['record_id'],)).fetchone()
            row['shop_name']=raw['shop_name']
            row['preview']='店铺：'+raw['shop_name'] if raw['shop_name'] else ''
        elif row['entity']=='social' and not row['fulltext_available']:row['preview']=''
        else: row['preview']=row['match']['text'][:230]
        if row['entity'] in ('specs','prices','rankings','coverage'):
            fields={'specs':'spec_text,net_weight_g,pack_count,total_weight_g',
                    'prices':'amount_minor,raw_amount,price_basis,price_per_100g,unit_price_verification,source_time_raw',
                    'rankings':'title,rank_number,scope,source_time_raw',
                    'coverage':'status,available_reviews,reported_total,historical_total_complete'}[row['entity']]
            raw=c.execute('SELECT '+fields+' FROM '+retrieval.ENTITIES[row['entity']]+' WHERE id=?',(row['record_id'],)).fetchone()
            row.update(dict(raw))
    if p.get('entity','products')=='products' and result['records']:
        # One grouped read per page, scoped by exact product IDs; titles never join records.
        ids=[row['record_id'] for row in result['records']]
        counts={pid:{} for pid in ids}
        for row in c.execute('SELECT product_id,entity,count(*) AS n FROM retrieval_documents WHERE product_id IN ('+','.join('?' for _ in ids)+') GROUP BY product_id,entity',ids):
            counts[row['product_id']][row['entity']]=row['n']
        prices={row['product_id']:{'minMinor':row['low'],'maxMinor':row['high'],'recordCount':row['n']}
                for row in c.execute('SELECT product_id,min(amount_minor) low,max(amount_minor) high,count(*) n FROM prices WHERE amount_minor IS NOT NULL AND product_id IN ('+','.join('?' for _ in ids)+') GROUP BY product_id',ids)}
        for row in result['records']:
            row['relatedCounts']=counts[row['record_id']]
            row['priceRange']=prices.get(row['record_id'])
    fields=[f for f in ('platform','brand','category','shopId','policyStatus','contentScope','specVerification','priceBasis') if f in retrieval.ENTITY_FIELDS[p.get('entity','products')]]
    facets={}
    for field in fields:
        fq={k:v for k,v in q.items() if k not in ('cursor','limit')}
        fq['filters']={k:v for k,v in q['filters'].items() if k!=field}
        # Social brands are mentions, not product ownership; enumerate distinct mentions explicitly.
        if field=='brand' and p.get('entity')=='social':
            where,args,_=retrieval.prepare(c,fq)
            rows=[dict(r) for r in c.execute('SELECT m.brand_code AS brand,count(DISTINCT d.record_id) AS recordCount FROM retrieval_documents d JOIN social_brand_mentions m ON m.post_id=d.record_id WHERE '+where+' GROUP BY m.brand_code ORDER BY recordCount DESC LIMIT 101',args)]
            facets[field]={'rows':rows[:100],'truncated':len(rows)>100}
        else:
            fq.update(groupBy=[field],limit=100)
            facets[field]=retrieval.aggregate(c,fq,facets=True)['groups'][field]
        if field=='shopId':
            for row in facets[field]['rows']:
                names=[r[0] for r in c.execute('SELECT DISTINCT shop_name FROM products WHERE shop_id=? AND shop_name IS NOT NULL',(row.get('shop_id'),))]
                row['shop_name']=' / '.join(names)
    result['facets']=facets
    where,args,_=retrieval.prepare(c,q)
    result['coverage']={}
    for field in ('sourceDate','reviewDate','publishedDate'):
        if field in retrieval.ENTITY_FIELDS[p.get('entity','products')]:
            col=retrieval.FIELDS[field][0]
            r=c.execute('SELECT min('+col+'),max('+col+'),sum('+col+' IS NULL) FROM retrieval_documents d WHERE '+where,args).fetchone()
            result['coverage'][field]={'min':r[0],'max':r[1],'unknown':r[2] or 0}
    result['pendingCount']=None
    if q['quality']=='analysis':
        aq={**q,'quality':'audit'}
        aw,av,_=retrieval.prepare(c,aq)
        result['pendingCount']=c.execute('SELECT count(*) FROM retrieval_documents d WHERE '+aw,av).fetchone()[0]-result['totalMatched']
    return result

def detail(c,p):
    entity=p.get('entity');rid=p.get('id')
    voc_db.require(entity in LABELS and isinstance(rid,str) and len(rid)<=300,'INVALID_RECORD')
    record=voc_db.get_record(c,entity,rid)['record']
    if entity=='reviews':
        body=c.execute("SELECT body FROM retrieval_documents WHERE entity='reviews' AND record_id=?",(rid,)).fetchone()[0]
        record['append_content_json']=body[len(record['content']):].strip() if body.startswith(record['content']) else ''
    citation=record.pop('citation')
    record.pop('productCitation',None)
    for row in record.get('classification',[]):row.pop('evidence_record_id',None)
    # Keep source locator and hashes; omit private upstream paths and raw JSON.
    record['source']={k:citation[k] for k in ('id','kind','asset_path','line_number','payload_sha256','assetSha256','archiveSha256','archiveSha256s','locatorType') if k in citation}
    if 'asset_path' in record['source']: record['source']['asset_path']=record['source']['asset_path'].replace('\\','/').split('/')[-1]
    issues=[dict(r) for r in c.execute('SELECT code,detail FROM quality_issues WHERE entity_id=?',(rid,))]
    # Issue detail can contain original private paths. Show the rule code only.
    record['issues']=[r['code'] for r in issues]
    relations=[]
    if entity=='products':
        record['overview'] = product_overview(c, record)
        for related in ('reviews','prices','specs','rankings','coverage'):
            q={'entities':[related],'filters':{'productId':rid},'quality':'audit'}
            where,args,_=retrieval.prepare(c,q)
            n=c.execute('SELECT count(*) FROM retrieval_documents d WHERE '+where,args).fetchone()[0]
            relations.append({'entity':related,'productId':rid,'count':n,'name':LABELS[related]})
    if entity=='documents':
        clauses=[{'id':r['id'],'name':r['clause']} for r in c.execute('SELECT id,clause FROM policy_clauses WHERE document_id=? ORDER BY id',(rid,))]
        record['clauses']=clauses
    if record.get('product_id'):
        parent=c.execute('SELECT id,title,url,marketplace,shop_name,brand_code,category_code FROM products WHERE id=?',(record['product_id'],)).fetchone()
        if parent:record['parentProduct']=dict(parent)
    if entity=='clauses':
        parent=c.execute('SELECT id,title,source_url FROM policy_documents WHERE id=?',(record['document_id'],)).fetchone()
        if parent:record['parentDocument']=dict(parent)
    return {'record':record,'relations':relations}

def run(config,p):
    voc_db.require(isinstance(p,dict) and p.get('op') in ('summary','browse','detail','version'),'READ_COMMAND_REQUIRED')
    voc_db.require(not set(p)-{'op','entity','filters','quality','limit','keyword','cursor','id','generation','sort'},'UNKNOWN_REQUEST_FIELD')
    db,pointer=active_database(config)
    with voc_db.readonly(db) as c:
        generation=c.execute("SELECT value FROM meta WHERE key='generation'").fetchone()[0]
        voc_db.require(not pointer or pointer['generation']==generation,'POINTER_VERSION_MISMATCH')
        voc_db.require(not p.get('generation') or p['generation']==generation,'LIBRARY_VERSION_CHANGED')
        if p['op']=='summary':data=summary(c,pointer)
        elif p['op']=='version':data={'updatedAt':pointer.get('updatedAt') if pointer else None}
        elif p['op']=='browse':data=browse(c,p)
        else:data=detail(c,p)
        return {'ok':True,'generation':generation,'readOnly':True,'databaseWrites':0,**data}

if __name__=='__main__':
    try:
        payload=json.loads(sys.stdin.read())
        print(json.dumps(run(payload['config'],payload['request']),ensure_ascii=False))
    except Exception as e:
        code=str(e) if isinstance(e,voc_db.LibraryError) else 'LIBRARY_READ_FAILED'
        print(json.dumps({'ok':False,'error':{'code':code}},ensure_ascii=False));sys.exit(1)