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

"""Typed retrieval across VOC entities; literal evidence + bounded term expansion.

No natural-language/semantic understanding is claimed here: the Agent translates
the question using the discoverable contract. FTS generates indexed candidates;
literal phrases are checked again and returned with original-text snippets.
"""
import base64
import collections
import hashlib
import json
import math
import re
import unicodedata

VERSION = 'ksf.voc-retrieval.v2'
AGENT_RULES_VERSION = 'ksf.agent-data-rules.20261106.v1'
ENTITIES = {'products':'products', 'reviews':'reviews', 'prices':'prices', 'specs':'specs',
            'social':'social_posts', 'documents':'policy_documents', 'clauses':'policy_clauses',
            'rankings':'ranking_signals', 'coverage':'coverage_observations'}
DEFAULT_ENTITIES = ['products','reviews','social','documents','clauses']
CONCEPTS = {
    'portion_size': {'name':'份量', 'terms':['份量','分量','面量','量少','不够吃','吃不饱','量太少']},
    'preparation': {'name':'制作便利', 'terms':['煮面','冲泡','不能泡','必须煮','沥水','免煮','煮不熟','煮了']},
    'taste': {'name':'口味体验', 'terms':['口味','味道','太辣','太咸','太淡','好吃','难吃','口感']},
    'packaging': {'name':'包装体验', 'terms':['包装','漏油','破损','破包','封口','餐具']},
    'value': {'name':'价格与价值', 'terms':['价格','性价比','太贵','便宜','划算','涨价']},
    'nutrition_label': {'name':'营养与标签', 'terms':['营养','标签','配料','钠含量','脂肪','能量','蛋白质']},
}
FIELDS = {
    'platform':('platform','enum'), 'brand':('brand','enum'), 'category':('category','enum'),
    'shopId':('shop_id','id'), 'productId':('product_id','id'), 'itemId':('item_id','id'), 'recordId':('record_id','id'),
    'sourceId':('source_record_id','id'), 'sourceDate':('source_day','date'), 'reviewDate':('review_day','date'),
    'publishedDate':('published_day','date'), 'priceMinor':('price_minor','number'), 'priceBasis':('price_basis','text'),
    'netWeightG':('net_weight_g','number'), 'packCount':('pack_count','number'), 'specVerification':('spec_verification','text'),
    'policyStatus':('policy_status','text'), 'documentId':('document_id','id'), 'contentScope':('content_scope','text'),
    'fulltextAvailable':('fulltext_available','bool'), 'isDefault':('is_default','bool'), 'identityState':('identity_state','text'),
    'ratingValue':('rating_value','number'), 'ratingScale':('rating_scale','text'),
    'categoryState':('category_state','text'), 'classificationState':('classification_state','text'),
    'rankNumber':('rank_number','number'), 'rankScope':('rank_scope','text'),
}
ENTITY_FIELDS = {
    'products':{'platform','brand','category','shopId','productId','itemId','sourceDate','classificationState'},
    'reviews':{'platform','brand','category','shopId','productId','itemId','sourceDate','reviewDate','isDefault','identityState','categoryState','classificationState','ratingValue','ratingScale'},
    'prices':{'platform','brand','category','shopId','productId','itemId','sourceDate','priceMinor','priceBasis','classificationState'},
    'specs':{'platform','brand','category','shopId','productId','itemId','sourceDate','netWeightG','packCount','specVerification','classificationState'},
    'social':{'platform','brand','publishedDate','contentScope','fulltextAvailable'},
    'documents':{'sourceDate','documentId','policyStatus','fulltextAvailable'},
    'clauses':{'sourceDate','documentId','policyStatus'},
    'rankings':{'platform','brand','category','shopId','productId','itemId','sourceDate','rankNumber','rankScope','classificationState'},
    'coverage':{'platform','brand','category','shopId','productId','itemId','classificationState'},
}
COMMON_FIELDS = {'recordId','sourceId'}
PROJECTION_COLUMNS = ['entity','record_id','title','body','normalized_text','tokens','content_fingerprint','platform','brand','category','shop_id','product_id','item_id','source_record_id',
                      'source_day','review_day','published_day','price_minor','price_basis','net_weight_g','pack_count','spec_verification','policy_status','document_id',
                      'content_scope','fulltext_available','is_default','identity_state','category_state','classification_state','rank_number','rank_scope','rating_value','rating_scale']


