kangshifu-board/scripts/audit-enrichment-coverage.py

"""Distinguish pending, rejected and published enrichment per original product ID."""
from collections import Counter
from datetime import datetime, timezone
import importlib.util
import json
from pathlib import Path

spec = importlib.util.spec_from_file_location('enrichment', Path(__file__).with_name('enrich-research-library.py'))
e = importlib.util.module_from_spec(spec)
spec.loader.exec_module(e)


def audit():
    queue = e.read(e.WORK / 'product-queue.json')
    published = {}
    for directory in sorted((e.WORK / 'batches').iterdir()):
        file = directory / 'apply-result.json'
        if not file.exists(): continue
        result = e.read(file)['body']['result']
        if not (result.get('ok') and result.get('applied') and result.get('allRowsVerified')): continue
        receipt = e.read(directory / 'receipt.json')
        for row in receipt['products']:
            published[row['id']] = {**row, 'batch': directory.name, 'generation': result['generation'], 'allRowsVerified': True}
    states, rows = Counter(), []
    for product in queue:
        pid = product['productId']
        family, item = pid.split(':', 1)
        endpoints = []
        if product['missingVerifiedSpec'] or product['missingPrice']:
            endpoints.append(('detail', 'jd.get_item_detail_v1' if family == 'jd' else 'taobao.get_item_detail_v6'))
        if product['missingReviews']:
            endpoints.append(('reviews', 'jd.get_item_comments_v1' if family == 'jd' else 'taobao.get_item_comment_v3'))
        fields = {}
        for kind, endpoint in endpoints:
            prefix = e.WORK / 'collected' / (e.filename(pid) + '--' + e.filename(endpoint))
            meta_file = Path(str(prefix) + '.meta.json')
            if not meta_file.exists():
                fields[kind] = {'state': 'PENDING_COLLECTION'}
                continue
            meta = e.read(meta_file)
            if not meta.get('ok'):
                fields[kind] = {'state': 'REQUEST_FAILED', 'httpStatus': meta.get('httpStatus'), 'reason': meta.get('reason'), 'meta': meta_file.name}
                continue
            raw_file = prefix.parent / meta['responseFile']
            raw = raw_file.read_bytes()
            if e.sha(raw) != meta['sha256']: raise ValueError('ORIGINAL_RESPONSE_HASH_MISMATCH')
            field = {'state': 'VALID_RESPONSE', 'responseFile': meta['responseFile'], 'sha256': meta['sha256']}
            try:
                if kind == 'reviews':
                    accepted, excluded = e.reviews(json.loads(raw), family, item)
                    field.update(accepted=len(accepted), defaultReviews=sum(r['isDefault'] is True for r in accepted), excluded=dict(excluded))
                    if not accepted:
                        field['state'] = 'ALL_RETURNED_REVIEWS_REJECTED' if excluded else 'REQUESTED_PAGE_RETURNED_NO_REVIEWS'
                else:
                    specs, price = e.detail(json.loads(raw), family, item)
                    field.update(verifiedSpecs=len(specs), numericPrice=price is not None)
                    if not specs and price is None: field['state'] = 'NO_USABLE_SOURCE_ATTRIBUTES_OR_PRICE'
            except ValueError as error:
                field['state'] = str(error)
            fields[kind] = field
        publication = published.get(pid)
        if publication: state = 'PUBLISHED_BATCH_WITH_VERIFIED_ROWS'
        elif any(f['state'] == 'PENDING_COLLECTION' for f in fields.values()): state = 'PENDING_COLLECTION'
        elif any(f['state'] == 'REQUEST_FAILED' for f in fields.values()): state = 'REQUEST_FAILED'
        elif any(f.get('accepted', 0) or f.get('verifiedSpecs', 0) or f.get('numericPrice') for f in fields.values()): state = 'COLLECTED_AWAITING_PUBLICATION'
        else: state = 'NO_USABLE_ROWS_IN_RESPONSES'
        states[state] += 1
        rows.append({'id': pid, 'state': state, 'publication': publication, 'fields': fields})
    summary = {'queuedProducts': len(queue), 'states': dict(states), 'productionWrites': 0,
               'meaning': 'published is a batch state, not a claim that every product field or all platform reviews are complete'}
    result = {'schema': 'ksf.enrichment.coverage-audit.v1', 'checkedAt': datetime.now(timezone.utc).isoformat(), 'summary': summary, 'products': rows}
    e.save(e.WORK / 'coverage-audit.json', result)
    print(json.dumps(summary, ensure_ascii=False))
    return result


if __name__ == '__main__': audit()