kangshifu-board/scripts/export-industry-snapshot.py

"""Export the existing library through its read-only API, never an upstream collector."""
import argparse
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone
import hashlib
import json
from pathlib import Path
from urllib.request import Request, urlopen

ENTITIES = ['products', 'reviews', 'prices', 'specs', 'documents', 'clauses', 'rankings', 'coverage']


def main():
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument('--url', default='https://kangshifu.fmode.cn/api/library')
    p.add_argument('--out', type=Path, required=True)
    args = p.parse_args()
    if args.out.exists():
        raise ValueError('NEW_OUTPUT_REQUIRED')

    def read(payload):
        request = Request(args.url, json.dumps(payload).encode(), {'Content-Type': 'application/json'})
        with urlopen(request, timeout=45) as response:
            data = json.load(response)
        if not data.get('ok'):
            raise ValueError('READ_FAILED:' + str(data.get('error')))
        return data

    summary = read({'op': 'summary'})
    generation = summary['generation']

    def bound(payload):
        data = read({**payload, 'generation': generation})
        if data['generation'] != generation:
            raise ValueError('GENERATION_CHANGED')
        return data

    def export(entity):
        records, cursor, seen, pages = [], None, set(), 0
        total = None
        while True:
            request = {'op': 'browse', 'entity': entity, 'quality': 'audit', 'limit': 100}
            if cursor:
                request['cursor'] = cursor
            data = bound(request)
            if total is None:
                total = data['totalMatched']
            if total != data['totalMatched']:
                raise ValueError('COUNT_CHANGED')
            records.extend(data['records'])
            pages += 1
            cursor = data.get('nextCursor')
            if not cursor:
                break
            if cursor in seen or pages >= 500:
                raise ValueError('CURSOR_LOOP_OR_EXPORT_TOO_LARGE')
            seen.add(cursor)
        if len(records) != total or len({r['record_id'] for r in records}) != total:
            raise ValueError('INCOMPLETE_DATABASE_EXPORT:' + entity)
        print(json.dumps({'entity': entity, 'records': total, 'pages': pages}), flush=True)
        return entity, records, pages

    rows, page_counts = {}, {}
    with ThreadPoolExecutor(max_workers=3) as pool:
        for job in as_completed([pool.submit(export, entity) for entity in ENTITIES]):
            entity, records, pages = job.result()
            rows[entity], page_counts[entity] = records, pages
    for entity in ('documents', 'clauses', 'rankings'):
        def detail(row):
            data = bound({'op': 'detail', 'entity': entity, 'id': row['record_id']})
            return {**row, **data['record']}
        with ThreadPoolExecutor(max_workers=3) as pool:
            rows[entity] = list(pool.map(detail, rows[entity]))
    bound({'op': 'version'})
    snapshot = {'schema': 'ksf.industry-snapshot.v1', 'generation': generation,
                'exportedAt': datetime.now(timezone.utc).isoformat(), 'summary': summary,
                'entities': rows, 'pageCounts': page_counts,
                'scope': 'complete existing database projection, not complete upstream platforms',
                'paidCalls': 0, 'productionWrites': 0}
    args.out.parent.mkdir(parents=True, exist_ok=True)
    with args.out.open('x', encoding='utf-8') as f:
        json.dump(snapshot, f, ensure_ascii=False, separators=(',', ':'))
    print(json.dumps({'ok': True, 'generation': generation, 'file': str(args.out),
                      'bytes': args.out.stat().st_size,
                      'sha256': hashlib.sha256(args.out.read_bytes()).hexdigest()}, ensure_ascii=False))


if __name__ == '__main__':
    main()