def norm(text):
    return unicodedata.normalize('NFKC', str(text or '')).casefold()


def tokens(text):
    out=set()
    for word in re.findall(r'[\u3400-\u9fff]+|[^\W_]+', norm(text), re.UNICODE):
        if re.fullmatch(r'[\u3400-\u9fff]+',word):
            out.update(word)
            out.update(word[i:i+2] for i in range(len(word)-1))
        else:
            out.add(word)
    return sorted(out)


def canonical(value):
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':'), allow_nan=False)


def require(value, code):
    if not value:
        from voc_db import LibraryError
        raise LibraryError(code)


def create_index(c):
    c.executescript('''
    CREATE TABLE retrieval_documents(
      doc_id INTEGER PRIMARY KEY, entity TEXT NOT NULL, record_id TEXT NOT NULL, title TEXT NOT NULL, body TEXT NOT NULL,
      normalized_text TEXT NOT NULL, tokens TEXT NOT NULL, content_fingerprint TEXT NOT NULL, platform TEXT, brand TEXT, category TEXT, shop_id TEXT, product_id TEXT, item_id TEXT,
      source_record_id TEXT NOT NULL REFERENCES source_records(id), source_day TEXT, review_day TEXT, published_day TEXT,
      price_minor INTEGER, price_basis TEXT, net_weight_g REAL, pack_count INTEGER, spec_verification TEXT, policy_status TEXT, document_id TEXT,
      content_scope TEXT, fulltext_available INTEGER, is_default INTEGER, identity_state TEXT, category_state TEXT, classification_state TEXT,
      rank_number INTEGER, rank_scope TEXT, rating_value REAL, rating_scale TEXT, UNIQUE(entity,record_id));
    CREATE INDEX idx_retrieval_scope ON retrieval_documents(entity,platform,brand,category,doc_id);
    CREATE INDEX idx_retrieval_brand ON retrieval_documents(brand,entity,doc_id);
    CREATE INDEX idx_retrieval_category ON retrieval_documents(category,entity,doc_id);
    CREATE INDEX idx_retrieval_product ON retrieval_documents(product_id,entity,doc_id);
    CREATE INDEX idx_retrieval_item ON retrieval_documents(item_id,entity,doc_id);
    CREATE INDEX idx_retrieval_shop ON retrieval_documents(shop_id,entity,doc_id);
    CREATE INDEX idx_retrieval_document ON retrieval_documents(document_id,entity,doc_id);
    CREATE INDEX idx_retrieval_policy ON retrieval_documents(policy_status,entity,doc_id);
    CREATE INDEX idx_retrieval_content ON retrieval_documents(entity,content_fingerprint);
    CREATE INDEX idx_retrieval_review_date ON retrieval_documents(entity,review_day,doc_id);
    CREATE INDEX idx_retrieval_source_date ON retrieval_documents(entity,source_day,doc_id);
    CREATE INDEX idx_retrieval_published_date ON retrieval_documents(entity,published_day,doc_id);
    CREATE VIRTUAL TABLE retrieval_fts USING fts5(tokens,content='retrieval_documents',content_rowid='doc_id',tokenize='unicode61');
    ''')
    c.row_factory = __import__('sqlite3').Row
    products = {r['id']:dict(r) for r in c.execute('SELECT * FROM products')}
    docs = {r['id']:dict(r) for r in c.execute('SELECT * FROM policy_documents')}
    from voc_db import calendar_date

    def add(entity, row, parent=None):
        p=parent or {}
        data={col:None for col in PROJECTION_COLUMNS}
        data.update(entity=entity,record_id=row['id'],title='',body='',source_record_id=row['source_record_id'])
        if entity=='products':p=row
        if entity in ('products','reviews','prices','specs','rankings','coverage'):
            data.update(platform=p.get('marketplace'),brand=p.get('brand_code'),category=p.get('category_code'),shop_id=p.get('shop_id'),product_id=p.get('id'),item_id=p.get('item_id'),classification_state=p.get('classification_state'),title=p.get('title',''))
        if entity=='products':
            data['body']=row['title']
        elif entity=='reviews':
            # Initial review and append are separate; only decoded textual append
            # values are indexed, never usernames/ratings/whole JSON as prose.
            appendix=json.loads(row.get('append_content_json') or 'null')
            parts=[]
            if isinstance(appendix,str):parts.append(appendix)
            elif isinstance(appendix,dict):parts.extend(str(appendix[k]) for k in ('content','comment','appendComment') if isinstance(appendix.get(k),str))
            elif isinstance(appendix,list):
                for part in appendix:
                    if isinstance(part,dict):parts.extend(part[k] for k in ('content','comment') if isinstance(part.get(k),str))
            data.update(body=row['content']+'\n'+'\n'.join(parts),review_day=row.get('review_calendar_date'),is_default=row.get('is_default'),identity_state=row['identity_state'],category_state=row['category_state'])
            raw_rating=str(row.get('rating_raw') or '')
            data.update(rating_value=float(raw_rating) if re.fullmatch(r'-?\d+(?:\.\d+)?',raw_rating) else None,rating_scale=row.get('rating_scale'))
        elif entity=='prices':data.update(body=row['raw_amount']+' '+row['price_basis'],price_minor=row['amount_minor'],price_basis=row['price_basis'])
        elif entity=='specs':data.update(body=row.get('spec_text') or '',net_weight_g=row['net_weight_g'],pack_count=row['pack_count'],spec_verification=row['verification'])
        elif entity=='social':
            data.update(platform=row['platform'],title=row.get('title') or '',body=row.get('content') or '',content_scope=row['content_scope'],published_day=calendar_date(row['published_at_raw']),fulltext_available=row['fulltext_available'])
        elif entity=='documents':
            data.update(title=row['title'],body=row.get('full_text') or '',document_id=row['id'],policy_status=row['declared_status'],fulltext_available=int(bool(row.get('full_text'))))
        elif entity=='clauses':
            data.update(title=(p.get('title') or '')+' '+row['clause'],body=row['excerpt']+'\n'+str(row.get('application_note') or ''),document_id=row['document_id'],policy_status=p.get('declared_status'))
        elif entity=='rankings':data.update(body=str(row.get('title') or ''),rank_number=row.get('rank_number'),rank_scope=row['scope'])
        elif entity=='coverage':data.update(body=str(row.get('status') or ''))
        data['source_day']=calendar_date(row.get('source_time_raw'))
        if entity=='reviews':
            # Avoid a product title making a review falsely match a pain point.
            searchable=data['body']
        else:
            searchable=data['title']+'\n'+data['body']
        data['normalized_text']=norm(searchable)
        data['content_fingerprint']=hashlib.sha256(data['normalized_text'].encode('utf-8')).hexdigest()
        data['tokens']=' '.join(tokens(searchable))
        c.execute('INSERT INTO retrieval_documents('+','.join(PROJECTION_COLUMNS)+') VALUES('+','.join('?' for _ in PROJECTION_COLUMNS)+')',[data[k] for k in PROJECTION_COLUMNS])

    for entity,table in ENTITIES.items():
        for row in c.execute('SELECT * FROM '+table+' ORDER BY id').fetchall():
            row=dict(row)
            parent=products.get(row.get('product_id')) if entity not in ('products','documents','clauses','social') else docs.get(row.get('document_id')) if entity=='clauses' else None
            add(entity,row,parent)
    c.execute("INSERT INTO retrieval_fts(retrieval_fts) VALUES('rebuild')")
    c.execute("INSERT INTO retrieval_fts(retrieval_fts) VALUES('optimize')")


