kangshifu-board/scripts/workshop-self-run-evidence.py

"""Bounded read-only research requests for one local workshop rehearsal."""
import argparse
import datetime
import hashlib
import json
from pathlib import Path
import urllib.error
import urllib.request

ROOT = Path(__file__).resolve().parents[1]
WORK = ROOT / 'work/workshop-self-run-20261110'
EVIDENCE = WORK / 'evidence'
GENERATION = '82b104fc024b49a75d2f5376bf9778ced6f63a9433b5fc5a6daa3c458d05d5ea'


def read(name, endpoint, body):
    assert endpoint in ('library', 'industry')
    assert body['op'] in {'summary', 'version', 'browse', 'detail', 'contract', 'overview', 'policies', 'rankings', 'workshop'}
    EVIDENCE.mkdir(parents=True, exist_ok=True)
    target = EVIDENCE / (name + '.json')
    assert not target.exists(), 'Evidence is immutable; use another request name'
    request = {**body}
    if body['op'] not in ('summary', 'version', 'contract'):
        request['generation'] = GENERATION
    url = 'https://kangshifu.fmode.cn/api/' + endpoint
    req = urllib.request.Request(url, data=json.dumps(request, ensure_ascii=False).encode(),
                                 headers={'Content-Type': 'application/json'}, method='POST')
    try:
        response = urllib.request.urlopen(req, timeout=40)
    except urllib.error.HTTPError as error:
        response = error
    raw = response.read()
    target.write_bytes(raw)
    data = json.loads(raw)
    receipt = {'name': name, 'requestedAt': datetime.datetime.now(datetime.timezone.utc).isoformat(),
               'url': url, 'request': request, 'httpStatus': response.code,
               'responseSha256': hashlib.sha256(raw).hexdigest(), 'bytes': len(raw),
               'ok': data.get('ok'), 'generation': data.get('generation'),
               'readOnly': data.get('readOnly'), 'databaseWrites': data.get('databaseWrites'),
               'error': data.get('error')}
    (EVIDENCE / (name + '.request.json')).write_text(json.dumps(receipt, ensure_ascii=False, indent=2), encoding='utf-8')
    assert not data.get('ok') or data.get('readOnly') is True and data.get('databaseWrites') == 0
    print(json.dumps(receipt, ensure_ascii=False))
    return data


if __name__ == '__main__':
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument('name')
    p.add_argument('endpoint', choices=['library', 'industry'])
    p.add_argument('request_json')
    args = p.parse_args()
    read(args.name, args.endpoint, json.loads(args.request_json))