kangshifu-board/scripts/verify-industry-existing-data-public.py

"""Verify the read-only industry release and public static bytes; never collect upstream."""
import argparse
from concurrent.futures import ThreadPoolExecutor
import hashlib
import json
from pathlib import Path
from urllib.parse import urlsplit
from urllib.request import Request, urlopen

ROOT = Path(__file__).resolve().parents[1]


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--base-url', default='https://kangshifu.fmode.cn')
    args = parser.parse_args()
    base = args.base_url.rstrip('/')
    url = urlsplit(base)
    assert url.scheme in ('http', 'https') and url.netloc and not url.username and not url.password and not url.query and not url.fragment, 'INVALID_PUBLIC_URL'
    release = json.loads((ROOT / 'RELEASE.json').read_bytes())
    manifest = json.loads((ROOT / 'deploy/industry-ui-files.json').read_bytes())
    generation = release['generation']

    def post(op, **fields):
        body = json.dumps(dict(op=op, generation=generation, **fields)).encode()
        request = Request(base + '/api/industry', data=body, headers={'Content-Type': 'application/json'}, method='POST')
        with urlopen(request, timeout=30) as response:
            data = json.load(response)
        assert data['ok'] and data['generation'] == generation and data['readOnly'] and data['databaseWrites'] == 0, op
        return data

    contract = post('contract')
    assert 'workshop' in contract['commands']
    primary = post('reviews', filters={'platform': 'jd', 'shopId': '1000084104'})
    assert (primary['products'], primary['reviewRows'], primary['uniqueReadableTexts']) == (165, 746, 701)
    policies = post('policies', policyQuery={'application': 'sodium-claims'})
    evidence = policies['applications']['reviewedEvidence']
    assert evidence['connected'] and (evidence['verifiedClauses'], evidence['pendingClauses'], evidence['originalDocuments']) == (12, 19, 13)
    assert evidence['inputSha256'] == release['policyReviewSha256']
    for op in ('listings', 'rankings', 'workshop'):
        post(op)

    def static(entry):
        request = Request(base + '/' + entry['path'], headers={'Cache-Control': 'no-cache'})
        with urlopen(request, timeout=25) as response:
            raw = response.read(3_000_000)
        digest = hashlib.sha256(raw).hexdigest()
        assert digest == entry['sha256'], 'PUBLIC_STATIC_SHA_MISMATCH:' + entry['path']
        return dict(path=entry['path'], sha256=digest, matched=True)

    files = list(manifest['files']) + [dict(path='index.html', sha256=manifest['homepageSha256'])]
    with ThreadPoolExecutor(max_workers=6) as pool:
        verified = list(pool.map(static, files))
    print(json.dumps(dict(ok=True, generation=generation, readOnly=True, databaseWrites=0, paidCalls=0,
          primaryStore={key: primary[key] for key in ('products', 'reviewRows', 'uniqueReadableTexts')},
          verifiedPolicyClauses=12, pendingPolicyClauses=19, originalDocuments=13,
          publicStaticFiles=verified, browserVerificationRequired=True,
          overallMainlineComplete=False), ensure_ascii=False, indent=2))


if __name__ == '__main__':
    try:
        main()
    except Exception as error:
        print(json.dumps(dict(ok=False, error=str(error) if isinstance(error, AssertionError) else type(error).__name__,
                             databaseWrites=0, paidCalls=0)))
        raise SystemExit(1)