kangshifu-board/scripts/verify-review-display-hotfix.py

"""Verify the staged module or its published version against real public data."""
import argparse
import base64
import hashlib
import json
from pathlib import Path
import sqlite3
import subprocess
import tempfile
import time
from urllib.parse import quote
from urllib.request import Request, urlopen

import websocket

ROOT = Path(__file__).resolve().parents[1]
WORK = ROOT / 'work/review-display-hotfix-20261106-r1'
payload = (WORK / 'package/payload/src/pages/library-product-detail.js').read_bytes()
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--production', action='store_true', help='Read the deployed module without any browser override')
args = parser.parse_args()
production_files = []
if args.production:
    for relative, expected in [('src/pages/library-product-detail.js', hashlib.sha256(payload).hexdigest()),
                               ('index.html', 'dfeb6a6d0b63e8e1190f9308bf6619a5ca523a5a2f91af5f7ca5ad5fc9642db3')]:
        with urlopen(Request('https://kangshifu.fmode.cn/' + relative, headers={'Cache-Control': 'no-cache'}), timeout=30) as response:
            raw = response.read()
            actual = hashlib.sha256(raw).hexdigest()
            assert actual == expected, 'PUBLIC_FILE_SHA_MISMATCH:' + relative
            production_files.append({'path': relative, 'status': response.status, 'bytes': len(raw), 'sha256': actual,
                                     'contentType': response.headers.get('Content-Type'),
                                     'cacheControl': response.headers.get('Cache-Control')})


def api(request):
    req = Request('https://kangshifu.fmode.cn/api/library', json.dumps(request).encode(), {'Content-Type': 'application/json'})
    with urlopen(req, timeout=35) as response:
        result = json.load(response)
    assert result.get('ok'), result.get('error')
    return result


summary = api({'op': 'summary'})
generation = summary['generation']
empty = api({'op': 'browse', 'entity': 'products', 'quality': 'audit', 'sort': 'reviews_asc', 'limit': 1,
             'generation': generation})['records'][0]
assert empty['relatedCounts'].get('reviews', 0) == 0
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 connection:
        connection.row_factory = sqlite3.Row
        pid = connection.execute('SELECT product_id FROM reviews ORDER BY length(content) DESC,id LIMIT 1').fetchone()[0]
        rows = [dict(r) for r in connection.execute('SELECT id,content FROM reviews WHERE product_id=?', (pid,))]
        samples.append({'batch': folder.name, 'productId': pid, 'reviews': rows})

