kangshifu-board/scripts/verify-original-review-browser.py

"""Read real published review pages and record the known append-text omissions."""
import argparse
import base64
from collections import defaultdict
from datetime import datetime, timezone
import json
from pathlib import Path
import subprocess
import tempfile
import time
from urllib.parse import quote
from urllib.request import Request, urlopen

import websocket

parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--evidence', required=True, type=Path)
parser.add_argument('--out', required=True, type=Path)
args = parser.parse_args()
source = args.evidence.resolve()
out = args.out.resolve()
out.mkdir(exist_ok=False)
shots = out / 'screenshots'
shots.mkdir()
verification = json.loads((source / 'verification.json').read_text(encoding='utf-8'))
originals = json.loads((source / 'selected-original-reviews.json').read_text(encoding='utf-8'))['records']
products = json.loads((source / 'product-verification.json').read_text(encoding='utf-8'))['records']
expected = defaultdict(dict)
for row in originals:
    expected[row['product_id']][row['record_id']] = row
sample_ids = list(dict.fromkeys([r['product_id'] for r in originals if r['append_content']] + [
    max(originals, key=lambda r: r['likeCount'] if type(r['likeCount']) is int else -1)['product_id'],
    next(r['productId'] for r in products if not r['eligibleExactReviews']),
]))
base = 'https://kangshifu.fmode.cn'
generation = verification['generation']


def api(query):
    request = Request(base + '/api/library', json.dumps(query).encode(), {'Content-Type': 'application/json'})
    with urlopen(request, timeout=40) as response:
        data = json.load(response)
    assert data.get('ok') and data['generation'] == generation
    return data


api({'op': 'summary', 'generation': generation})
profile = Path(tempfile.mkdtemp(prefix='real-review-browser-', dir=out))
process = subprocess.Popen([
    'C:/Program Files (x86)/Microsoft/Edge/Application/msedge.exe', '--headless=new',
    '--disable-gpu', '--remote-debugging-port=0', '--disable-background-networking',
    '--disable-extensions', '--no-first-run', '--user-data-dir=' + str(profile), 'about:blank',
], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, creationflags=subprocess.CREATE_NO_WINDOW)
ws = None
serial = 0
events = []
report = {'ok': False, 'checkedAt': datetime.now(timezone.utc).isoformat(), 'generation': generation,
          'realProductionPage': True, 'moduleOverrides': 0, 'tokenProvided': False,
          'products': [], 'productionWrites': 0, 'paidCalls': 0}
