kangshifu-board/scripts/import-product-images.py

"""Recover exact product images from original archives, collect bytes, then publish.

The frozen VOC SQLite schema is unchanged. KsfProductImage is an additive Parse
collection containing actual JPEG bytes (base64), thumbnails and source evidence.
No paid collection API is used. Credentials are read only by the publish command.
"""
import argparse
import base64
import concurrent.futures
import datetime as dt
import hashlib
import io
import json
import re
import time
import urllib.error
import urllib.parse
import urllib.request
import zipfile
from collections import Counter
from pathlib import Path, PurePosixPath

from PIL import Image, ImageOps

CLASS = 'KsfProductImage'
VERSION = 'ksf.product-images.v1'
IMAGE_FIELDS = {'imageUrl', 'imageurl', 'itemPic', 'picUrlFull', 'pict_url', 'pic_url', 'mainPic', 'mainImage', 'picUrl', 'images'}
ID_FIELDS = ('itemId', 'item_id', 'skuId', 'sku_id', 'num_iid', 'itemNumId', 'id')
MAX_BYTES = 12 * 1024 * 1024
Image.MAX_IMAGE_PIXELS = 40_000_000


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


def write(path, data):
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + '.tmp')
    tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    tmp.replace(path)


def request(url, body=None, headers=None, method=None):
    encoded = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(url, data=encoded, headers={'Content-Type': 'application/json', **(headers or {})}, method=method)
    with urllib.request.urlopen(req, timeout=35) as response:
        return json.load(response)


def config(path):
    values = {m[1].strip(): m[2].strip() for m in re.finditer(r'^\|\s*([^|]+)\s*\|\s*([^|]+)\s*\|', Path(path).read_text(encoding='utf-8'), re.M)}
    return values['Parse 端点'].rstrip('/'), {'X-Parse-Application-Id': values['appId'], 'X-Parse-Master-Key': values['masterKey']}


def image_url(value, family):
    if not isinstance(value, str):
        return None
    value = value.strip()
    if value.startswith('jfs/') and family == 'jd':
        value = 'https://img14.360buyimg.com/n1/' + value
    elif value.startswith('//'):
        value = 'https:' + value
    parsed = urllib.parse.urlsplit(value)
    host = (parsed.hostname or '').lower()
    allowed = ('360buyimg.com',) if family == 'jd' else ('alicdn.com', 'tbcdn.cn')
    if parsed.scheme not in ('http', 'https') or parsed.username or parsed.password or parsed.port not in (None, 80, 443):
        return None
    if not any(host == domain or host.endswith('.' + domain) for domain in allowed):
        return None
    if not parsed.path or any(part in parsed.path.lower() for part in ('avatar', 'default.image', 'shaidan')):
        return None
    return urllib.parse.urlunsplit(('https', parsed.netloc, parsed.path, parsed.query, ''))


def candidates(payload, known_ids):
    """Only product endpoints; identity comes from the same object or detail request."""
    endpoint = str(payload.get('id') or payload.get('endpoint') or '')
    family = endpoint.split('.')[0]
    if family not in ('jd', 'taobao') or not any(s in endpoint for s in ('search_item_list', 'get_shop_item_list', 'get_item_detail')):
        return []
    detail = 'get_item_detail' in endpoint
    params = payload.get('params') or {}
    fallback = str(params.get('itemId') or params.get('skuId') or '') if detail else ''
    timestamp = payload.get('sourceTime') or payload.get('collected_at') or payload.get('finishedAt') or ''
    found = []

    def walk(node, identity='', path='result'):
        if isinstance(node, list):
            for index, child in enumerate(node):
                walk(child, identity, path + '[' + str(index) + ']')
        elif isinstance(node, dict):
            own = next((str(node[k]) for k in ID_FIELDS if k in node and re.fullmatch(r'\d+', str(node[k]))), '')
            current = own or identity
            pid = family + ':' + current
            product_node = own or detail and path.endswith(('.product', '.item', '.itemInfoModel'))
            if pid in known_ids and product_node:
                for key in IMAGE_FIELDS:
                    value = node.get(key)
                    values = value if isinstance(value, list) else [value]
                    for index, raw in enumerate(values[:8]):
                        if isinstance(raw, dict):
                            raw = raw.get('url') or raw.get('imageUrl')
                        url = image_url(raw, family)
                        if url:
                            found.append({'productId': pid, 'sourceUrl': url, 'sourceField': path + '.' + key + (f'[{index}]' if isinstance(value, list) else ''),
                                          'observedAt': timestamp, 'priority': 3 if detail and current == fallback else 2,
                                          'endpoint': endpoint})
            for key, child in node.items():
                # Related products, SKU variants, reviews and seller logos are not the main image.
                if key.lower() in ('recommend', 'recommendations', 'recommendlist', 'comments', 'reviews', 'seller', 'shop', 'skulist', 'sku2info'):
                    continue
                if isinstance(child, (list, dict)) and key not in IMAGE_FIELDS:
                    walk(child, current, path + '.' + key)

    walk(payload.get('result', {}), fallback)
    return found


