kangshifu-board/scripts/verify-expanded-review-publication.py

"""Read production reviews at one generation and compare every round-two review."""
import collections
import datetime as dt
import json
from pathlib import Path
import sqlite3
import urllib.error
import urllib.request

ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / 'docs/industry-product-reviews-20261106/round2'
API = 'https://kangshifu.fmode.cn/api/library'


def api(request):
    req = urllib.request.Request(API, json.dumps(request).encode(), {'Content-Type': 'application/json'})
    try:
        response = urllib.request.urlopen(req, timeout=40)
    except urllib.error.HTTPError as error:
        data = json.loads(error.read())
        raise ValueError(data.get('error', {}).get('code', 'READ_FAILED')) from error
    with response:
        data = json.load(response)
    if not data.get('ok'):
        raise ValueError(data.get('error', {}).get('code', 'READ_FAILED'))
    return data


def append_text(raw):
    value = json.loads(raw or 'null')
    if isinstance(value, str):
        return value.strip()
    if isinstance(value, dict):
        return '\n'.join(value[k] for k in ('content', 'comment', 'appendComment') if isinstance(value.get(k), str)).strip()
    if isinstance(value, list):
        return '\n'.join(part[k] for part in value if isinstance(part, dict)
                         for k in ('content', 'comment') if isinstance(part.get(k), str)).strip()
    return ''


def verify():
    summary = api({'op': 'summary'})
    generation = summary['generation']
    actual = {}
    cursor = None
    pages = 0
    while True:
        query = {'op': 'browse', 'entity': 'reviews', 'quality': 'audit', 'limit': 100, 'generation': generation}
        if cursor:
            query['cursor'] = cursor
        data = api(query)
        assert data['generation'] == generation
        assert data['totalMatched'] == summary['counts']['reviews']
        for row in data['records']:
            assert row['record_id'] not in actual
            actual[row['record_id']] = row
        cursor = data['nextCursor']
        pages += 1
        if pages % 10 == 0:
            print('Read production reviews: ' + str(len(actual)), flush=True)
        if not cursor:
            break
    assert len(actual) == summary['counts']['reviews']
    assert len({r['product_id'] for r in actual.values()}) == data['distinctLinkedProducts']
    selected = {}
    batches = []
    failures = []
    product_counts = collections.Counter()
    fields = ('product_id', 'content', 'rating_raw', 'rating_scale', 'review_date_raw',
              'purchased_sku_text', 'is_default', 'identity_state', 'category_state')
    for folder in sorted((ROOT / 'work/product-reviews/round2/intake').glob('batch-*')):
        receipt = json.loads((folder / 'verification.json').read_text(encoding='utf-8'))
        uri = (folder / 'staging.sqlite').resolve().as_uri() + '?mode=ro'
        matched = 0
        with sqlite3.connect(uri, uri=True) as connection:
            connection.row_factory = sqlite3.Row
            for row in connection.execute('SELECT * FROM reviews ORDER BY id'):
                row = dict(row)
                rid = row['id']
                assert rid not in selected
                selected[rid] = row
                production = actual.get(rid)
                if production is None:
                    failures.append({'reviewId': rid, 'code': 'MISSING'})
                    continue
                different = [key for key in fields if row[key] != production.get(key)]
                if append_text(row['append_content_json']) != production.get('append_content', ''):
                    different.append('append_content')
                if different:
                    failures.append({'reviewId': rid, 'code': 'FIELD_MISMATCH', 'fields': different})
                    continue
                matched += 1
                product_counts[row['product_id']] += 1
        batches.append({'batchId': receipt['batchId'], 'expectedReviews': receipt['reviews'], 'verifiedReviews': matched})
    assert len(selected) == 512
    api({'op': 'version', 'generation': generation})
    coverage = len({r['product_id'] for r in actual.values()})
    result = {'ok': not failures and len(product_counts) == 57, 'checkedAt': dt.datetime.now(dt.timezone.utc).isoformat(),
              'generation': generation, 'productionCounts': summary['counts'],
              'analysisReviews': summary['analysisCounts']['reviews'],
              'productionReviewsFullyRead': len(actual), 'pagesRead': pages,
              'productsWithStoredReviews': coverage,
              'productsWithoutStoredReviews': summary['counts']['products'] - coverage,
              'expectedProducts': 57, 'verifiedProducts': len(product_counts),
              'expectedReviews': len(selected), 'verifiedReviews': sum(product_counts.values()),
              'comparedFields': ['review_id', *fields, 'append_content'],
              'products': dict(sorted(product_counts.items())), 'batches': batches, 'failures': failures,
              'productionImported': not failures, 'readOnly': True, 'productionWrites': 0, 'paidCalls': 0}
    OUT.mkdir(parents=True, exist_ok=True)
    (OUT / 'production-readback.json').write_text(json.dumps(result, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    print(json.dumps({k: v for k, v in result.items() if k not in ('products', 'productionCounts')}, ensure_ascii=False, indent=2))
    assert result['ok'], 'PRODUCTION_READBACK_FAILED'


if __name__ == '__main__':
    verify()