kangshifu-board/scripts/enrich-research-library.py

"""Build traceable enrichment batches from byte-preserved VOC responses."""
import argparse
from collections import Counter
import hashlib
import json
from pathlib import Path
import re
import sys

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / 'server/voc-library'))
WORK = ROOT / 'work/database-enrichment-20261110'


def read(p): return json.loads(Path(p).read_bytes())
def sha(b): return hashlib.sha256(b).hexdigest()
def save(p, value):
    p = Path(p); p.parent.mkdir(parents=True, exist_ok=True)
    p.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding='utf-8')
def filename(s): return re.sub(r'[^a-zA-Z0-9_-]', '_', s)


def payload(doc):
    outer = doc.get('data', {})
    if not isinstance(outer, dict) or outer.get('code', 0) not in (0, 200):
        raise ValueError('UPSTREAM_BUSINESS_ERROR')
    return outer.get('data', outer), outer.get('recordTime')


def reviews(doc, family, item):
    d, observed = payload(doc)
    if not isinstance(d, dict): raise ValueError('REVIEW_PAYLOAD_NOT_OBJECT')
    rows = d.get('data' if family == 'jd' else 'comments', [])
    if not isinstance(rows, list): raise ValueError('REVIEW_ARRAY_REQUIRED')
    accepted, rejected, seen = [], Counter(), set()
    for i, r in enumerate(rows):
        actual = r.get('productId') if family == 'jd' else r.get('auction', {}).get('aucNumId')
        rid = r.get('commentId' if family == 'jd' else 'rateId')
        if str(actual) != item:
            rejected['REVIEW_PRODUCT_ID_MISMATCH'] += 1; continue
        if type(rid) not in (str, int) or not str(rid):
            rejected['REVIEW_ID_MISSING'] += 1; continue
        if str(rid) in seen: rejected['REPEATED_REVIEW_ID_IN_PAGE'] += 1; continue
        seen.add(str(rid))
        content = r.get('commentData' if family == 'jd' else 'content')
        if not isinstance(content, str) or not content.strip():
            rejected['EMPTY_REVIEW'] += 1; continue
        template = bool(re.fullmatch(r'(该用户觉得商品非常好，给出5星好评|此用户未填写评价内容|此用户没有填写评价|此用户未填写评价|此用户未及时填写评价内容，系统默认好评[！!。]?)', content.strip()))
        accepted.append({'itemId': item, 'reviewId': str(rid), 'content': content,
            'appendContent': r.get('afterComment') if family == 'jd' else r.get('appendList') or r.get('append'),
            'reviewDate': r.get('commentDate' if family == 'jd' else 'date'),
            'rating': r.get('commentScore' if family == 'jd' else 'rate'),
            'ratingScale': '1-5' if family == 'jd' else None,
            'purchasedSku': r.get('productColor') if family == 'jd' else r.get('auction', {}).get('sku'),
            'isDefault': True if template else None, 'sourceTime': observed,
            'originPointer': '/data/data/' + ('data/' if family == 'jd' else 'comments/') + str(i)})
    return accepted, rejected


def detail(doc, family, item):
    d, observed = payload(doc); specs = []; price = None
    if not isinstance(d, dict): raise ValueError('DETAIL_PAYLOAD_NOT_OBJECT')
    p = d.get('product' if family == 'jd' else 'item', {})
    if not isinstance(p, dict): raise ValueError('DETAIL_PRODUCT_NOT_OBJECT')
    if str(p.get('skuId' if family == 'jd' else 'itemId')) != item:
        raise ValueError('DETAIL_PRODUCT_ID_MISMATCH')
    if family == 'jd':
        parts = [(k, p[k]) for k in ('color', 'size', 'spec') if isinstance(p.get(k), str) and p[k].strip()]
        if parts:
            specs.append({'itemId': item, 'skuId': item, 'specText': '；'.join(v for k, v in parts),
                          'verification': 'source_attribute',
                          'basis': '当前SKU销售属性原文；字段=' + ','.join(k for k, v in parts) + '；不将运输重量或标题推断为净含量。',
                          'originPointers': ['/data/data/product/' + k for k, v in parts]})
        amount = d.get('priceFloor', {}).get('price')
        basis = 'VOC jd.get_item_detail_v1 priceFloor.price 页面报价；活动、地区与登录条件见原始响应'
    else:
        base = d.get('skuBase') or {}; props = {}
        for prop in base.get('props', []):
            for v in prop.get('values', []): props[str(prop.get('pid')) + ':' + str(v.get('vid'))] = (prop.get('name'), v.get('name'))
        for i, sku in enumerate(base.get('skus', [])):
            tokens = str(sku.get('propPath', '')).split(';')
            if not sku.get('skuId') or not tokens or any(t not in props for t in tokens): continue
            text = '；'.join(str(props[t][0]) + ':' + str(props[t][1]) for t in tokens)
            specs.append({'itemId': item, 'skuId': str(sku['skuId']), 'specText': text,
                'verification': 'source_attribute', 'basis': 'skuBase.skus.propPath逐项连接props原始属性；不同SKU分别存储，不将备选口味合并为一个SKU。',
                'originPointers': ['/data/data/skuBase/skus/' + str(i), '/data/data/skuBase/props']})
        amount = p.get('price'); basis = 'VOC taobao.get_item_detail_v6 item.price 商品级展示价；非特定SKU成交价'
        if any(t in str(p.get('title', '')) for t in ('积分', '定金', '补差', '运费')): amount = None
    if type(amount) in (str, float, int) and re.fullmatch(r'\d+(\.\d{1,2})?', str(amount)) and float(amount) > 0:
        price = {'itemId': item, 'amount': str(amount), 'currency': 'CNY', 'priceBasis': basis, 'sourceTime': observed}
    return specs, price