def catalog(c):
    tax=json.loads(c.execute("SELECT value FROM meta WHERE key='taxonomy'").fetchone()[0])
    return {'ok':True,'retrievalContract':VERSION,'agentRulesVersion':AGENT_RULES_VERSION,'readOnly':True,'databaseWrites':0,
      'scopeContract':{'filterCombination':'AND','explicitDimensionsMustMatch':True,'unrequestedDimensionsMayBeUnknown':True,
                       'automaticCategoryExpansion':False,'reviewJoin':'platform_family + item_id → product_id',
                       'reviewAnalysis':'matched identity, non-default, no SKU conflict; explicit category additionally requires consistent purchased SKU'},
      'entities':{e:{'count':c.execute('SELECT count(*) FROM '+t).fetchone()[0],'fields':sorted(ENTITY_FIELDS[e]|COMMON_FIELDS)} for e,t in ENTITIES.items()},
      'fields':{k:{'type':typ,'operators':['eq','in','exists']+(['min','max'] if typ in ('number','date') else []),'unit':'CNY分' if k=='priceMinor' else 'g' if k=='netWeightG' else '原文日历日期；时区未统一' if typ=='date' else None} for k,(_,typ) in FIELDS.items()},
      'productSorts':{'reviews_asc':'已收录评论从少到多（含默认及待核）','reviews_desc':'已收录评论从多到少（含默认及待核）',
                      'price_asc':'最低历史观察价从低到高，未知置后','price_desc':'最低历史观察价从高到低，未知置后'},
      'taxonomy':tax,'concepts':CONCEPTS,
      'relationships':['products.id → reviews/prices/specs/rankings/coverage.product_id','documents.id → clauses.document_id','all.source_record_id → source_records.id → source_links/assets/collection_requests','social品牌仅为提及关系，不能推为商品归属'],
      'questionRoutes':[
        {'question':'查某品牌全部资料/找某个商品','entities':['products','reviews','prices','specs'],'method':'只填问题实际约束；商品/平台ID优先精确关联；品牌未知、品类未知不影响未限定该维度的查询'},
        {'question':'份量/便利性/口味/包装等消费反馈','entities':['reviews','social'],'method':'正文概念检索＋同义表达＋原句回读；平台与商品条件继续AND；社媒与购买评价分层计数'},
        {'question':'价格带/指定SKU价格/规格/每100g价格','entities':['prices','specs'],'method':'按商品关联、时间和价格口径筛选；价格快照数≠商品数；未核净重不比较单位价格'},
        {'question':'政策依据/适用条款/营养标签要求','entities':['documents','clauses'],'method':'标题＋正文＋条款检索；读官方原文；应用提示与法规分开；状态按声明截至日'},
        {'question':'榜单、新品、增长趋势','entities':['rankings','products','coverage'],'method':'检查是否只有详情页标识及抽样；无完整榜单或连续快照时明确无法证明排名/增长趋势'},
        {'question':'是否采过、缺什么、来源能否复用','entities':['coverage','products'],'method':'查覆盖与实际记录，并回溯来源请求；失败与无记录分开'},
        {'question':'已做报告/机会卡/项目/配方/工艺','entities':[],'method':'转ksf-workshop-runtime查询业务库；当前资料缺结构化配方/工艺事实时不得由营销词补造'}],
      'unsupported':['不提供自然语言自动理解保证或任意语义向量检索；Agent负责问题拆解，后端返回实际展开词','不执行网络采集、不证明全行业覆盖、不从标题猜配方/净重/功效','看板与Agent是否共享当前代际须核实际registry配置，不能由本地库存在推断已接入生产'],
      'commands':['search','aggregate','facets','get','source'],'defaultEntities':DEFAULT_ENTITIES,
      'rules':['仅约束用户实际指定的维度；不得擅自添加/删除条件','正文检索与字段筛选同时满足；概念同义组内部OR，组间AND','facets只枚举可用数据，不能用其结果自动改变原需求','跨类型返回分开计数；不把社媒评论量、观点数、来源数、商品数混算','来源指纹一致才复用，时效不足标缺口']}


