kangshifu-board/scripts/recover-product-completion.py

"""Recover exact-SKU attributes and audit review originals without new collection.

Public HTML is retained byte-for-byte. JSON is explicitly a parsed adapter, never
an upstream response. Historical capture times and unknown numeric units survive.
"""
import argparse
from collections import Counter, defaultdict
from datetime import datetime, timezone
import hashlib
import importlib.util
import json
from pathlib import Path
import re
import zipfile

ROOT = Path(__file__).resolve().parents[1]
WORK = ROOT / 'work/product-completion-20261108-r1'
DOCS = ROOT / 'docs/product-completion-20261108'
PLACEHOLDER = re.compile(r'未填写评价|该用户觉得商品|此用户没有填写|系统默认')


def module(name, filename):
    spec = importlib.util.spec_from_file_location(name, ROOT / 'scripts' / filename)
    result = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(result)
    return result


links = module('product_links', 'enrich-product-links.py')
unbound = module('unbound_evidence', 'audit-industry-unbound-evidence.py')


def sha(raw):
    return hashlib.sha256(raw).hexdigest()


def read(path):
    return json.loads(Path(path).read_bytes())


def save(path, value):
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')


def select_spec(raw, expected_id, expected_sha):
    """Only explicit sale-property labels on the exact selected SKU are facts."""
    if sha(raw) != expected_sha:
        raise ValueError('PAGE_HASH_MISMATCH')
    text = raw.decode('utf-8')
    match = re.search(r'window\._itemOnly\s*=\s*\(?\s*', text)
    if not match:
        raise ValueError('NO_LITERAL_PAYLOAD')
    item_only, end = json.JSONDecoder().raw_decode(text[match.end():])
    # A literal must finish at its assignment boundary, not be an expression.
    if not re.match(r'\s*\)?\s*;', text[match.end() + end:]):
        raise ValueError('NON_LITERAL_ASSIGNMENT')
    item = item_only.get('item', {})
    if str(item.get('skuId')) != expected_id:
        raise ValueError('PAGE_PRODUCT_MISMATCH')
    matches = [(i, v) for i, v in enumerate(item.get('newColorSize', []))
               if isinstance(v, dict) and str(v.get('skuId')) == expected_id]
    if len(matches) != 1:
        raise ValueError('SELECTED_SKU_NOT_UNIQUE')
    index, selected = matches[0]
    attributes = []
    for key, label in item.get('saleProp', {}).items():
        value = selected.get(key)
        if isinstance(label, str) and label.strip() and isinstance(value, str) and value.strip():
            attributes.append({'label': label, 'value': value,
                               'labelPointer': '/item/saleProp/' + key,
                               'valuePointer': '/item/newColorSize/' + str(index) + '/' + key})
    if not attributes:
        raise ValueError('NO_EXPLICIT_SELECTED_ATTRIBUTE')
    spec_text = '；'.join(a['label'] + '：' + a['value'] for a in attributes)
    return {'payload': item_only, 'literalCharacterOffset': match.end(),
            'literalCharacterLength': end, 'selectedIndex': index,
            'attributes': attributes, 'specText': spec_text}