def catalog(url, out):
    summary = request(url, {'op': 'summary'})
    if not summary.get('ok'):
        raise RuntimeError('CATALOG_SUMMARY_FAILED')
    rows, cursor = [], None
    while True:
        query = {'op': 'browse', 'entity': 'products', 'quality': 'audit', 'limit': 100, 'generation': summary['generation']}
        if cursor:
            query['cursor'] = cursor
        page = request(url, query)
        if not page.get('ok') or page.get('generation') != summary['generation']:
            raise RuntimeError('CATALOG_VERSION_CHANGED')
        rows.extend(page['records'])
        cursor = page.get('nextCursor')
        if not cursor:
            break
    if len({r['record_id'] for r in rows}) != summary['counts']['products']:
        raise RuntimeError('CATALOG_COUNT_MISMATCH')
    write(out / 'catalog.json', {'generation': summary['generation'], 'records': rows, 'url': url})
    return rows


def plan(args):
    out = Path(args.out).resolve()
    rows = catalog(args.catalog_url, out)
    known = {r['record_id']: r for r in rows}
    choices, cached = {}, {}
    archive_path = Path(args.archive).resolve()
    with archive_path.open('rb') as stream:
        archive_sha = hashlib.file_digest(stream, 'sha256').hexdigest()
    with zipfile.ZipFile(archive_path) as archive:
        names = set(archive.namelist())
        # Complete this map before resolving candidates, including normalized JD paths.
        for name in sorted(names):
            if name.endswith('/media-manifest.json'):
                for item in json.loads(archive.read(name)).get('items', []):
                    if item.get('localPath') and item.get('cacheStatus') == 'downloaded':
                        path = str(PurePosixPath(name).parent / item['localPath'])
                        if path in names:
                            source = item.get('sourceUrl') or ''
                            family = 'jd' if '360buyimg.com' in source else 'taobao'
                            normalized = image_url(source, family)
                            if normalized:
                                cached[normalized] = path
                                cached[urllib.parse.urlsplit(normalized).path.split('/jfs/')[-1]] = path
        for name in sorted(names):
            if not any(part.startswith('raw') for part in PurePosixPath(name).parts) or not name.endswith('.json'):
                continue
            try:
                payload = json.loads(archive.read(name))
            except (ValueError, UnicodeError):
                continue
            if not isinstance(payload, dict):
                continue
            for item in candidates(payload, known):
                item.update(sourceArchive=archive_sha, sourceMember=name)
                choices.setdefault(item['productId'], []).append(item)
    planned = []
    for pid, options in sorted(choices.items()):
        unique = {}
        for item in sorted(options, key=lambda i: (i['priority'], i['observedAt']), reverse=True):
            unique.setdefault(item['sourceUrl'], item)
        options = list(unique.values())[:8]
        for item in options:
            item['cachedMember'] = cached.get(item['sourceUrl']) or cached.get(urllib.parse.urlsplit(item['sourceUrl']).path.split('/jfs/')[-1])
        row = known[pid]
        planned.append({'productId': pid, 'title': row.get('title'), 'platform': row.get('platform'), 'candidates': options})
    result = {'schema': VERSION, 'archive': str(archive_path), 'archiveSha256': archive_sha, 'products': len(known),
              'matchedProducts': len(planned), 'missingIds': sorted(set(known) - set(choices)), 'records': planned}
    write(out / 'plan.json', result)
    print(json.dumps({'products': len(known), 'matchedProducts': len(planned), 'missingProducts': len(result['missingIds'])}), flush=True)


class CdnRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        family = 'jd' if '360buyimg.com' in (urllib.parse.urlsplit(req.full_url).hostname or '') else 'taobao'
        if not image_url(newurl, family):
            raise ValueError('UNSAFE_IMAGE_REDIRECT')
        return super().redirect_request(req, fp, code, msg, headers, newurl)


def image_bytes(raw, edge):
    if not 0 < len(raw) <= MAX_BYTES:
        raise ValueError('INVALID_IMAGE_SIZE')
    with Image.open(io.BytesIO(raw)) as source:
        source.load()
        if min(source.size) < 120:
            raise ValueError('IMAGE_TOO_SMALL')
        image = ImageOps.exif_transpose(source).convert('RGB')
        image.thumbnail((edge, edge), Image.Resampling.LANCZOS)
        output = io.BytesIO()
        image.save(output, 'JPEG', quality=85 if edge > 300 else 80, optimize=True)
        return output.getvalue(), image.width, image.height


def page_images(html, pid):
    """Parse JSON literals; never execute the merchant's JavaScript."""
    if not pid.startswith('jd:'):
        return []
    match = re.search(r'window\._itemOnly\s*=\s*\(?\s*', html)
    if not match:
        return []
    try:
        data, _ = json.JSONDecoder().raw_decode(html[match.end():])
    except ValueError:
        return []
    item = data.get('item') or {}
    if str(item.get('skuId')) != pid.split(':', 1)[1]:
        return []
    return [url for raw in (item.get('image') or [])[:4] if (url := image_url(raw, 'jd'))]


def discover(args):
    out = Path(args.out).resolve()
    data = json.loads((out / 'plan.json').read_text(encoding='utf-8'))
    known = {r['productId'] for r in data['records']}
    records = {r['record_id']: r for r in json.loads((out / 'catalog.json').read_text(encoding='utf-8'))['records']}
    missing = [pid for pid in data['missingIds'] if pid not in known][:args.limit or None]
    errors = []

    def one(pid):
        if not pid.startswith('jd:'):
            return None, {'productId': pid, 'reason': 'NO_SAFE_PUBLIC_PAGE_ADAPTER'}
        url = 'https://item.m.jd.com/product/' + pid.split(':', 1)[1] + '.html'
        try:
            req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
            with urllib.request.urlopen(req, timeout=12) as response:
                if urllib.parse.urlsplit(response.url).hostname not in ('item.m.jd.com', 'item.jd.com'):
                    return None, {'productId': pid, 'reason': 'PAGE_REDIRECTED'}
                raw = response.read(3 * 1024 * 1024)
            urls = page_images(raw.decode('utf-8', 'replace'), pid)
            if not urls:
                return None, {'productId': pid, 'reason': 'NO_VERIFIED_IMAGE_ON_PAGE'}
            page_path = 'pages/' + pid.replace(':', '-') + '.html'
            target = out / page_path
            target.parent.mkdir(exist_ok=True)
            target.write_bytes(raw)
            stamp = dt.datetime.now(dt.timezone.utc).isoformat()
            images = [{'productId': pid, 'sourceUrl': u, 'sourcePageUrl': url, 'sourcePageSha256': sha(raw),
                       'sourceMember': page_path, 'sourceField': 'window._itemOnly.item.image[' + str(i) + ']',
                       'observedAt': stamp, 'priority': 1, 'endpoint': 'public_product_page', 'cachedMember': None}
                      for i, u in enumerate(urls)]
            return {'productId': pid, 'title': records[pid].get('title'), 'platform': records[pid].get('platform'), 'candidates': images}, None
        except Exception as error:
            return None, {'productId': pid, 'reason': type(error).__name__}
        finally:
            time.sleep(.15)

    with concurrent.futures.ThreadPoolExecutor(max_workers=min(args.workers, 4)) as pool:
        for index, (record, error) in enumerate(pool.map(one, missing), 1):
            if record:
                data['records'].append(record)
                known.add(record['productId'])
            if error:
                errors.append(error)
            if index % 100 == 0:
                data['missingIds'] = sorted(set(records) - known)
                data['matchedProducts'] = len(known)
                write(out / 'plan.json', data)
                write(out / 'discovery.json', {'errors': errors, 'discovered': index - len(errors)})
                print(json.dumps({'processed': index, 'discovered': index - len(errors)}), flush=True)
    data['missingIds'] = sorted(set(records) - known)
    data['matchedProducts'] = len(known)
    write(out / 'plan.json', data)
    write(out / 'discovery.json', {'errors': errors, 'discovered': len(missing) - len(errors), 'paidApiCalls': 0})
    print(json.dumps({'discovered': len(missing) - len(errors), 'remaining': len(data['missingIds'])}), flush=True)