def normalize_scalar(c, field, value):
    col,typ=FIELDS[field]
    if typ=='bool':require(type(value) is bool,'BOOLEAN_REQUIRED:'+field);return int(value)
    if typ=='number':require(type(value) in (int,float) and math.isfinite(value),'NUMBER_REQUIRED:'+field);return value
    require(isinstance(value,str) and bool(value.strip()),'TEXT_REQUIRED:'+field)
    value=value.strip()
    if typ=='date':
        from voc_db import calendar_date
        require(bool(re.fullmatch(r'\d{4}-\d{2}-\d{2}',value)) and calendar_date(value)==value,'ISO_DATE_REQUIRED:'+field)
    elif field in ('brand','category'):
        from voc_db import resolve_code
        tax=json.loads(c.execute("SELECT value FROM meta WHERE key='taxonomy'").fetchone()[0])
        value=resolve_code(value,tax['brands' if field=='brand' else 'categories'])
        require(value!='all','OMIT_UNCONSTRAINED_FIELD_INSTEAD_OF_ALL:'+field)
    elif field=='platform':
        aliases={'京东':'jd','天猫':'tmall','淘宝或待核':'taobao_unverified','小红书':'xiaohongshu','抖音':'douyin','哔哩哔哩':'bilibili'}
        value=aliases.get(value,value)
        require(value in ('jd','tmall','taobao_unverified','xiaohongshu','douyin','bilibili'),'UNKNOWN_PLATFORM')
    return value