def reviews(snapshot):
    """Audit requested/returned identity before any full-key deduplication."""
    existing = {r['record_id'] for r in snapshot['entities']['reviews']}
    products = {r['record_id'] for r in snapshot['entities']['products']}
    seen, unique, counts, excluded, missing = set(), set(), Counter(), [], {}
    roots = [ROOT / 'work/product-reviews',
             ROOT / 'work/industry-review-source-recovery-20261107-r2/verified/private-recovered-envelopes',
             ROOT / 'work/voc-shop-chain-audit-20261106-r1/received-r2/originals',
             ROOT / 'work/existing-voc-mcp-readonly-20261106-r1']
    original = next(Path('F:/文档').glob('*VOC*全部已采集*.zip'))
    archives = [original, *sorted((ROOT / 'work/product-reviews').rglob('staging.source.zip')),
                ROOT / 'work/voc-review-restoration-20261106-r1/batch/staging.source.zip',
                ROOT / 'work/industry-review-source-recovery-20261107-r2/received/original-return.zip']

    def consume(raw, locator):
        counts['jsonFilesRead'] += 1
        hashed = sha(raw)
        if hashed in seen:
            counts['duplicateBytes'] += 1
            return
        seen.add(hashed)
        try:
            doc = json.loads(raw)
        except (ValueError, UnicodeDecodeError):
            counts['nonJson'] += 1
            return
        if not isinstance(doc, dict):
            return
        meta = unbound.metadata(doc)
        endpoint, params = meta['endpointId'], meta['params']
        if endpoint == 'jd.get_item_comments_v1':
            family, rows, pointer = 'jd', meta['records'], meta['recordArrayPointer']
        elif endpoint == 'taobao.get_item_comment_v3':
            family, rows, pointer = 'taobao', None, None
            for p in ['/result/data/data/comments', '/data/data/comments', '/data/result/comments',
                      '/response/structuredContent/data/result/comments']:
                candidate = unbound.at(doc, p)
                if isinstance(candidate, list):
                    rows, pointer = candidate, p
                    break
        else:
            return
        counts['uniqueReviewEnvelopes'] += 1
        if not isinstance(rows, list):
            counts['reviewResponseWithoutRows'] += 1
            return
        request = str(params.get('itemId')) if isinstance(params, dict) and params.get('itemId') is not None else None
        for i, r in enumerate(rows):
            counts['rawReviewRows'] += 1
            if family == 'jd':
                item, rid, content, append = r.get('productId'), r.get('commentId'), r.get('commentData'), r.get('afterComment')
            else:
                item, rid, content = r.get('auction', {}).get('aucNumId'), r.get('rateId'), r.get('content')
                append = r.get('append') or r.get('appendList')
            # Floats, booleans and missing IDs must not become plausible strings.
            if type(item) not in (str, int) or type(rid) not in (str, int) or not str(item).isdigit() or not str(rid).isdigit():
                counts['invalidIdentityRows'] += 1
                continue
            pid = family + ':' + str(item)
            key = pid + ':' + str(rid)
            evidence = {'recordId': key, 'requestedItemId': request, 'returnedItemId': str(item),
                        'rawSha256': hashed, 'locator': locator, 'pointer': pointer + '/' + str(i)}
            if request != str(item):
                counts['identityMismatchRows'] += 1
                if key not in existing:
                    excluded.append({**evidence, 'reason': 'REQUEST_RETURN_IDENTITY_MISMATCH'})
                continue
            if key in unique:
                counts['duplicateFullKeys'] += 1
            else:
                unique.add(key)
            if key in existing:
                counts['alreadyStoredRowOccurrences'] += 1
                continue
            if pid not in products:
                counts['productNotInCurrentLibrary'] += 1
                excluded.append({**evidence, 'reason': 'PRODUCT_NOT_IN_LIBRARY'})
                continue
            placeholder = not isinstance(content, str) or not content.strip() or bool(PLACEHOLDER.search(content))
            # Append content needs review by field, never stringify metadata into prose.
            def meaningful_append(value):
                if isinstance(value, dict):
                    return any(isinstance(value.get(k), str) and value[k].strip() and not PLACEHOLDER.search(value[k])
                               for k in ('content', 'commentData'))
                return isinstance(value, list) and any(meaningful_append(v) for v in value)
            if placeholder and not meaningful_append(append):
                counts['templateOnlyMissingRowOccurrences'] += 1
                excluded.append({**evidence, 'reason': 'PLACEHOLDER_WITHOUT_MEANINGFUL_APPEND'})
                continue
            # Preserve one complete response row, never splice fields from calls.
            candidate = {**evidence, 'content': content, 'append': append}
            prior = missing.get(key)
            if prior and prior['content'] != content:
                raise ValueError('MISSING_REVIEW_CONTENT_CONFLICT:' + key)
            if not prior or meaningful_append(append) and not meaningful_append(prior['append']):
                missing[key] = candidate

    inventory = []
    for archive in archives:
        if not archive.exists():
            continue
        hashed = sha(archive.read_bytes())
        inventory.append({'archive': str(archive), 'sha256': hashed})
        with zipfile.ZipFile(archive) as z:
            for info in z.infolist():
                if info.filename.endswith('.json') and ('/raw/' in info.filename or archive.name == 'original-return.zip') and info.file_size <= 20_000_000:
                    consume(z.read(info), {'archive': str(archive), 'archiveSha256': hashed, 'member': info.filename})
    for folder in roots:
        for path in sorted(folder.rglob('*.json')):
            if path.stat().st_size <= 20_000_000:
                consume(path.read_bytes(), {'file': str(path)})
    counts['meaningfulMissingFullKeys'] = len(missing)
    counts['alreadyStoredFullKeys'] = len(unique & existing)
    counts['templateOnlyMissingFullKeys'] = len({r['recordId'] for r in excluded
        if r['reason'] == 'PLACEHOLDER_WITHOUT_MEANINGFUL_APPEND'} - set(missing))
    result = {'schema': 'ksf.product-review-recovery-audit.v1', 'generation': snapshot['generation'],
              'checkedAt': datetime.now(timezone.utc).isoformat(), 'scope': 'All current product IDs; known JD/Taobao review endpoints in listed bytes',
              'counts': dict(counts), 'archives': inventory, 'filesystemRoots': [str(p) for p in roots],
              'missingMeaningful': list(missing.values()), 'excluded': excluded, 'paidCalls': 0, 'productionWrites': 0}
    save(DOCS / 'review-recovery-audit.json', result)
    print(json.dumps({'reviews': result['counts']}, ensure_ascii=False))