def supplement(args):
    out = Path(args.out).resolve()
    data = json.loads((out / 'plan.json').read_text(encoding='utf-8'))
    catalog_rows = {r['record_id']: r for r in json.loads((out / 'catalog.json').read_text(encoding='utf-8'))['records']}
    known = {r['productId'] for r in data['records']}
    archive_path = Path(args.archive).resolve()
    with archive_path.open('rb') as stream:
        archive_sha = hashlib.file_digest(stream, 'sha256').hexdigest()
    added = {}
    with zipfile.ZipFile(archive_path) as archive:
        names = archive.namelist()
        batch_name = next((n for n in names if n.endswith('/batch.json')), None)
        if not batch_name:
            for name in names:
                if not name.endswith('.json') or not any(part.startswith('raw') for part in PurePosixPath(name).parts):
                    continue
                try:
                    payload = json.loads(archive.read(name))
                except (ValueError, UnicodeError):
                    continue
                if not isinstance(payload, dict):
                    continue
                for item in candidates(payload, catalog_rows):
                    if item['productId'] not in known:
                        item.update(sourceArchive=archive_sha, sourceMember=name, cachedMember=None)
                        added.setdefault(item['productId'], []).append(item)
            batch = {'sources': []}
        else:
            batch = json.loads(archive.read(batch_name))
        for source in batch['sources']:
            matches = [n for n in names if n.endswith('/' + source['path'])]
            if len(matches) != 1:
                continue
            raw = archive.read(matches[0])
            if sha(raw) != source['sha256']:
                raise RuntimeError('SUPPLEMENT_SOURCE_HASH_MISMATCH')
            normalized = json.loads(raw)
            original = normalized.get('original') or {}
            originals = [n for n in names if n.endswith('/' + original.get('path', '!'))]
            if len(originals) != 1:
                continue
            body = archive.read(originals[0])
            if sha(body) != original['sha256']:
                raise RuntimeError('SUPPLEMENT_ORIGINAL_HASH_MISMATCH')
            # The batch declares the platform; exact IDs are still required in each object.
            family = source['platform']
            wrapper = {'id': family + '.search_item_list', 'result': json.loads(body), 'sourceTime': source.get('observedAt', '')}
            for item in candidates(wrapper, catalog_rows):
                pid = item['productId']
                if pid in known:
                    continue
                item.update(sourceArchive=archive_sha, sourceMember=originals[0], sourceResponseSha256=original['sha256'],
                            endpoint='archived_product_response', cachedMember=None)
                added.setdefault(pid, []).append(item)
    for pid, options in sorted(added.items()):
        row = catalog_rows[pid]
        data['records'].append({'productId': pid, 'title': row.get('title'), 'platform': row.get('platform'), 'candidates': options[:8]})
        known.add(pid)
    data['missingIds'] = sorted(set(catalog_rows) - known)
    data['matchedProducts'] = len(known)
    write(out / 'plan.json', data)
    print(json.dumps({'added': len(added), 'remaining': len(data['missingIds'])}), flush=True)