def text_groups(request):
    text=request.get('text',{})
    require(isinstance(text,dict) and not set(text)-{'all','any','exclude','concepts'},'INVALID_TEXT_QUERY')
    groups=[];expanded={};excluded=[]
    for section in ('all','any','exclude','concepts'):
        require(isinstance(text.get(section,[]),list) and len(text.get(section,[]))<=12,'INVALID_TEXT_GROUPS')
    for entry in text.get('all',[]):
        group=[entry] if isinstance(entry,str) else entry.get('any') if isinstance(entry,dict) and set(entry)=={'any'} else None
        require(isinstance(group,list) and 1<=len(group)<=20,'INVALID_TEXT_GROUP')
        groups.append(group)
    if text.get('any'):groups.append(text['any'])
    for concept in text.get('concepts',[]):
        require(isinstance(concept,str) and concept in CONCEPTS,'UNKNOWN_CONCEPT')
        group=CONCEPTS[concept]['terms'];groups.append(group);expanded[concept]=group
    for group in groups+[text.get('exclude',[])]:
        for term in group:require(isinstance(term,str) and 0<len(term.strip())<=80 and bool(tokens(term)),'INVALID_LITERAL_TERM')
    return [[norm(t.strip()) for t in group] for group in groups],[norm(t.strip()) for t in text.get('exclude',[])],expanded