try:
    deadline = time.monotonic() + 20
    while not (profile / 'DevToolsActivePort').exists():
        assert time.monotonic() < deadline and process.poll() is None
        time.sleep(.1)
    port = (profile / 'DevToolsActivePort').read_text().splitlines()[0]
    page = next(p for p in json.load(urlopen('http://127.0.0.1:' + port + '/json/list', timeout=5)) if p['type'] == 'page')
    ws = websocket.create_connection(page['webSocketDebuggerUrl'], timeout=45, suppress_origin=True,
                                     http_no_proxy=['127.0.0.1', 'localhost'])

    def call(method, params=None):
        global serial
        serial += 1
        ws.send(json.dumps({'id': serial, 'method': method, 'params': params or {}}))
        while True:
            response = json.loads(ws.recv())
            if response.get('id') == serial:
                assert 'error' not in response, response
                return response.get('result', {})
            events.append(response)

    def ev(expression):
        data = call('Runtime.evaluate', {'expression': expression, 'returnByValue': True, 'awaitPromise': True})
        assert 'exceptionDetails' not in data, data
        return data['result'].get('value')

    def wait(expression):
        ev('new Promise((resolve,reject)=>{const end=Date.now()+30000;const tick=()=>{try{if(' + expression +
           '){resolve(true);return}}catch{}if(Date.now()>end)reject(new Error("review page timeout"));'
           'else setTimeout(tick,70)};tick()})')

    view = 'document.querySelector("kf-app")?.renderRoot.querySelector("view-library")'
    detail = view + '?.renderRoot.querySelector("library-product-detail")'

    def action(body):
        return ev('(()=>{const p=' + detail + ';' + body + '})()')

    call('Page.enable')
    call('Runtime.enable')
    call('Network.enable')
    for index, pid in enumerate(sample_ids):
        call('Emulation.setDeviceMetricsOverride', {'width': 1440, 'height': 1100,
                                                   'deviceScaleFactor': 1, 'mobile': False})
        url = base + '/library/products/' + quote(pid, safe='') + '?panel=reviews'
        call('Page.navigate', {'url': url})
        wait(detail + '?.product?.id===' + json.dumps(pid) + ' && ' + detail +
             '.panel==="reviews" && !' + detail + '._loading && ' + detail + '._data')
        assert ev(view + '._summary.generation') == generation
        snapshot = action('return {generation:p._data.generation,total:p._data.totalMatched,'
                          'scope:p.renderRoot.querySelector("select[aria-label=商品评价范围]").value,'
                          'rows:Array.from(p.renderRoot.querySelectorAll("article.review")).map(el=>({'
                          'id:el.dataset.childId,productId:el.dataset.productId,'
                          'content:el.querySelector("p.full-review").textContent,'
                          'append:el.querySelector(".appended p")?.textContent||""}))};')
        expected_rows = expected[pid]
        assert snapshot['generation'] == generation
        assert snapshot['total'] == len(expected_rows) == len(snapshot['rows'])
        if expected_rows:
            assert snapshot['scope'] == 'audit'
        assert {r['id']: r['content'] for r in snapshot['rows']} == {
            rid: row['content'] for rid, row in expected_rows.items()}
        assert all(r['productId'] == pid for r in snapshot['rows'])
        omissions = [r['id'] for r in snapshot['rows']
                     if r['append'] != expected_rows[r['id']]['append_content']]
        known = {r['reviewId'] for r in json.loads((source / 'failures.json').read_text(encoding='utf-8'))['selected']}
        assert set(omissions) <= known
        layouts = []
        for width in ([1440, 390, 320] if index == 0 else [390]):
            call('Emulation.setDeviceMetricsOverride', {'width': width, 'height': 1100,
                                                       'deviceScaleFactor': 1, 'mobile': width < 600})
            ev('new Promise(resolve=>requestAnimationFrame(()=>requestAnimationFrame(resolve)))')
            assert ev('document.documentElement.scrollWidth<=innerWidth')
            name = pid.replace(':', '-') + '-' + str(width) + '.png'
            data = call('Page.captureScreenshot', {'format': 'png', 'captureBeyondViewport': False})
            (shots / name).write_bytes(base64.b64decode(data['data']))
            layouts.append({'width': width, 'noDocumentOverflow': True, 'screenshot': 'screenshots/' + name})
        report['products'].append({'productId': pid, 'url': url, 'defaultScope': snapshot['scope'],
                                   'visibleMainReviews': len(expected_rows), 'allMainTextMatched': True,
                                   'appendOmissionsConfirmed': omissions, 'layouts': layouts})
        print(pid + ': main review text verified; append omissions=' + str(len(omissions)), flush=True)
    report['consoleExceptions'] = [e['params']['exceptionDetails'].get('text')
                                   for e in events if e.get('method') == 'Runtime.exceptionThrown']
    assert not report['consoleExceptions']
    api({'op': 'summary', 'generation': generation})
    report['mainReviewDisplayVerified'] = True
    report['knownAppendOmissions'] = sum(len(r['appendOmissionsConfirmed']) for r in report['products'])
    report['completeDataAcceptancePassed'] = report['knownAppendOmissions'] == 0
    report['ok'] = True
finally:
    if ws:
        ws.close()
    process.terminate()
    try:
        process.wait(timeout=8)
    except subprocess.TimeoutExpired:
        process.kill()
    (out / 'browser-verification.json').write_text(json.dumps(report, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
print(json.dumps(report, ensure_ascii=False, indent=2))