kangshifu-board/scripts/expand-product-review-coverage.py

"""Plan a bounded second round of missing-product review collection.

Uses existing production counts and original historical volume evidence, never
joins different links by a similar title. Does not collect or publish itself.
"""
import argparse
from collections import Counter, defaultdict
import hashlib
import json
from pathlib import Path
import re
import sys
import zipfile

ROOT = Path(__file__).resolve().parents[1]
WORK = ROOT / 'work/product-reviews/round2'
sys.path.insert(0, str(ROOT / 'server/voc-library'))
import agent_batch


def write(path, value):
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open('x', encoding='utf-8', newline='\n') as f:
        json.dump(value, f, ensure_ascii=False, indent=2)
        f.write('\n')


def displayed_count(value):
    match = re.search(r'\d+(?:\.\d+)?', str(value))
    return float(match[0]) * (10000 if '万' in str(value) else 1) if match else 0


def plan():
    catalog = json.loads((WORK / 'catalog.json').read_text(encoding='utf-8'))
    volumes = json.loads((ROOT / 'work/product-reviews/volume-evidence-r1.json').read_text(encoding='utf-8'))
    prior = []
    for p in (ROOT / 'work/product-reviews').glob('live-*/*/reservation.json'):
        r = json.loads(p.read_text(encoding='utf-8'))
        prior.append((r['endpoint'].split('.')[0], r['params']['itemId']))
    skip = set(prior)
    pools = defaultdict(list)
    for r in catalog['records']:
        pid = r['record_id']
        family, iid = pid.split(':')
        evidence = volumes.get(pid)
        if r.get('relatedCounts', {}).get('reviews') or (family, iid) in skip or not evidence:
            continue
        if displayed_count(evidence['countText']) < 10000 or r.get('category') not in (
                'banmian', 'soup_noodle', 'instant_unspecified', 'mianpi', 'reganmian', 'fried_noodle'):
            continue
        pools[r.get('brand')].append({'productId': pid, 'title': r['title'], 'brand': r.get('brand'),
            'category': r.get('category'), 'shopId': r.get('shop_id'), 'marketplace': r['platform'],
            'volumeEvidence': evidence, 'storedReviews': 0, 'page': '1' if family == 'jd' else 1,
            'endpoint': 'jd.get_item_comments_v1' if family == 'jd' else 'taobao.get_item_comment_v3'})
    quotas = {'ksf': 26, 'tongyi': 6, 'baixiang': 6, 'jinmailang': 6,
              'riqing': 5, 'nongxin': 4, 'akuan': 4, 'wugudaochang': 3}
    selected_by_brand = {}
    for brand, quota in quotas.items():
        candidates = sorted(pools[brand], key=lambda r: displayed_count(r['volumeEvidence']['countText']), reverse=True)
        chosen = []
        # Spread shop/category buckets while preserving volume priority within
        # each bucket; exact identity is retained for every candidate.
        buckets = defaultdict(list)
        for r in candidates:
            buckets[(r['shopId'], r['category'])].append(r)
        while buckets and len(chosen) < quota:
            for key in list(buckets):
                if len(chosen) >= quota:
                    break
                chosen.append(buckets[key].pop(0))
                if not buckets[key]:
                    del buckets[key]
        selected_by_brand[brand] = chosen
    selected = []
    # Interleave brands to learn actual access before expanding the full plan.
    while any(selected_by_brand.values()):
        for brand in quotas:
            if selected_by_brand[brand]:
                selected.append(selected_by_brand[brand].pop(0))
    if len(selected) > 60:
        raise ValueError('SIXTY_PRODUCT_LIMIT')
    original = Path('F:/文档/康师傅VOC_全部已采集数据_20261105_191821.zip')
    archive_sha = agent_batch.digest(original)
    evidence_dir = WORK / 'volume-sources'
    with zipfile.ZipFile(original) as archive:
        for r in selected:
            member = r['volumeEvidence']['sourceMember']
            raw = archive.read(member)
            name = hashlib.sha256(member.encode()).hexdigest()[:20] + '.json'
            target = evidence_dir / name
            target.parent.mkdir(parents=True, exist_ok=True)
            if not target.exists():
                target.write_bytes(raw)
            payload = json.loads(raw)
            iid = r['productId'].split(':')[1]
            matches = []
            def walk(node, pointer=''):
                if isinstance(node, dict):
                    if any(str(node.get(k)) == iid for k in ('itemId', 'id', 'skuId')) and str(node.get('cc')) == r['volumeEvidence']['countText']:
                        matches.append(pointer)
                    for k, v in node.items():
                        if isinstance(v, (dict, list)):
                            walk(v, pointer + '/' + k.replace('~', '~0').replace('/', '~1'))
                elif isinstance(node, list):
                    for i, v in enumerate(node):
                        walk(v, pointer + '/' + str(i))
            walk(payload)
            if not matches:
                raise ValueError('VOLUME_EVIDENCE_POINTER_NOT_VERIFIED:' + r['productId'])
            r['volumeEvidence'].update({'originalArchiveSha256': archive_sha, 'file': 'volume-sources/' + name,
                'sha256': hashlib.sha256(raw).hexdigest(), 'pointer': matches[0] + '/cc',
                'scope': 'historical_platform_display_not_current_exact_count'})
    result = {'schema': 'ksf.review-coverage-plan.v1', 'round': '20261106-r2', 'generation': catalog['generation'],
        'products': len(catalog['records']), 'productsWithReviews': sum(bool(r.get('relatedCounts', {}).get('reviews')) for r in catalog['records']),
        'candidates': selected, 'candidateCount': len(selected), 'maxNewGatewayRequests': len(selected),
        'maxNewCostCny': round(len(selected) * .30, 2), 'unitPriceCny': .30, 'retries': 0,
        'priorTaskConservativeReservationCny': 9.90, 'firstReviewRoundConfirmedAndReservedCny': 2.10,
        'sharedTaskBudgetCny': 50, 'worstCaseOccupiedCny': round(9.90 + 2.10 + len(selected) * .30, 2),
        'brandCounts': dict(Counter(r['brand'] for r in selected)), 'maxReviewsPerProduct': 20,
        'selection': 'missing reviews; historical display >=10000; own brand priority; shop/category spread',
        'noGlobalLikesSortClaim': True, 'productionImported': False}
    assert result['worstCaseOccupiedCny'] <= 50
    write(WORK / 'plan-expanded.json', result)
    print(json.dumps({k: v for k, v in result.items() if k != 'candidates'}, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('command', choices=['plan'])
    args = parser.parse_args()
    if args.command == 'plan':
        plan()