def prepare(c,request):
    require(isinstance(request,dict),'INVALID_SEARCH_REQUEST')
    allowed={'entities','filters','text','quality','limit','cursor','groupBy','sort'}
    require(not set(request)-allowed,'UNKNOWN_SEARCH_FIELD')
    entities=request.get('entities',DEFAULT_ENTITIES)
    require(isinstance(entities,list) and 0<len(entities)<=len(ENTITIES) and all(isinstance(e,str) and e in ENTITIES for e in entities) and len(entities)==len(set(entities)),'UNKNOWN_OR_DUPLICATE_ENTITY')
    filters=request.get('filters',{})
    require(isinstance(filters,dict) and not set(filters)-set(FIELDS),'UNKNOWN_FILTER')
    quality=request.get('quality','analysis')
    require(quality in ('analysis','audit'),'UNKNOWN_QUALITY_MODE')
    if quality=='analysis':require(not set(filters)&{'isDefault','identityState','categoryState'},'USE_AUDIT_MODE_FOR_REVIEW_QUALITY_FILTERS')
    if 'ratingValue' in filters:require('ratingScale' in filters and 'platform' in filters,'RATING_PLATFORM_AND_SCALE_REQUIRED')
    clauses=['d.entity IN ('+','.join('?' for _ in entities)+')'];values=list(entities);normalized={}
    for field,spec in filters.items():
        applicable=[e for e in entities if field in ENTITY_FIELDS[e]|COMMON_FIELDS]
        require(bool(applicable),'FILTER_NOT_APPLICABLE:'+field)
        # A hard filter never gets silently ignored for an entity without it.
        col,typ=FIELDS[field];column='d.'+col
        op=spec if isinstance(spec,dict) else {'in':spec} if isinstance(spec,list) else {'eq':spec}
        require(bool(op) and not set(op)-{'eq','in','exists','min','max'},'UNKNOWN_FILTER_OPERATOR:'+field)
        require(not ('eq' in op and 'in' in op),'CONFLICTING_FILTER_OPERATOR:'+field)
        if 'min' in op or 'max' in op:require(typ in ('date','number'),'RANGE_NOT_SUPPORTED:'+field)
        n={}
        for operator,value in op.items():
            if operator=='exists':
                require(type(value) is bool,'BOOLEAN_REQUIRED:'+field)
                if field=='brand' and 'social' in applicable:
                    exists="EXISTS(SELECT 1 FROM social_brand_mentions m WHERE m.post_id=d.record_id)"
                    clauses.append('('+column+' IS NOT NULL'+" OR (d.entity='social' AND "+exists+'))' if value else '('+column+" IS NULL AND (d.entity!='social' OR NOT "+exists+'))')
                else:clauses.append(column+(' IS NOT NULL' if value else ' IS NULL'))
                n[operator]=value
            elif operator=='in':
                require(isinstance(value,list) and 1<=len(value)<=100,'INVALID_IN_FILTER:'+field)
                vals=[normalize_scalar(c,field,v) for v in value]
                if field=='brand' and 'social' in applicable:
                    marks=','.join('?' for _ in vals)
                    clauses.append('('+column+' IN ('+marks+") OR (d.entity='social' AND EXISTS(SELECT 1 FROM social_brand_mentions m WHERE m.post_id=d.record_id AND m.brand_code IN ("+marks+'))))');values+=vals+vals
                else:clauses.append(column+' IN ('+','.join('?' for _ in vals)+')');values+=vals
                n[operator]=vals
            else:
                val=normalize_scalar(c,field,value)
                operator_sql={'eq':'=','min':'>=','max':'<='}[operator]
                if field=='brand' and operator=='eq' and 'social' in applicable:
                    clauses.append('('+column+"=? OR (d.entity='social' AND EXISTS(SELECT 1 FROM social_brand_mentions m WHERE m.post_id=d.record_id AND m.brand_code=?)))");values += [val,val]
                else:clauses.append(column+operator_sql+'?');values.append(val)
                n[operator]=val
        require(not ('min' in n and 'max' in n) or n['min']<=n['max'],'INVALID_RANGE:'+field)
        # Restrict to applicable entity types, including exists:false.
        clauses.append('d.entity IN ('+','.join('?' for _ in applicable)+')');values+=applicable
        normalized[field]=n
    if quality=='analysis':
        clauses.append("(d.entity!='reviews' OR (d.is_default=0 AND d.identity_state='matched' AND d.category_state!='conflict'))")
        if 'category' in filters:
            clauses.append("(d.entity!='reviews' OR d.category_state='consistent')")
        if any(e in entities for e in ('prices',)):
            clauses.append("(d.entity!='prices' OR d.price_minor IS NOT NULL)")
    groups,excluded,expansion=text_groups(request)
    if groups:
        fts_groups=[]
        for group in groups:
            alternatives=[]
            for term in group:
                alternatives.append('('+' AND '.join('"'+t.replace('"','""')+'"' for t in tokens(term))+')')
            fts_groups.append('('+' OR '.join(alternatives)+')')
        clauses.append('d.doc_id IN (SELECT rowid FROM retrieval_fts WHERE retrieval_fts MATCH ?)');values.append(' AND '.join(fts_groups))
        for group in groups:
            clauses.append('('+' OR '.join('instr(d.normalized_text,?)>0' for _ in group)+')');values+=group
    for term in excluded:clauses.append('instr(d.normalized_text,?)=0');values.append(term)
    effective={'entities':entities,'filters':normalized,'text':{'groupsAll':groups,'exclude':excluded,'conceptExpansion':expansion},'quality':quality,
               'defaultReviewPolicy':'非默认＋商品身份匹配；指定品类时购买规格也须明确，未指定品类时不强制猜品类' if quality=='analysis' else '审计模式，包含默认/待核/冲突；不得当有效观点样本'}
    return ' AND '.join(clauses),values,effective