def social_body(doc, platform, iid):
    d, _ = payload(doc)
    if platform == 'bilibili':
        # Only the primary video object is eligible. Recommendations, playlists
        # and comments can repeat the requested ID and are not author body text.
        if isinstance(d, dict) and d.get('message') and not d.get('bvid') and not d.get('videoData'):
            raise ValueError('UPSTREAM_DETAIL_UNAVAILABLE')
        note = d.get('videoData', d) if isinstance(d, dict) else None
        if not isinstance(note, dict) or note.get('bvid') != iid:
            raise ValueError('SOCIAL_ID_MISMATCH')
        text = note.get('desc') or ''.join(x.get('raw_text', '') for x in note.get('desc_v2') or [] if isinstance(x, dict))
        title = note.get('title', '')
    elif platform == 'xiaohongshu':
        notes = []
        if isinstance(d, dict):
            notes = d.get('note_list', [d])
        elif isinstance(d, list):
            for entry in d:
                if isinstance(entry, dict): notes.extend(entry.get('note_list', [entry]))
        found = [n for n in notes if isinstance(n, dict) and str(n.get('id') or n.get('note_id')) == iid]
        if not found: raise ValueError('SOCIAL_ID_MISMATCH')
        # Two conflicting primary objects require review, not a longest-text guess.
        if len({(n.get('desc') or '', n.get('title') or '') for n in found}) > 1:
            raise ValueError('SOCIAL_PRIMARY_OBJECT_CONFLICT')
        note = found[0]
        text, title = note.get('desc') or '', note.get('title') or ''
    else: raise ValueError('SOCIAL_PLATFORM_UNSUPPORTED')
    if not isinstance(text, str) or not isinstance(title, str): raise ValueError('SOCIAL_TEXT_TYPE_INVALID')
    if not text.strip(): raise ValueError('UPSTREAM_BODY_EMPTY')
    norm = lambda s: re.sub(r'\W', '', s)
    if not norm(text) or text.strip().lower() in ('null', 'undefined', 'none', '暂无简介', '暂无描述', '无简介'):
        raise ValueError('UPSTREAM_BODY_PLACEHOLDER')
    if norm(text) == norm(title): raise ValueError('BODY_ONLY_REPEATS_TITLE')
    return text


