kangshifu-board/scripts/build-expanded-review-intake.py

"""Build small source-verified review batches from the second collection round.

Only exact request/returned identities and non-placeholder texts are retained.
Likes stay in evidence, and absent default/rating-scale fields stay unknown.
"""
import argparse
from collections import Counter
import importlib.util
import json
from pathlib import Path
import shutil
import sys
import zipfile

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / 'server/voc-library'))
import agent_batch
import board_library
import library_channel
import structured_batch
import update_library
import voc_db

spec = importlib.util.spec_from_file_location('sample_summary', ROOT / 'scripts/summarize-product-review-samples.py')
samples = importlib.util.module_from_spec(spec)
spec.loader.exec_module(samples)


def save(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 copy_source(source, destination):
    destination.parent.mkdir(parents=True, exist_ok=True)
    shutil.copyfile(source, destination)
    return {'path': destination, 'sha256': agent_batch.digest(destination)}


def build_group(records, ordinal, plan, target):
    target.mkdir(parents=True, mode=0o700)
    batch = {'schema': 'ksf.voc-agent-batch.v2', 'batchId': 'review-coverage-20261106-r2-' + str(ordinal).zfill(2),
        'producer': 'Codex: direct review coverage expansion; exact product identity; observed likes retained',
        'sources': [], 'attachments': [], 'products': [], 'reviews': []}
    chosen = []
    diagnostics = []
    for item in records:
        pid = item['productId']
        family, iid = pid.split(':')
        label = item['label']
        folder = ROOT / 'work/product-reviews/live-r2' / label
        envelope_file = folder / 'mcp-result.json'
        envelope = json.loads(envelope_file.read_text(encoding='utf-8'))
        reservation = json.loads((folder / 'reservation.json').read_text(encoding='utf-8'))
        detail = json.loads((folder / 'product-context.json').read_text(encoding='utf-8'))
        if reservation['catalogGeneration'] != plan['generation'] or detail['generation'] != plan['generation']:
            raise ValueError('CONTEXT_GENERATION_MISMATCH')
        context = detail['record']
        if context['id'] != pid or context['item_id'] != iid:
            raise ValueError('PRODUCT_CONTEXT_ID_MISMATCH')
        candidate = next(r for r in plan['candidates'] if r['productId'] == pid)
        name = 'raw/context/' + label + '.json'
        copy_source(folder / 'product-context.json', target / name)
        sid = 'product-' + label
        batch['sources'].append({'id': sid, 'path': name, 'sha256': agent_batch.digest(target / name),
            'platform': family, 'marketplace': context['marketplace'], 'requestItemId': iid,
            'observedAt': context.get('source_time_raw'), 'url': context.get('url')})
        batch['products'].append({'sourceId': sid, 'pointer': '/record', 'fields': {
            'itemId': '/item_id', 'title': '/title', 'url': '/url', 'shopId': '/shop_id',
            'shopName': '/shop_name', 'sourceTime': '/source_time_raw'}})
        # Retain exact MCP envelopes plus media URL mapping. Already-downloaded
        # buyer images/avatars remain in private live output; they are not
        # product main images and are unnecessary for the textual review batch.
        source_files = [folder / 'mcp-result.json', folder / 'reservation.json']
        source_files.extend(f for f in (folder / 'tool-output').rglob('*.json') if f.is_file())
        for source in source_files:
            relative = 'raw/live/' + label + '/' + source.relative_to(folder).as_posix()
            copy_source(source, target / relative)
            batch['attachments'].append({'path': relative, 'sha256': agent_batch.digest(target / relative)})
        volume = candidate['volumeEvidence']
        vname = 'raw/volume/' + Path(volume['file']).name
        if not (target / vname).exists():
            copy_source(ROOT / 'work/product-reviews/round2' / volume['file'], target / vname)
            batch['attachments'].append({'path': vname, 'sha256': agent_batch.digest(target / vname)})
        payload = envelope['response']['structuredContent']['data']['result']
        upstream = payload['data'] if family == 'jd' else payload['comments']
        meaningful = [r for r in item['records'] if r['matchesRequestedProduct'] and not r['isPlaceholder'] and r['content'].strip()]
        meaningful.sort(key=lambda r: r['likes'] if type(r['likes']) is int else -1, reverse=True)
        seen = set()
        selected = []
        for inspected in meaningful:
            rid = inspected['reviewId']
            if rid in seen:
                continue
            seen.add(rid)
            original = upstream[inspected['index']]
            original_id = original['commentId'] if family == 'jd' else original['rateId']
            if type(original_id) is int and original_id > 2 ** 53 - 1:
                raise ValueError('UNSAFE_NUMERIC_REVIEW_ID')
            pointer = '/response/structuredContent/data/result/' + ('data/' if family == 'jd' else 'comments/') + str(inspected['index'])
            if agent_batch.pointer(envelope, pointer) != original:
                raise ValueError('SOURCE_POINTER_MISMATCH')
            selected.append({'itemId': iid, 'reviewId': rid, 'content': inspected['content'],
                'appendContent': original.get('afterComment') if family == 'jd' else original.get('appendList'),
                'reviewDate': original.get('commentDate') if family == 'jd' else original.get('date'),
                'purchasedSku': original.get('productColor') if family == 'jd' else original.get('auction', {}).get('sku'),
                'rating': original.get('commentScore') if family == 'jd' else original.get('rate'),
                'ratingScale': None, 'isDefault': None, 'sourceTime': item['sourceTime'], 'likeCount': inspected['likes'],
                'productId': pid, 'marketplace': context['marketplace'],
                'origin': {'tool': 'voc_api_call', 'endpoint': reservation['endpoint'],
                    'sourceFile': 'raw/live/' + label + '/mcp-result.json', 'sourcePointer': pointer,
                    'sourceSha256': agent_batch.digest(envelope_file), 'usageTraceId': reservation['usageTraceId'],
                    'mediaUrlsRewrittenByExistingMcp': True},
                'selection': {'page': reservation['params']['page'], 'order': reservation['params'].get('orderType', 'upstream_default'),
                    'scope': 'observed_likes_within_one_returned_page_not_platform_global_top',
                    'historicalVolume': volume, 'rawDefaultFlag': 'not_provided_unknown'}})
            if len(selected) == 20:
                break
        chosen.extend(selected)
        diagnostics.append({k: v for k, v in item.items() if k != 'records'})
    if not chosen:
        raise ValueError('NO_QUALIFYING_REVIEWS')
    save(target / 'raw/derived/reviews.json', {'method': 'exact_identity_non_placeholder_reviews_sorted_by_observed_likes',
        'maxPerProduct': 20, 'globalTopLikesVerified': False, 'records': chosen})
    digest = agent_batch.digest(target / 'raw/derived/reviews.json')
    for index, row in enumerate(chosen):
        family, iid = row['productId'].split(':')
        sid = 'review-' + str(index)
        batch['sources'].append({'id': sid, 'path': 'raw/derived/reviews.json', 'sha256': digest,
            'platform': family, 'marketplace': row['marketplace'], 'requestItemId': iid, 'observedAt': row['sourceTime']})
        fields = {name: '/' + name for name in ['itemId', 'reviewId', 'content', 'reviewDate', 'purchasedSku',
            'rating', 'ratingScale', 'isDefault', 'sourceTime']}
        if row['appendContent'] is not None:
            fields['appendContent'] = '/appendContent'
        batch['reviews'].append({'sourceId': sid, 'pointer': '/records/' + str(index), 'fields': fields})
    save(target / 'batch.json', batch)
    dry = structured_batch.build(target / 'batch.json', target / 'staging.sqlite')
    built = structured_batch.build(target / 'batch.json', target / 'staging.sqlite', True)
    meta = update_library.verified(target / 'staging.sqlite')
    with voc_db.readonly(target / 'staging.sqlite') as c:
        staged = {r['id']: dict(r) for r in c.execute('SELECT * FROM reviews')}
        if len(staged) != len(chosen):
            raise ValueError('DUPLICATE_REVIEW_IDS')
        for row in chosen:
            current = staged[row['productId'] + ':' + row['reviewId']]
            if current['content'] != row['content'] or current['product_id'] != row['productId'] or current['is_default'] is not None:
                raise ValueError('READBACK_CONTENT_IDENTITY_OR_UNKNOWN_STATE_MISMATCH')
        issues = {r[0]: r[1] for r in c.execute('SELECT code,count(*) FROM quality_issues GROUP BY code')}
    wire, fingerprint = library_channel.pack(target / 'batch.json')
    (target / 'channel-source.zip').write_bytes(wire)
    report = {'ok': True, 'batchId': batch['batchId'], 'expectedProductionGeneration': plan['generation'],
        'products': len({r['productId'] for r in chosen}), 'reviews': len(chosen),
        'verifiedReviewReadback': len(staged), 'reviewCounts': dict(Counter(r['productId'] for r in chosen)),
        'positiveLikeReviews': sum(type(r['likeCount']) is int and r['likeCount'] > 0 for r in chosen),
        'maxObservedLikes': max((r['likeCount'] for r in chosen if type(r['likeCount']) is int), default=None),
        'unknownDefaultStatesRetained': len(chosen), 'issues': issues, 'stagingGeneration': meta['generation'],
        'channelArchiveSha256': fingerprint, 'channelArchiveBytes': len(wire),
        'sourceFilesVerified': dry['sourceFilesVerified'], 'paidCollectionCallsInBuild': 0,
        'productionImported': False, 'preflight': dry, 'staging': built, 'collectionDiagnostics': diagnostics}
    save(target / 'verification.json', report)
    return report


def main():
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument('--out', type=Path, required=True)
    args = p.parse_args()
    if args.out.exists():
        raise ValueError('NEW_OUTPUT_DIRECTORY_REQUIRED')
    root = args.out.absolute()
    plan = json.loads((ROOT / 'work/product-reviews/round2/plan-expanded.json').read_text(encoding='utf-8'))
    collection = samples.summarize(ROOT / 'work/product-reviews/live-r2')
    if any(r['status'] == 'pending' for r in collection['records']):
        raise ValueError('PENDING_REQUEST_NEEDS_STATUS_CHECK')
    qualifying = [r for r in collection['records'] if r['status'] == 'ok' and r['meaningfulMatchingReviews'] > 0]
    reports = []
    for start in range(0, len(qualifying), 10):
        ordinal = len(reports) + 1
        reports.append(build_group(qualifying[start:start + 10], ordinal, plan, root / ('batch-' + str(ordinal).zfill(2))))
    report = {'ok': True, 'expectedProductionGeneration': plan['generation'], 'plannedCandidates': plan['candidateCount'],
        'collectionCalls': collection['collectionCalls'], 'successfulCalls': collection['successfulCalls'],
        'confirmedNewCostCny': collection['confirmedCostCny'], 'unsettledNewReservationCny': collection['unsettledReservationCny'],
        'sourceVerifiedProducts': sum(r['products'] for r in reports), 'selectedReviews': sum(r['reviews'] for r in reports),
        'positiveLikeReviews': sum(r['positiveLikeReviews'] for r in reports),
        'maxObservedLikes': max((r['maxObservedLikes'] for r in reports if r['maxObservedLikes'] is not None), default=None),
        'batches': [{k: r[k] for k in ['batchId', 'products', 'reviews', 'reviewCounts', 'channelArchiveSha256', 'channelArchiveBytes']} for r in reports],
        'productionImported': False, 'publication': 'prepared_not_published'}
    root.mkdir(parents=True, exist_ok=True)
    save(root / 'verification.json', report)
    save(root / 'collection-audit.json', {**collection, 'records': [{k: v for k, v in r.items() if k != 'records'} for r in collection['records']]})
    print(json.dumps(report, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    main()