def collect(args):
    out = Path(args.out).resolve()
    plan_data = json.loads((out / 'plan.json').read_text(encoding='utf-8'))
    records = plan_data['records'][:args.limit or None]
    result_path = out / 'collected.json'
    previous = json.loads(result_path.read_text(encoding='utf-8')) if result_path.exists() else {'records': [], 'failures': []}
    saved = {r['productId']: r for r in previous['records']}
    failures = []
    opener = urllib.request.build_opener(CdnRedirect())

    def one(row):
        if row['productId'] in saved and (out / saved[row['productId']]['imagePath']).is_file():
            return saved[row['productId']], None
        attempts = []
        with zipfile.ZipFile(plan_data['archive']) as archive:
            for item in row['candidates']:
                try:
                    if item.get('cachedMember'):
                        raw = archive.read(item['cachedMember'])
                        method = 'archive'
                    else:
                        download_url = item['sourceUrl']
                        parsed = urllib.parse.urlsplit(download_url)
                        # Old search subdomains present an invalid host certificate today.
                        # The identical asset path is also served by Alibaba's image host.
                        if re.fullmatch(r'g\.search\d*\.alicdn\.com', parsed.hostname or ''):
                            download_url = urllib.parse.urlunsplit(('https', 'img.alicdn.com', parsed.path, parsed.query, ''))
                        req = urllib.request.Request(download_url, headers={'User-Agent': 'Mozilla/5.0', 'Accept': 'image/*'})
                        with opener.open(req, timeout=12) as response:
                            if not response.headers.get('Content-Type', '').startswith('image/'):
                                raise ValueError('NOT_AN_IMAGE')
                            raw = response.read(MAX_BYTES + 1)
                        method = 'cdn'
                    image, width, height = image_bytes(raw, 800)
                    thumbnail, tw, th = image_bytes(raw, 256)
                    digest = sha(image)
                    image_path = 'images/' + digest + '.jpg'
                    thumbnail_path = 'thumbnails/' + digest + '.jpg'
                    for path, body in ((image_path, image), (thumbnail_path, thumbnail)):
                        target = out / path
                        target.parent.mkdir(parents=True, exist_ok=True)
                        target.write_bytes(body)
                    return {**{k: v for k, v in row.items() if k != 'candidates'}, **item, 'method': method,
                            'downloadUrl': item['sourceUrl'] if method == 'archive' else download_url,
                            'imagePath': image_path, 'thumbnailPath': thumbnail_path, 'sha256': digest,
                            'sourceSha256': sha(raw), 'thumbnailSha256': sha(thumbnail), 'sizeBytes': len(image),
                            'width': width, 'height': height, 'thumbnailWidth': tw, 'thumbnailHeight': th,
                            'collectedAt': dt.datetime.now(dt.timezone.utc).isoformat()}, None
                except Exception as error:
                    attempts.append({'sourceUrl': item['sourceUrl'], 'error': type(error).__name__})
        return None, {'productId': row['productId'], 'attempts': attempts}

    with concurrent.futures.ThreadPoolExecutor(max_workers=args.workers) as pool:
        for index, (record, failure) in enumerate(pool.map(one, records), 1):
            if record:
                saved[record['productId']] = record
            if failure:
                failures.append(failure)
            if index % 100 == 0:
                write(result_path, {'schema': VERSION, 'records': list(saved.values()), 'failures': failures})
                print(json.dumps({'processed': index, 'stored': len(saved), 'failed': len(failures)}), flush=True)
    result = {'schema': VERSION, 'records': list(saved.values()), 'failures': failures,
              'summary': {'products': plan_data['products'], 'stored': len(saved), 'failed': len(failures),
                          'noArchivedUrl': len(plan_data['missingIds']), 'methods': dict(Counter(r['method'] for r in saved.values())),
                          'paidApiCalls': 0}}
    write(result_path, result)
    print(json.dumps(result['summary']), flush=True)