def specs(snapshot):
    products = {r['record_id']: r for r in snapshot['entities']['products']}
    scope = {r['productId'] for r in read(WORK / 'products.json') if r['instantCandidate']}
    existing = {(r['product_id'], r.get('spec_text')) for r in snapshot['entities']['specs']}
    counts, candidates, excluded = Counter(), [], []
    provenance_path = ROOT / 'work/product-media/collected.json'
    provenance_raw = provenance_path.read_bytes()
    for meta in json.loads(provenance_raw)['records']:
        if not meta.get('sourcePageSha256'):
            continue
        counts['pageProvenanceRows'] += 1
        pid = meta['productId']
        if pid not in scope:
            continue
        counts['researchPages'] += 1
        path = ROOT / 'work/product-media' / meta['sourceMember']
        raw = path.read_bytes()
        try:
            selected = select_spec(raw, pid.split(':')[1], meta['sourcePageSha256'])
        except ValueError as error:
            excluded.append({'productId': pid, 'reason': str(error), 'sha256': sha(raw)})
            counts[str(error)] += 1
            continue
        if (pid, selected['specText']) in existing:
            counts['alreadyStoredExactText'] += 1
            continue
        # Archived page extraction captures no weights, prices, stock or dates from title.
        candidates.append({'productId': pid, 'title': products[pid]['title'], 'brand': products[pid].get('brand'),
                           'specText': selected['specText'], 'meta': meta, 'pagePath': str(path),
                           'selected': selected})
    counts['safeProducts'] = len(candidates)
    report = {'schema': 'ksf.product-spec-recovery-plan.v1', 'generation': snapshot['generation'],
              'provenanceInventorySha256': sha(provenance_raw), 'counts': dict(counts),
              'brands': dict(Counter(r['brand'] for r in candidates)),
              'candidates': [{k: v for k, v in r.items() if k not in ('selected', 'meta', 'pagePath')}
                             for r in candidates], 'excluded': excluded, 'numericWeightFieldsInferred': 0,
              'newObservations': False, 'paidCalls': 0, 'productionWrites': 0}
    save(DOCS / 'spec-recovery-plan.json', report)
    return candidates, report


