"""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))