def publish(args):
    out = Path(args.out).resolve()
    data = json.loads((out / 'collected.json').read_text(encoding='utf-8'))
    base, headers = config(args.config)
    schema = {'className': CLASS, 'fields': {k: {'type': t} for k, t in {
        'productId': 'String', 'title': 'String', 'platform': 'String', 'sourceUrl': 'String',
        'imageBase64': 'String', 'thumbnailBase64': 'String', 'mimeType': 'String',
        'sha256': 'String', 'thumbnailSha256': 'String', 'sourceSha256': 'String',
        'width': 'Number', 'height': 'Number', 'sizeBytes': 'Number', 'provenance': 'Object', 'schemaVersion': 'String'}.items()},
        'classLevelPermissions': {'find': {'*': True}, 'get': {'*': True}, 'count': {'*': True},
                                  'create': {}, 'update': {}, 'delete': {}, 'addField': {}, 'protectedFields': {'*': ['provenance']}}}
    try:
        existing_schema = request(base + '/schemas/' + CLASS, headers=headers)
        if any(existing_schema.get('fields', {}).get(k, {}).get('type') != v['type'] for k, v in schema['fields'].items()):
            raise RuntimeError('EXISTING_IMAGE_SCHEMA_MISMATCH')
        permissions = existing_schema.get('classLevelPermissions', {})
        if any(permissions.get(action, {}).get('*') for action in ('create', 'update', 'delete', 'addField')):
            raise RuntimeError('EXISTING_IMAGE_WRITE_PERMISSIONS_UNSAFE')
    except urllib.error.HTTPError as error:
        if error.code not in (400, 404) or json.loads(error.read()).get('code') not in (103, 101):
            raise
        request(base + '/schemas/' + CLASS, schema, headers, 'POST')
    existing, skip = {}, 0
    while True:
        query = urllib.parse.urlencode({'limit': 1000, 'skip': skip, 'keys': 'productId,sha256,thumbnailSha256', 'order': 'objectId'})
        page = request(base + '/classes/' + CLASS + '?' + query, headers=headers)['results']
        for row in page:
            if row['productId'] in existing:
                raise RuntimeError('DUPLICATE_PRODUCT_IMAGE')
            existing[row['productId']] = row
        if len(page) < 1000:
            break
        skip += len(page)
    receipt_path = out / 'published.json'
    receipt = json.loads(receipt_path.read_text(encoding='utf-8')) if receipt_path.exists() else {'schema': VERSION, 'records': []}
    receipts = {r['productId']: r for r in receipt['records']}
    created = unchanged = 0
    pending = []
    selected = data['records'][:args.limit or None]
    for row in selected:
        pid = row['productId']
        old = existing.get(pid)
        if old:
            if old.get('sha256') != row['sha256'] or old.get('thumbnailSha256') != row['thumbnailSha256']:
                raise RuntimeError('EXISTING_IMAGE_DIFFERS:' + pid)
            unchanged += 1
            receipts[pid] = {'productId': pid, 'objectId': old['objectId'], 'sha256': row['sha256']}
        else:
            image, thumbnail = (out / row['imagePath']).read_bytes(), (out / row['thumbnailPath']).read_bytes()
            if sha(image) != row['sha256'] or sha(thumbnail) != row['thumbnailSha256']:
                raise RuntimeError('LOCAL_IMAGE_CHANGED:' + pid)
            body = {k: row[k] for k in ('productId', 'title', 'platform', 'sourceUrl', 'sha256', 'thumbnailSha256', 'sourceSha256', 'width', 'height', 'sizeBytes')}
            body.update(imageBase64=base64.b64encode(image).decode(), thumbnailBase64=base64.b64encode(thumbnail).decode(), mimeType='image/jpeg', schemaVersion=VERSION,
                        provenance={k: row.get(k) for k in ('sourceArchive', 'sourceMember', 'sourceField', 'sourcePageUrl', 'sourcePageSha256', 'downloadUrl', 'endpoint', 'observedAt', 'collectedAt', 'method')}, ACL={'*': {'read': True}})
            pending.append((row, body))
    def send_chunk(chunk):
        prefix = urllib.parse.urlsplit(base).path.rstrip('/')
        response = request(base + '/batch', {'requests': [{'method': 'POST', 'path': prefix + '/classes/' + CLASS, 'body': body} for row, body in chunk]}, headers, 'POST')
        responses = response if isinstance(response, list) else response.get('results', [])
        if len(responses) != len(chunk):
            raise RuntimeError('BATCH_RESULT_UNCERTAIN_RESTART_TO_RECONCILE')
        successful, batch_receipts = [], []
        for (row, body), saved_row in zip(chunk, responses):
            if not saved_row.get('success', {}).get('objectId'):
                raise RuntimeError('BATCH_RESULT_PARTIAL_RESTART_TO_RECONCILE')
            oid = saved_row['success']['objectId']
            successful.append(oid)
            batch_receipts.append({'productId': row['productId'], 'objectId': oid, 'sha256': row['sha256']})
        query = urllib.parse.urlencode({'where': json.dumps({'objectId': {'$in': successful}}),
                                       'keys': 'productId,imageBase64,thumbnailBase64', 'limit': 20})
        checked = {r['productId']: r for r in request(base + '/classes/' + CLASS + '?' + query, headers=headers)['results']}
        for row, body in chunk:
            check = checked.get(row['productId']) or {}
            if sha(base64.b64decode(check.get('imageBase64', ''))) != row['sha256'] or sha(base64.b64decode(check.get('thumbnailBase64', ''))) != row['thumbnailSha256']:
                raise RuntimeError('REMOTE_READBACK_MISMATCH:' + row['productId'])
        return batch_receipts
    chunks = [pending[offset:offset + 20] for offset in range(0, len(pending), 20)]
    with concurrent.futures.ThreadPoolExecutor(max_workers=min(args.workers, 3)) as pool:
        for batch_receipts in pool.map(send_chunk, chunks):
            for record in batch_receipts:
                receipts[record['productId']] = record
            created += len(batch_receipts)
            write(receipt_path, {'schema': VERSION, 'records': list(receipts.values()), 'createdThisRun': created, 'unchangedThisRun': unchanged})
            print(json.dumps({'processed': unchanged + created, 'created': created, 'unchanged': unchanged}), flush=True)
    result = {'schema': VERSION, 'className': CLASS, 'records': list(receipts.values()), 'createdThisRun': created, 'unchangedThisRun': unchanged,
              'completedAt': dt.datetime.now(dt.timezone.utc).isoformat(), 'paidApiCalls': 0}
    write(receipt_path, result)
    print(json.dumps({'published': len(receipts), 'created': created, 'unchanged': unchanged}), flush=True)


