kangshifu-board/scripts/fetch-workshop-self-run-detail.py

"""Record exact read requests; no collection, publishing or business writes."""
import concurrent.futures
import json
import runpy
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
read = runpy.run_path(str(ROOT / 'scripts/workshop-self-run-evidence.py'))['read']
base = ROOT / 'work/workshop-self-run-20261110/evidence'


def existing_or_read(name, endpoint, request):
    p = base / (name + '.json')
    if p.exists():
        return json.loads(p.read_bytes())
    return read(name, endpoint, request)


def job(item):
    name, endpoint, request = item
    d = existing_or_read(name, endpoint, request)
    return {'name': name, 'ok': d.get('ok'), 'error': d.get('error')}


if __name__ == '__main__':
    jobs = []
    reviews = json.loads((base / 'banmian-reviews.json').read_bytes())['records']
    for i, row in enumerate(reviews):
        jobs.append((f'review-{i+1:02}', 'library', {'op': 'detail', 'entity': 'reviews', 'id': row['record_id']}))
    products = ['jd:100005702440', 'jd:100034276081', 'jd:100151245426', 'jd:100004189813', 'jd:100017748144', 'jd:100051133922', 'jd:100090896362']
    for i, pid in enumerate(products):
        jobs.append((f'product-{i+1:02}', 'library', {'op': 'detail', 'entity': 'products', 'id': pid}))
        for entity in ['prices', 'specs']:
            jobs.append((f'product-{i+1:02}-{entity}', 'library', {'op': 'browse', 'entity': entity, 'filters': {'productId': pid}, 'limit': 50}))
    social = ['bilibili:BV1Ce4y1y7ae', 'bilibili:BV1KvxL6REnb', 'xiaohongshu:6a20e1cb0000000008030f1e',
              'xiaohongshu:6a43a84e000000001702b983', 'xiaohongshu:6a1e81c5000000003701e466',
              'bilibili:BV1uS25BuEPa', 'bilibili:BV12rkfBmEu5', 'xiaohongshu:6a1d6cbd0000000038020669',
              'xiaohongshu:653c86840000000025017ba3', 'bilibili:BV13p8K62EZW']
    for i, rid in enumerate(social):
        jobs.append((f'social-{i+1:02}', 'library', {'op': 'detail', 'entity': 'social', 'id': rid}))
    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
        results = list(pool.map(job, jobs))
    print(json.dumps({'requests': len(results), 'allOk': all(r['ok'] for r in results)}, ensure_ascii=False))