"""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()