def build(pilot=False, selected=None, name='pilot'):
    import library_channel
    import structured_batch
    snapshot = read(ROOT / 'work/industry-monitoring-20261107-r1/snapshot.json')
    products = {r['record_id']: r for r in snapshot['entities']['products']}
    queue = read(WORK / 'product-queue.json')
    if pilot:
        ids = [p['productId'] for p in queue if p['platform'] == 'jd'][:3] + [p['productId'] for p in queue if p['platform'] != 'jd'][:3]
    else: ids = selected or [p['productId'] for p in queue]
    directory = WORK / 'batches' / name
    if (directory / 'transport.zip').exists(): raise ValueError('BATCH_ALREADY_SEALED')
    batch = {'schema': 'ksf.voc-agent-batch.v2', 'batchId': 'ksf-enrichment-20261110-' + name,
             'producer': 'Codex: VOC 0.3.44 exact-ID enrichment; preserved upstream bytes and explicit mapping adapters',
             'sources': [], 'attachments': [], 'products': [], 'reviews': [], 'specs': [], 'prices': [], 'social': []}
    report = {'baselineGeneration': snapshot['generation'], 'selectedProducts': len(ids), 'accepted': Counter(), 'excluded': Counter(), 'products': []}
    seen_reviews = set(); adapters = {}; originals = {}
    for pid in ids:
        p = products[pid]; family, iid = pid.split(':', 1)
        marketplace = 'jd' if family == 'jd' else 'tmall' if p['platform'] == 'tmall' else 'taobao_unverified'
        group = marketplace
        adapter = adapters.setdefault(group, {'kind': 'explicit_mapping_adapter_not_upstream_response',
            'context': {'kind': 'existing_library_product_context_not_new_observation', 'generation': snapshot['generation'], 'exportedAt': snapshot['exportedAt']},
            'products': [], 'reviews': [], 'specs': [], 'prices': []})
        accepted = Counter()
        for endpoint in (('jd.get_item_detail_v1' if family == 'jd' else 'taobao.get_item_detail_v6'), ('jd.get_item_comments_v1' if family == 'jd' else 'taobao.get_item_comment_v3')):
            prefix = WORK / 'collected' / (filename(pid) + '--' + filename(endpoint))
            mp, rp = Path(str(prefix) + '.meta.json'), Path(str(prefix) + '.response.json')
            if not mp.exists() or not rp.exists(): continue
            meta = read(mp); raw = rp.read_bytes()
            if not meta.get('ok'): report['excluded']['REQUEST_FAILED'] += 1; continue
            if meta['sha256'] != sha(raw): raise ValueError('ORIGINAL_RESPONSE_HASH_MISMATCH')
            doc = json.loads(raw)
            origin = {'path': 'raw/original/' + rp.name, 'sha256': sha(raw), 'endpoint': endpoint,
                      'requestParams': meta['params'], 'collectedAt': meta['finishedAt']}
            try:
                if 'comment' in endpoint:
                    rows, rejects = reviews(doc, family, iid); report['excluded'].update(rejects)
                    for r in rows:
                        key = pid + ':' + r['reviewId']
                        if key in seen_reviews: continue
                        seen_reviews.add(key); adapter['reviews'].append({**r, 'origin': origin}); accepted['reviews'] += 1
                else:
                    specs, price = detail(doc, family, iid)
                    for r in specs: adapter['specs'].append({**r, 'origin': origin}); accepted['specs'] += 1
                    if price: adapter['prices'].append({**price, 'origin': origin}); accepted['prices'] += 1
                originals[origin['path']] = raw
            except ValueError as e: report['excluded'][str(e)] += 1
        if accepted:
            adapter['products'].append({'itemId': iid, 'title': p['title'], 'shopId': p.get('shop_id'), 'shopName': p.get('shop_name'),
                                         'sourceTime': p.get('source_time_raw'), 'originRecordId': pid})
            report['products'].append({'id': pid, **accepted}); report['accepted'].update(accepted)
    from intake_contract import ENTITIES
    for group, adapter in adapters.items():
        if not adapter['products']: continue
        rel = 'raw/adapter-' + group + '.json'; save(directory / rel, adapter)
        sid = 'adapter-' + group
        batch['sources'].append({'id': sid, 'path': rel, 'sha256': sha((directory / rel).read_bytes()),
                                 'platform': 'jd' if group == 'jd' else 'taobao', 'marketplace': group})
        for entity in ('products', 'reviews', 'specs', 'prices'):
            for i, row in enumerate(adapter[entity]):
                batch[entity].append({'sourceId': sid, 'pointer': '/' + entity + '/' + str(i),
                                      'fields': {k: '/' + k for k in row if k in ENTITIES[entity][2]}})
    for rel, raw in originals.items():
        p = directory / rel; p.parent.mkdir(parents=True, exist_ok=True); p.write_bytes(raw)
        batch['attachments'].append({'path': rel, 'sha256': sha(raw)})
    save(directory / 'batch.json', batch)
    if not batch['products']:
        save(directory / 'receipt.json', {**report, 'batchId': batch['batchId'], 'empty': True})
        print(json.dumps({'empty': True, 'selectedProducts': len(ids)})); return
    check = structured_batch.build(directory / 'batch.json', directory / 'staging.sqlite', False)
    packed, digest = library_channel.pack(directory / 'batch.json')
    (directory / 'transport.zip').write_bytes(packed)
    receipt = {**report, 'validation': check, 'archiveSha256': digest, 'archiveBytes': len(packed), 'batchId': batch['batchId'],
               'originalResponses': len(originals), 'originalBytes': sum(map(len, originals.values()))}
    save(directory / 'receipt.json', receipt)
    print(json.dumps(receipt, ensure_ascii=False))


if __name__ == '__main__':
    p = argparse.ArgumentParser(); p.add_argument('--pilot', action='store_true'); p.add_argument('--selection'); p.add_argument('--name', default='pilot')
    a = p.parse_args(); build(a.pilot, read(a.selection) if a.selection else None, a.name)