def review_batch(directory, snapshot):
    audit = read(DOCS / 'review-recovery-audit.json')
    assert audit['generation'] == snapshot['generation']
    rows = audit['missingMeaningful']
    if not rows:
        return None
    products = {r['record_id']: r for r in snapshot['entities']['products']}
    ids = sorted({r['recordId'].rsplit(':', 1)[0] for r in rows})
    batch = {'schema': 'ksf.voc-agent-batch.v2', 'batchId': 'product-review-recovery-20261108-r1',
             'producer': 'Codex: byte-verified historical review recovery; no new collection or high-like claim',
             'sources': [], 'attachments': [], 'products': [], 'reviews': []}
    context = {'kind': 'existing_library_context_not_new_observation', 'generation': snapshot['generation'],
               'exportedAt': snapshot['exportedAt'], 'products': [products[pid] for pid in ids]}
    save(directory / 'raw/context.json', context)
    for i, pid in enumerate(ids):
        p = products[pid]
        sid = 'context-' + pid.replace(':', '-')
        batch['sources'].append({'id': sid, 'path': 'raw/context.json', 'sha256': sha((directory / 'raw/context.json').read_bytes()),
                                 'platform': pid.split(':')[0], 'marketplace': p['platform'] if pid.startswith('jd:') else 'tmall' if p['platform'] == 'tmall' else 'taobao_unverified',
                                 'requestItemId': pid.split(':')[1]})
        batch['products'].append({'sourceId': sid, 'pointer': '/products/' + str(i),
                                  'fields': {'itemId': '/item_id', 'title': '/title', 'shopId': '/shop_id', 'shopName': '/shop_name'}})
    originals = {}
    records = []
    for candidate in rows:
        family, item, rid = candidate['recordId'].split(':')
        loc = candidate['locator']
        if 'file' in loc:
            raw = Path(loc['file']).read_bytes()
        else:
            assert sha(Path(loc['archive']).read_bytes()) == loc['archiveSha256']
            with zipfile.ZipFile(loc['archive']) as z:
                raw = z.read(loc['member'])
        assert sha(raw) == candidate['rawSha256']
        if re.search(rb'\bsk-[A-Za-z0-9_-]{20,}|Bearer\s+[A-Za-z0-9._-]{20,}', raw):
            raise ValueError('POSSIBLE_CREDENTIAL_IN_SOURCE')
        doc = json.loads(raw)
        meta = unbound.metadata(doc)
        assert str(meta['params']['itemId']) == item
        original = unbound.at(doc, candidate['pointer'])
        if family == 'jd':
            actual, review_id, content = original['productId'], original['commentId'], original['commentData']
            append, date, rating, sku = original.get('afterComment'), original.get('commentDate'), original.get('commentScore'), original.get('productColor')
            likes = original.get('praiseCnt')
        else:
            actual, review_id, content = original['auction']['aucNumId'], original['rateId'], original['content']
            # Keep both original append fields in the attachment; prefer nonempty
            # list if present, otherwise the actual append object. Do not invent dates.
            append = original.get('appendList') or original.get('append')
            date, rating, sku = original.get('date'), original.get('rate'), original['auction'].get('sku')
            likes = original.get('likeCount')
        assert str(actual) == item and str(review_id) == rid and content == candidate['content']
        rel = 'raw/original/' + candidate['rawSha256'] + '.json'
        originals[rel] = raw
        record = {'itemId': item, 'reviewId': rid, 'content': content, 'appendContent': append,
                  'reviewDate': date, 'rating': rating, 'ratingScale': None, 'purchasedSku': sku,
                  'isDefault': None, 'sourceTime': meta['observedAtRaw'], 'likeCount': likes,
                  'origin': {'path': rel, 'sha256': sha(raw), 'pointer': candidate['pointer'],
                             'endpoint': meta['endpointId'], 'requestParams': meta['params'],
                             'requestIdentities': meta['requestIdentities']},
                  'selection': 'exact_request_return_identity_and_missing_full_key; no_high_like_claim'}
        records.append(record)
    for rel, raw in originals.items():
        (directory / rel).parent.mkdir(parents=True, exist_ok=True)
        (directory / rel).write_bytes(raw)
        batch['attachments'].append({'path': rel, 'sha256': sha(raw)})
    adapter_rel = 'raw/parsed/reviews.json'
    save(directory / adapter_rel, {'kind': 'archived_response_mapping_adapter_not_new_upstream_response',
                                    'generation': snapshot['generation'], 'records': records})
    for i, record in enumerate(records):
        pid = record['origin']['endpoint'].split('.')[0] + ':' + record['itemId']
        p = products[pid]
        sid = 'review-' + str(i)
        batch['sources'].append({'id': sid, 'path': adapter_rel, 'sha256': sha((directory / adapter_rel).read_bytes()),
                                 'platform': pid.split(':')[0], 'marketplace': 'jd' if pid.startswith('jd:') else 'tmall' if p['platform'] == 'tmall' else 'taobao_unverified',
                                 'requestItemId': record['itemId'], 'observedAt': record['sourceTime']})
        batch['reviews'].append({'sourceId': sid, 'pointer': '/records/' + str(i),
                                 'fields': {k: '/' + k for k in ('itemId', 'reviewId', 'content', 'appendContent', 'reviewDate', 'rating', 'ratingScale', 'purchasedSku', 'isDefault', 'sourceTime')}})
    save(directory / 'batch.json', batch)
    return {'directory': str(directory.relative_to(WORK)), 'batchId': batch['batchId'], 'products': len(ids),
            'specs': 0, 'reviews': len(records), 'productIds': ids, 'reviewIds': [r['recordId'] for r in rows],
            'baselineGeneration': snapshot['generation']}


