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

"""Verify published review text on the public board and local preview, without a token."""
import base64
import datetime as dt
import json
from pathlib import Path
import sqlite3
import subprocess
import tempfile
import time
from urllib.parse import quote
from urllib.request import urlopen

import websocket

ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / 'docs/industry-product-reviews-20261106/round2'
SHOTS = OUT / 'screenshots'
SHOTS.mkdir(parents=True, exist_ok=True)
evidence = json.loads((OUT / 'production-readback.json').read_text(encoding='utf-8'))
assert evidence['ok'] and evidence['verifiedReviews'] == 512
samples = []
for folder in sorted((ROOT / 'work/product-reviews/round2/intake').glob('batch-*')):
    with sqlite3.connect((folder / 'staging.sqlite').resolve().as_uri() + '?mode=ro', uri=True) as c:
        c.row_factory = sqlite3.Row
        pid = c.execute('SELECT product_id FROM reviews ORDER BY length(content) DESC,id LIMIT 1').fetchone()[0]
        rows = [dict(r) for r in c.execute('SELECT id,content FROM reviews WHERE product_id=?', (pid,))]
        samples.append({'batch': folder.name, 'productId': pid, 'reviews': rows})

profile = Path(tempfile.mkdtemp(prefix='published-reviews-', dir=ROOT / 'work/library-preview'))
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': dt.datetime.now(dt.timezone.utc).isoformat(),
          'expectedGeneration': evidence['generation'], 'tokenProvided': False,
          'publicBoardConnectionVerified': False, 'localPreviewVerified': False,
          'productionWrites': 0, 'paidCalls': 0, 'sites': []}
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=40, 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):
        result = call('Runtime.evaluate', {'expression': expression, 'returnByValue': True, 'awaitPromise': True})
        assert 'exceptionDetails' not in result, result
        return result['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("browser wait timeout"));'
           'else setTimeout(tick,70)};tick()})')

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

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

    def shot(name):
        data = call('Page.captureScreenshot', {'format': 'png', 'captureBeyondViewport': False})
        (SHOTS / (name + '.png')).write_bytes(base64.b64decode(data['data']))

    call('Page.enable')
    call('Runtime.enable')
    call('Network.enable')
    for label, base in [('public', 'https://kangshifu.fmode.cn'), ('local', 'http://127.0.0.1:4181')]:
        site = {'baseUrl': base, 'products': [], 'screenshots': []}
        for index, sample in enumerate(samples):
            call('Emulation.setDeviceMetricsOverride', {'width': 1440, 'height': 1100, 'deviceScaleFactor': 1, 'mobile': False})
            path = '/library/products/' + quote(sample['productId'], safe='') + '?panel=reviews'
            call('Page.navigate', {'url': base + path})
            wait(P + '?.panel==="reviews" && !' + P + '._loading && ' + P + '._data')
            assert ev(V + '._summary.generation') == evidence['generation']
            default_scope = action('return p.renderRoot.querySelector("select[aria-label=商品评价范围]").value;')
            default_count = action('return p._data.totalMatched;')
            if default_scope != 'audit':
                action('const select=p.renderRoot.querySelector("select[aria-label=商品评价范围]");'
                       'select.value="audit";select.dispatchEvent(new Event("change",{bubbles:true}));')
                wait(P + '?.query.relatedQuality==="audit" && !' + P + '._loading && ' + P + '._data?.effectiveQuery.quality==="audit"')
            snapshot = action('return {generation:p._data.generation,total:p._data.totalMatched,'
                              '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}))};')
            expected = {r['id']: r['content'] for r in sample['reviews']}
            assert snapshot['generation'] == evidence['generation']
            assert snapshot['total'] == len(expected) == len(snapshot['rows']), snapshot
            assert {r['id']: r['content'] for r in snapshot['rows']} == expected, sample['productId']
            assert all(r['productId'] == sample['productId'] for r in snapshot['rows'])
            layouts = []
            for width in [1440, 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'), (label, width)
                layouts.append({'width': width, 'noDocumentOverflow': True})
                if index == 0:
                    name = label + '-published-reviews-' + str(width)
                    shot(name)
                    site['screenshots'].append('screenshots/' + name + '.png')
            site['products'].append({'batch': sample['batch'], 'productId': sample['productId'],
                                     'url': base + path, 'defaultScope': default_scope,
                                     'defaultReviewCount': default_count, 'verifiedVisibleReviews': len(expected),
                                     'allVisibleTextMatched': True, 'layouts': layouts})
            print(label + ' ' + sample['batch'] + ': visible review text verified', flush=True)
        report['sites'].append(site)
        if label == 'public':
            report['publicBoardConnectionVerified'] = True
        else:
            report['localPreviewVerified'] = True
    failures = [e['params']['exceptionDetails'].get('text') for e in events if e.get('method') == 'Runtime.exceptionThrown']
    report['consoleExceptions'] = failures
    assert not failures, failures
    responses = [e['params']['response'] for e in events if e.get('method') == 'Network.responseReceived'
                 and e['params']['response']['url'].endswith('/api/library')]
    report['libraryHttpStatuses'] = sorted({int(r['status']) for r in responses})
    assert responses and report['libraryHttpStatuses'] == [200]
    report['ok'] = True
finally:
    if ws:
        ws.close()
    process.terminate()
    try:
        process.wait(timeout=8)
    except subprocess.TimeoutExpired:
        process.kill()
    (OUT / 'published-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))