kangshifu-board/scripts/research-workshop-quality-r13.py

"""Bounded read-only retrieval for report revision. Never publishes or writes business data."""
import concurrent.futures
import datetime
import hashlib
import json
import urllib.error
import urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / 'work/workshop-report-quality-20261110/data'
OUT.mkdir(parents=True, exist_ok=True)


def read(name, body):
    target = OUT / (name + '.json')
    if target.exists():
        return json.loads(target.read_bytes())
    req = urllib.request.Request('https://kangshifu.fmode.cn/api/library',
        data=json.dumps(body, ensure_ascii=False).encode(),
        headers={'Content-Type': 'application/json'}, method='POST')
    try:
        response = urllib.request.urlopen(req, timeout=45)
    except urllib.error.HTTPError as e:
        response = e
    raw = response.read()
    target.write_bytes(raw)
    data = json.loads(raw)
    receipt = dict(name=name, url=req.full_url, request=body,
        checkedAt=datetime.datetime.now(datetime.timezone.utc).isoformat(),
        httpStatus=response.code, responseSha256=hashlib.sha256(raw).hexdigest(),
        readOnly=data.get('readOnly'), databaseWrites=data.get('databaseWrites'),
        ok=data.get('ok'), generation=data.get('generation'), error=data.get('error'))
    (OUT/(name+'.request.json')).write_text(json.dumps(receipt, ensure_ascii=False, indent=2), encoding='utf8')
    assert not data.get('ok') or data.get('readOnly') is True and data.get('databaseWrites') == 0
    return data


def main():
    version = read('version', {'op': 'version'})
    assert version.get('ok'), version.get('error')
    gen = version['generation']
    jobs = [
        ('ksf-banmian-products', dict(op='browse', entity='products', filters={'brand':'ksf','category':'banmian'}, limit=50)),
        ('ksf-mixed', dict(op='browse', entity='products', filters={'brand':'ksf','category':'banmian'}, keyword='组合', limit=50)),
        ('flavour-switch', dict(op='browse', entity='reviews', filters={'category':'banmian'}, keyword='口味', limit=50)),
        ('flavour-fatigue', dict(op='browse', entity='reviews', filters={'category':'banmian'}, keyword='吃腻', limit=50)),
        ('small-pack', dict(op='browse', entity='products', filters={'category':'banmian'}, keyword='3', limit=50)),
    ]
    def job(kv):
        name, q = kv
        d = read(name, {**q, 'generation':gen})
        return dict(name=name, ok=d.get('ok'), matched=d.get('totalMatched'), returned=len(d.get('records',[])), error=d.get('error'))
    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
        results = list(pool.map(job, jobs))
    print(json.dumps(dict(generation=gen, results=results), ensure_ascii=False))


if __name__ == '__main__':
    main()