def build(candidates, snapshot):
    folder = WORK / 'intake'
    folder.mkdir(exist_ok=False)
    # 160 specs = 320 sources + 160 original attachments; below the 1,000-file cap.
    # Product contexts come from the fixed generation, not a new product observation.
    batches = []
    snapshot_sha = sha((ROOT / 'work/industry-monitoring-20261107-r1/snapshot.json').read_bytes())
    inventory_sha = sha((ROOT / 'work/product-media/collected.json').read_bytes())
    products = {r['record_id']: r for r in snapshot['entities']['products']}
    for offset in range(0, len(candidates), 160):
        group = candidates[offset:offset + 160]
        number = offset // 160 + 1
        directory = folder / ('batch-%02d' % number)
        batch = {'schema': 'ksf.voc-agent-batch.v2', 'batchId': 'product-spec-recovery-20261108-r1-%02d' % number,
                 'producer': 'Codex: exact-SKU archived public attribute recovery; historical evidence, no new collection',
                 'sources': [], 'attachments': [], 'products': [], 'specs': []}
        context = {'kind': 'existing_library_context_not_new_observation', 'generation': snapshot['generation'],
                   'exportedAt': snapshot['exportedAt'], 'snapshotSha256': snapshot_sha,
                   'products': [products[r['productId']] for r in group]}
        save(directory / 'raw/context.json', context)
        for index, candidate in enumerate(group):
            pid = candidate['productId']
            iid = pid.split(':')[1]
            meta, selected = candidate['meta'], candidate['selected']
            name = 'jd-' + iid
            html_rel = 'raw/original/' + name + '.html'
            adapter_rel = 'raw/parsed/' + name + '.json'
            raw = Path(candidate['pagePath']).read_bytes()
            assert sha(raw) == meta['sourcePageSha256']
            (directory / html_rel).parent.mkdir(parents=True, exist_ok=True)
            (directory / html_rel).write_bytes(raw)
            basis = ('京东商品页的当前SKU销售属性原文；仅核规格文字，不核净含量/件数/总重。'
                     '原页SHA256=' + sha(raw) + '；SKU=' + iid)
            adapter = {'kind': 'parsed_public_html_adapter_not_upstream_json',
                       'original': {'path': html_rel, 'sha256': sha(raw), 'url': meta['sourcePageUrl'],
                                    'observedAt': meta['observedAt'], 'collectedAt': meta['collectedAt']},
                       'provenance': {'inventory': 'work/product-media/collected.json',
                                      'inventorySha256': inventory_sha,
                                      'record': meta},
                       'extraction': {k: v for k, v in selected.items() if k not in ('payload', 'specText')},
                       'literal': selected['payload'],
                       'spec': {'itemId': iid, 'skuId': iid, 'specText': selected['specText'],
                                'verification': 'source_attribute', 'basis': basis}}
            save(directory / adapter_rel, adapter)
            product = products[pid]
            context_id, source_id = 'context-' + name, 'page-' + name
            batch['sources'].append({'id': context_id, 'path': 'raw/context.json', 'sha256': sha((directory / 'raw/context.json').read_bytes()),
                                     'platform': 'jd', 'marketplace': 'jd', 'requestItemId': iid})
            batch['products'].append({'sourceId': context_id, 'pointer': '/products/' + str(index),
                                      'fields': {'itemId': '/item_id', 'title': '/title', 'shopId': '/shop_id', 'shopName': '/shop_name'}})
            batch['sources'].append({'id': source_id, 'path': adapter_rel, 'sha256': sha((directory / adapter_rel).read_bytes()),
                                     'platform': 'jd', 'marketplace': 'jd', 'requestItemId': iid,
                                     'observedAt': meta['observedAt'], 'collectedAt': meta['collectedAt'], 'url': meta['sourcePageUrl']})
            batch['attachments'].append({'path': html_rel, 'sha256': sha(raw)})
            batch['specs'].append({'sourceId': source_id, 'pointer': '/spec',
                                   'fields': {k: '/' + k for k in adapter['spec']}})
        save(directory / 'batch.json', batch)
        rows = {'directory': str(directory.relative_to(WORK)), 'batchId': batch['batchId'], 'products': len(group), 'specs': len(group),
                'productIds': [r['productId'] for r in group], 'baselineGeneration': snapshot['generation'],
                'originalBytes': sum((directory / a['path']).stat().st_size for a in batch['attachments'])}
        batches.append(rows)
    recovered = review_batch(folder / ('batch-%02d' % (len(batches) + 1)), snapshot)
    if recovered:
        batches.append(recovered)
    save(WORK / 'batch-plan.json', {'generation': snapshot['generation'], 'batches': batches, 'paidCalls': 0, 'productionWrites': 0})
    print(json.dumps({'specsPrepared': len(candidates), 'batches': len(batches)}, ensure_ascii=False))