def verify(args):
    out = Path(args.out).resolve()
    local = json.loads((out / 'collected.json').read_text(encoding='utf-8'))
    catalog_data = json.loads((out / 'catalog.json').read_text(encoding='utf-8'))
    expected = {row['productId']: row for row in local['records']}
    ids = {row['record_id'] for row in catalog_data['records']}
    if set(expected) != ids:
        raise RuntimeError('IMAGE_CATALOG_COVERAGE_INCOMPLETE')
    base, headers = config(args.config)
    remote, duplicates, skip = {}, [], 0
    while True:
        query = urllib.parse.urlencode({'limit': 1000, 'skip': skip, 'keys': 'productId,sha256,thumbnailSha256,width,height,sizeBytes', 'order': 'objectId'})
        rows = request(base + '/classes/' + CLASS + '?' + query, headers=headers)['results']
        for row in rows:
            if row['productId'] in remote:
                duplicates.append(row['productId'])
            remote[row['productId']] = row
        if len(rows) < 1000:
            break
        skip += len(rows)
    if duplicates or set(remote) != ids:
        raise RuntimeError('REMOTE_IMAGE_COVERAGE_MISMATCH')
    for pid, row in remote.items():
        source = expected[pid]
        if any(source[k] != row.get(k) for k in ('sha256', 'thumbnailSha256', 'width', 'height', 'sizeBytes')):
            raise RuntimeError('REMOTE_METADATA_MISMATCH:' + pid)
    # Anonymous application reads match the board's public product browsing surface.
    public_headers = {'X-Parse-Application-Id': headers['X-Parse-Application-Id']}
    samples = []
    selected = ['jd:100000283767', 'jd:100005702440', 'jd:2848752', 'taobao:1040511334739', sorted(ids)[-1]]
    for pid in selected:
        row = request(base + '/classes/' + CLASS + '/' + remote[pid]['objectId'], headers=public_headers)
        if sha(base64.b64decode(row['imageBase64'])) != expected[pid]['sha256'] or sha(base64.b64decode(row['thumbnailBase64'])) != expected[pid]['thumbnailSha256']:
            raise RuntimeError('PUBLIC_IMAGE_BYTES_MISMATCH:' + pid)
        if row.get('provenance'):
            raise RuntimeError('PUBLIC_PROVENANCE_NOT_PROTECTED')
        samples.append({'productId': pid, 'objectId': row['objectId'], 'bytesVerified': True})
    schema = request(base + '/schemas/' + CLASS, headers=headers)
    permissions = schema['classLevelPermissions']
    if any(permissions.get(action, {}).get('*') for action in ('create', 'update', 'delete', 'addField')):
        raise RuntimeError('PUBLIC_WRITE_ACCESS')
    before = json.loads((out / 'production-summary.json').read_text(encoding='utf-8'))
    current = request(args.catalog_url, {'op': 'summary'})
    if current['generation'] != before['generation'] or current['counts'] != before['counts']:
        raise RuntimeError('VOC_BASELINE_CHANGED_REVIEW_REQUIRED')
    report = {'ok': True, 'schema': VERSION, 'className': CLASS, 'verifiedAt': dt.datetime.now(dt.timezone.utc).isoformat(),
              'catalogProducts': len(ids), 'databaseImages': len(remote), 'coveragePercent': 100,
              'duplicates': 0, 'metadataHashesMatched': len(remote), 'newImageBytesRereadDuringPublish': True,
              'publicByteSamples': samples, 'publicWritesAllowed': False, 'publicSourcePathsHidden': True,
              'vocGenerationUnchanged': True, 'vocCountsUnchanged': True, 'generation': current['generation'],
              'counts': current['counts'], 'summary': local['summary'], 'totalImageBytes': sum(r['sizeBytes'] for r in local['records']),
              'paidApiCalls': 0}
    write(out / 'verification.json', report)
    print(json.dumps(report, ensure_ascii=False, indent=2), flush=True)


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('command', choices=['plan', 'discover', 'supplement', 'collect', 'publish', 'verify'])
    parser.add_argument('--out', default='work/product-media')
    parser.add_argument('--archive')
    parser.add_argument('--catalog-url', default='https://kangshifu.fmode.cn/api/library')
    parser.add_argument('--config')
    parser.add_argument('--limit', type=int, default=0)
    parser.add_argument('--workers', type=int, default=5)
    args = parser.parse_args()
    if args.command in ('plan', 'supplement') and not args.archive:
        parser.error(args.command + ' requires --archive')
    if args.command in ('publish', 'verify') and not args.config:
        parser.error(args.command + ' requires --config')
    try:
        globals()[args.command](args)
    except Exception as error:
        # Never log credential headers or response bodies.
        print(json.dumps({'ok': False, 'error': type(error).__name__, 'reason': str(error) if isinstance(error, RuntimeError) else None}), flush=True)
        raise SystemExit(1)