def snippet(text,groups):
    # Normalize only for locating; return the original string and an explicit
    # matching field. No generative summaries masquerade as source quotations.
    text=str(text or '');n=norm(text);hits=[t for g in groups for t in g if t in n]
    index=min((n.find(t) for t in hits),default=0)
    return {'text':text[max(0,index-55):index+160],'matchedTerms':sorted(set(hits)),'literalMatchOnly':True}


def search(c,request,explain=False):
    where,values,effective=prepare(c,request)
    limit=request.get('limit',20)
    require(type(limit) is int and 1<=limit<=100,'INVALID_LIMIT')
    sort=request.get('sort')
    sorts={
        'reviews_asc':('(SELECT count(*) FROM reviews r WHERE r.product_id=d.record_id)','ASC'),
        'reviews_desc':('(SELECT count(*) FROM reviews r WHERE r.product_id=d.record_id)','DESC'),
        'price_asc':('(SELECT min(p.amount_minor) FROM prices p WHERE p.product_id=d.record_id)','ASC'),
        'price_desc':('(SELECT min(p.amount_minor) FROM prices p WHERE p.product_id=d.record_id)','DESC'),
    }
    require(sort is None or isinstance(sort,str) and sort in sorts,'INVALID_SORT')
    require(sort is None or effective['entities']==['products'],'SORT_NOT_APPLICABLE')
    expression,direction=sorts[sort] if sort else ('d.doc_id','ASC')
    if sort:effective={**effective,'sort':sort}
    generation=c.execute("SELECT value FROM meta WHERE key='generation'").fetchone()[0]
    context=hashlib.sha256(canonical([generation,effective]).encode()).hexdigest()
    total=c.execute('SELECT count(*) FROM retrieval_documents d WHERE '+where,values).fetchone()[0]
    totals={r[0]:r[1] for r in c.execute('SELECT entity,count(*) FROM retrieval_documents d WHERE '+where+' GROUP BY entity',values)}
    text_counts={r[0]:r[1] for r in c.execute('SELECT entity,count(DISTINCT content_fingerprint) FROM retrieval_documents d WHERE '+where+' GROUP BY entity',values)}
    matched_products=c.execute('SELECT count(DISTINCT product_id) FROM retrieval_documents d WHERE '+where,values).fetchone()[0]
    if request.get('cursor'):
        try:
            cursor=json.loads(base64.urlsafe_b64decode(request['cursor']))
            require(set(cursor)==({'context','after','value'} if sort else {'context','after'}) and cursor['context']==context and type(cursor['after']) is int,'CURSOR_SCOPE_MISMATCH')
            if sort:require(cursor['value'] is None or type(cursor['value']) in (int,float) and math.isfinite(cursor['value']),'CURSOR_SCOPE_MISMATCH')
        except (ValueError,TypeError,KeyError):
            require(False,'CURSOR_SCOPE_MISMATCH')
        if not sort:
            where+=' AND d.doc_id>?';values.append(cursor['after'])
        elif cursor['value'] is None:
            where+=' AND ('+expression+' IS NULL AND d.doc_id>?)';values.append(cursor['after'])
        else:
            comparison='>' if direction=='ASC' else '<'
            where+=' AND ('+expression+' IS NULL OR '+expression+comparison+'? OR ('+expression+'=? AND d.doc_id>?))'
            values += [cursor['value'],cursor['value'],cursor['after']]
    order=expression+' IS NULL ASC,'+expression+' '+direction+',d.doc_id ASC' if sort else 'd.doc_id'
    sql='SELECT d.*'+(','+expression+' AS _sort_value' if sort else '')+' FROM retrieval_documents d WHERE '+where+' ORDER BY '+order+' LIMIT ?'
    rows=c.execute(sql,values+[limit+1]).fetchall();has_more=len(rows)>limit;rows=rows[:limit]
    next_cursor=None
    if has_more:
        last=dict(context=context,after=rows[-1]['doc_id'])
        if sort:last['value']=rows[-1]['_sort_value']
        next_cursor=base64.urlsafe_b64encode(canonical(last).encode()).decode()
    from voc_db import citation
    records=[]
    for row in rows:
        r=dict(row);r.pop('tokens');r.pop('normalized_text');r.pop('_sort_value',None);body=r.pop('body')
        r['match']=snippet(body if r['entity']=='reviews' else r['title']+'\n'+body,effective['text']['groupsAll'])
        if r['entity']=='clauses':
            raw=c.execute('SELECT excerpt,application_note,excerpt_type FROM policy_clauses WHERE id=?',(r['record_id'],)).fetchone()
            r['match']['matchedFields']=[name for name in ('excerpt','application_note') if any(term in norm(raw[name]) for group in effective['text']['groupsAll'] for term in group)]
            r['match']['excerptType']=raw['excerpt_type']
            r['match']['applicationNoteIsRegulation']=False
        if r['entity']=='documents':r['policyStatusAsOf']='2026-11-04'
        r['citation']=citation(c,r['source_record_id']);r['detailCommand']={'entity':r['entity'],'id':r['record_id']}
        if r['entity']=='social':r['brandRelationship']='text_mention_only_not_product_ownership'
        records.append(r)
    result={'ok':True,'retrievalContract':VERSION,'generation':generation,'readOnly':True,'databaseWrites':0,
            'effectiveQuery':effective,'totalMatched':total,'countsByEntity':totals,'distinctTextFingerprintsByEntity':text_counts,'distinctLinkedProducts':matched_products,'records':records,'nextCursor':next_cursor,
            'coverageNotice':'仅为当前资料及实际筛选命中；同义词为显示的有限展开，不能保证所有语义表达均命中。未知的未限定维度保留；指定维度未知则无法满足该条件。'}
    if explain:result['queryPlan']=[r['detail'] for r in c.execute('EXPLAIN QUERY PLAN '+sql,values+[limit+1])]
    return result