def correct_probe():
    # Derived correction only; retain original per-request receipts unchanged.
    report = read(DOCS / 'public-probe.json')
    original_sha = sha((DOCS / 'public-probe.json').read_bytes())
    for row in report['results']:
        if 'risk_handler' in (row.get('finalUrl') or '') or '验证' in (row.get('pageTitle') or ''):
            row['priorState'] = row['state']
            row['state'] = 'LOGIN_OR_ACCESS_CHALLENGE'
        elif row.get('error') == 'AssertionError' and (row.get('storedUrl') or '').startswith('https://sh.jd.com/') and not row.get('requestUrl'):
            row['priorState'] = row['state']
            row['state'] = 'SKIPPED_SHORT_LINK_BY_PROBE_VALIDATOR'
            row['externalRequestIssued'] = False
            row['explanation'] = '原探测器的域名白名单遗漏京东短链；本次未向此商品页面发外部请求，不能判为平台不可读。'
    report['states'] = dict(Counter(r['state'] for r in report['results']))
    report['derivedFromSha256'] = original_sha
    report['newNetworkRequests'] = 0
    save(DOCS / 'public-probe-corrected.json', report)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('command', choices=['scan', 'build'])
    args = parser.parse_args()
    snapshot = read(ROOT / 'work/industry-monitoring-20261107-r1/snapshot.json')
    assert snapshot['generation'] == read(WORK / 'summary-before.json')['generation']
    if args.command == 'scan':
        correct_probe()
        reviews(snapshot)
        _, result = specs(snapshot)
        print(json.dumps({'specs': result['counts']}, ensure_ascii=False))
    else:
        candidates, _ = specs(snapshot)
        build(candidates, snapshot)


if __name__ == '__main__':
    main()