profile = Path(tempfile.mkdtemp(prefix='staged-module-browser-', dir=WORK))
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 = []
overrides = 0
report = {'ok': False, 'generation': generation, 'payloadSha256': hashlib.sha256(payload).hexdigest(),
          'verificationMode': 'public-site-no-overrides' if args.production else 'browser-only-single-module-override-on-public-site',
          'productionInstalled': args.production, 'productionWrites': 0, 'paidCalls': 0, 'products': [],
          'productionFiles': production_files}
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 send(method, params=None):
        global serial
        serial += 1
        ws.send(json.dumps({'id': serial, 'method': method, 'params': params or {}}))
        return serial

    def call(method, params=None):
        global overrides
        rid = send(method, params)
        while True:
            response = json.loads(ws.recv())
            if response.get('method') == 'Fetch.requestPaused':
                request = response['params']
                assert request['request']['url'].split('?')[0] == 'https://kangshifu.fmode.cn/src/pages/library-product-detail.js'
                overrides += 1
                send('Fetch.fulfillRequest', {'requestId': request['requestId'], 'responseCode': 200,
                    'responseHeaders': [{'name': 'Content-Type', 'value': 'application/javascript; charset=utf-8'},
                                        {'name': 'Cache-Control', 'value': 'no-store'}],
                    'body': base64.b64encode(payload).decode()})
            elif response.get('id') == rid:
                assert 'error' not in response, response
                return response.get('result', {})
            else:
                events.append(response)

    def ev(expression):
        response = call('Runtime.evaluate', {'expression': expression, 'returnByValue': True, 'awaitPromise': True})
        assert 'exceptionDetails' not in response, response
        return response['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("staged module timeout"));'
           'else setTimeout(tick,60)};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 ready(pid):
        wait(P + '?.product?.id===' + json.dumps(pid) + ' && ' + P + '.panel==="reviews" && !' + P + '._loading && ' + P + '._data')
        assert ev(V + '._summary.generation') == generation

    def go(pid, quality=None):
        path = '/library/products/' + quote(pid, safe='') + '?panel=reviews'
        if quality:
            path += '&relatedQuality=' + quality
        call('Page.navigate', {'url': 'https://kangshifu.fmode.cn' + path})
        ready(pid)

    def choose(quality, expected):
        action('const s=p.renderRoot.querySelector("select[aria-label=商品评价范围]");s.value=' + json.dumps(quality) +
               ';s.dispatchEvent(new Event("change",{bubbles:true}));')
        wait(P + '.query.relatedQuality===' + json.dumps(quality) + ' && !' + P + '._loading && ' + P + '._data?.totalMatched===' + str(expected))

    call('Page.enable')
    call('Runtime.enable')
    call('Network.enable')
    call('Network.setCacheDisabled', {'cacheDisabled': True})
    if not args.production:
        call('Fetch.enable', {'patterns': [{'urlPattern': 'https://kangshifu.fmode.cn/src/pages/library-product-detail.js*',
                                           'resourceType': 'Script', 'requestStage': 'Request'}]})
    call('Emulation.setDeviceMetricsOverride', {'width': 1440, 'height': 1100, 'deviceScaleFactor': 1, 'mobile': False})
    for index, sample in enumerate(samples):
        pid = sample['productId']
        go(pid)
        assert action('return p.reviewQuality;') == 'audit'
        assert action('return p._data.totalMatched;') == len(sample['reviews'])
        assert action('return p.renderRoot.querySelector("select[aria-label=商品评价范围]").value;') == 'audit'
        assert action('return p.renderRoot.textContent.includes("尚未计入分析概况");')
        actual = action('return Array.from(p.renderRoot.querySelectorAll("article.review")).map(e=>({'
                        'id:e.dataset.childId,content:e.querySelector("p.full-review").textContent}));')
        assert {r['id']: r['content'] for r in actual} == {r['id']: r['content'] for r in sample['reviews']}
        if index == 0:
            call('Page.reload', {'ignoreCache': True})
            ready(pid)
            assert action('return p._data.totalMatched;') == len(actual)
            choose('analysis', 0)
            call('Page.reload', {'ignoreCache': True})
            ready(pid)
            assert action('return p.reviewQuality;') == 'analysis'
            assert action('return p._data.totalMatched;') == 0
            choose('audit', len(actual))
            for width in [1440, 390, 320]:
                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')
                data = call('Page.captureScreenshot', {'format': 'png', 'captureBeyondViewport': False})
                prefix = 'production-reviews-' if args.production else 'patched-reviews-'
                (WORK / (prefix + str(width) + '.png')).write_bytes(base64.b64decode(data['data']))
            call('Emulation.setDeviceMetricsOverride', {'width': 1440, 'height': 1100, 'deviceScaleFactor': 1, 'mobile': False})
        report['products'].append({'productId': pid, 'batch': sample['batch'], 'defaultScope': 'audit',
                                   'visibleReviews': len(actual), 'allTextsMatched': True})
        print(sample['batch'] + ' default display passed', flush=True)
    go('jd:100005702440')
    assert action('return p.reviewQuality;') == 'analysis'
    assert action('return p._data.totalMatched;') == 18
    go(empty['record_id'])
    assert action('return p.reviewQuality;') == 'analysis'
    assert action('return p._data.totalMatched;') == 0
    assert action('return p.renderRoot.textContent.includes("不代表平台上没有评论");')
    assert not [e for e in events if e.get('method') == 'Runtime.exceptionThrown']
    assert overrides == 0 if args.production else overrides >= 8
    api({'op': 'version', 'generation': generation})
    report.update(ok=True, explicitAnalysisHonoredAfterReload=True, explicitAuditRestoresAll=True,
                  eligibleProductKeepsAnalysis18=True, emptyProductAccurate=True,
                  widthsVerified=[1440, 390, 320], consoleExceptions=0, overriddenModuleRequests=overrides,
                  allApiResponsesRealProduction=True)
finally:
    if ws:
        ws.close()
    process.terminate()
    try:
        process.wait(timeout=8)
    except subprocess.TimeoutExpired:
        process.kill()
    report_name = 'production-browser-verification.json' if args.production else 'browser-verification.json'
    (WORK / report_name).write_text(json.dumps(report, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
print(json.dumps(report, ensure_ascii=False, indent=2))