def aggregate(c,request,facets=False):
    where,values,effective=prepare(c,request)
    fields=request.get('groupBy',['platform','brand','category'] if facets else ['platform'])
    require(isinstance(fields,list) and 1<=len(fields)<=4 and all(isinstance(f,str) and f in FIELDS for f in fields) and len(fields)==len(set(fields)),'INVALID_GROUP_BY')
    require(not ('brand' in fields and 'social' in effective['entities']),'SOCIAL_BRAND_IS_MENTION_NOT_GROUPABLE_OWNERSHIP')
    limit=request.get('limit',50);require(type(limit) is int and 1<=limit<=100,'INVALID_LIMIT')
    out={}
    groups=[[field] for field in fields] if facets else [fields]
    for group in groups:
        columns=[FIELDS[f][0] for f in group]
        select=','.join('d.'+col for col in columns)
        sql='SELECT '+select+',count(*) AS recordCount,count(DISTINCT d.product_id) AS linkedProductCount,count(DISTINCT d.content_fingerprint) AS distinctTextCount FROM retrieval_documents d WHERE '+where+' GROUP BY '+select+' ORDER BY recordCount DESC,'+select+' LIMIT ?'
        rows=[dict(row) for row in c.execute(sql,values+[limit+1])]
        out[','.join(group)]={'rows':rows[:limit],'truncated':len(rows)>limit}
    return {'ok':True,'retrievalContract':VERSION,'readOnly':True,'databaseWrites':0,'effectiveQuery':effective,'groups':out,
            'countingBasis':'recordCount是记录数；linkedProductCount是关联商品去重数；distinctTextCount按归一化文字去重，不等于独立消费者/来源或真实购买数；社媒品牌仅提及。'}