kangshifu-board/scripts/verify-product-ui-release.py

"""Verify frozen product UI with real production data, before or after deployment."""
import argparse
import base64
import datetime as dt
import hashlib
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

ROOT = Path(__file__).resolve().parents[1]
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--package-dir', type=Path)
parser.add_argument('--out', type=Path)
parser.add_argument('--url', default='https://kangshifu.fmode.cn')
parser.add_argument('--production', action='store_true')
args = parser.parse_args()
PACKAGE = (args.package_dir or (ROOT if (ROOT / 'deploy/product-ui-files.json').exists()
                               else ROOT / 'work/product-images-layout-20261106-r2/package')).resolve()
WORK = (args.out or PACKAGE.parent).resolve()
SHOTS = WORK / 'screenshots'
SHOTS.mkdir(parents=True, exist_ok=True)
manifest = json.loads((PACKAGE / 'deploy/product-ui-files.json').read_text(encoding='utf-8'))
payload = {}
for item in manifest['files']:
    raw = (PACKAGE / 'payload' / item['path']).read_bytes()
    assert hashlib.sha256(raw).hexdigest() == item['sha256']
    payload[item['path']] = raw
BASE = args.url.rstrip('/')


def read_public(relative, data=None):
    headers = {'Cache-Control': 'no-cache'}
    if data is not None:
        headers['Content-Type'] = 'application/json'
    with urlopen(Request(BASE + '/' + relative, data, headers), timeout=40) as response:
        return response.read(), response.headers.get('Content-Type', '')


def api(request):
    raw, _ = read_public('api/library', json.dumps(request).encode())
    result = json.loads(raw)
    assert result.get('ok'), result.get('error')
    return result


summary = api({'op': 'summary'})
generation = summary['generation']
if args.production:
    for item in manifest['files']:
        raw, mime = read_public(item['path'])
        assert 'javascript' in mime and hashlib.sha256(raw).hexdigest() == item['sha256'], item['path']
    raw, _ = read_public('index.html')
    assert hashlib.sha256(raw).hexdigest() == manifest['homepageSha256']

profile = Path(tempfile.mkdtemp(prefix='integrated-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 = {}
report = {'ok': False, 'checkedAt': dt.datetime.now(dt.timezone(dt.timedelta(hours=8))).isoformat(),
          'generation': generation, 'productionInstalled': args.production, 'productionWrites': 0, 'paidCalls': 0,
          'verificationMode': 'public-site-no-overrides' if args.production else 'frozen-modules-overridden-in-private-browser',
          'baseUrl': BASE, 'realProductionData': True, 'layouts': [], 'fixtures': []}
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 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):
        rid = send(method, params)
        while True:
            response = json.loads(ws.recv())
            if response.get('method') == 'Fetch.requestPaused':
                request = response['params']
                relative = request['request']['url'].split('?', 1)[0].removeprefix(BASE + '/')
                assert relative in payload and not args.production
                overrides[relative] = overrides.get(relative, 0) + 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[relative]).decode()})
            elif response.get('id') == rid:
                assert 'error' not in response, response
                return response.get('result', {})
            else:
                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()+35000;const tick=()=>{try{if(' + expression +
           '){resolve(true);return}}catch{}if(Date.now()>end)reject(new Error("UI wait timed out: "+' + json.dumps(expression) +
           '));else setTimeout(tick,70)};tick()})')

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

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

    def ready():
        wait(V + '?._summary && !' + V + '._loading && (' + V + '._data || ' + V + '._detail)')
        assert ev(V + '._summary.generation') == generation

    def go(path):
        call('Page.navigate', {'url': BASE + path})
        ready()

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

    def size(width):
        call('Emulation.setDeviceMetricsOverride', {'width': width, 'height': 1100, 'deviceScaleFactor': 1, 'mobile': width < 600})
        ev('new Promise(resolve=>requestAnimationFrame(()=>requestAnimationFrame(resolve)))')

    def product_ready():
        wait(P + '?.product?.overview && !' + P + '._loading')
        wait(V + '.renderRoot.querySelector(".product-hero>library-product-image")?.renderRoot.querySelector("img")?.naturalWidth>0')

    call('Page.enable')
    call('Runtime.enable')
    call('Network.enable')
    call('Network.setCacheDisabled', {'cacheDisabled': True})
    if not args.production:
        call('Fetch.enable', {'patterns': [{'urlPattern': BASE + '/' + name + '*', 'resourceType': 'Script',
                                           'requestStage': 'Request'} for name in payload]})

    catalog = '/library?type=products&platform=jd&shopId=1000084104&sort=reviews_desc'
    for width in [1440, 1024, 768, 390, 320]:
        size(width)
        go(catalog)
        wait(V + '.renderRoot.querySelector("library-product-image")?.renderRoot.querySelector("img")?.naturalWidth>0')
        state = action('return {documentWidth:document.documentElement.scrollWidth,records:v._data.records.length,'
                       'images:v._data.records.filter(r=>r.image).length,'
                       'columns:getComputedStyle(v.renderRoot.querySelector(".record-list")).gridTemplateColumns.split(" ").length,'
                       'exactShop:v._data.records.every(r=>r.shop_id==="1000084104" && r.platform==="jd"),'
                       'noNestedControls:!v.renderRoot.querySelector("a button,a input,a a"),'
                       'firstImageIsStoredBytes:v.renderRoot.querySelector("library-product-image").image.thumbnail.startsWith("data:image/jpeg;base64,")};')
        assert state['documentWidth'] <= width and state['images'] == state['records'] > 0
        assert state['exactShop'] and state['noNestedControls'] and state['firstImageIsStoredBytes']
        assert state['columns'] == 2 if width < 600 else state['columns'] >= (2 if width >= 1024 else 1)
        if width in [1440, 390, 320]:
            shot('catalog-' + str(width))
        go('/library/products/jd%3A100005702440')
        product_ready()
        position = action('const image=v.renderRoot.querySelector(".product-hero>library-product-image").getBoundingClientRect();'
                          'const info=v.renderRoot.querySelector(".product-hero .product-info").getBoundingClientRect();'
                          'return {image:{left:image.left,right:image.right,top:image.top,bottom:image.bottom},'
                          'info:{left:info.left,right:info.right,top:info.top,bottom:info.bottom},'
                          'facts:v.renderRoot.querySelectorAll(".product-facts>div").length,'
                          'noDuplicate:!p.renderRoot.querySelector(".product-info"),documentWidth:document.documentElement.scrollWidth};')
        assert position['documentWidth'] <= width and position['facts'] >= 4 and position['noDuplicate']
        if width >= 600:
            assert position['info']['left'] >= position['image']['right']
        else:
            assert position['info']['top'] >= position['image']['bottom']
        if width in [1440, 390, 320]:
            shot('detail-' + str(width))
        report['layouts'].append({'width': width, 'catalog': state, 'detail': position})
        print('Catalog images and detail facts: ' + str(width) + ' passed', flush=True)

    size(1440)
    go('/library/products/taobao%3A1040511334739')
    product_ready()
    report['crossPlatformImageVerified'] = action('return v._detail.record.id==="taobao:1040511334739" && !!v._detail.record.image;')
    assert report['crossPlatformImageVerified']
    go(catalog)
    action('v.renderRoot.querySelector("button[aria-label=列表视图]").click();')
    wait(V + '.renderRoot.querySelector(".record-list.list")')
    assert action('return !!v.renderRoot.querySelector(".record-list.list library-product-image");')
    action('v.renderRoot.querySelector("button[aria-label=卡片视图]").click();'
           'v.renderRoot.querySelectorAll(".compare-toggle")[0].click();')
    wait(V + '._compareIds.length===1')
    action('v.renderRoot.querySelectorAll(".compare-toggle")[1].click();')
    wait(V + '._compareIds.length===2')
    action('v.renderRoot.querySelector("#open-compare").click();')
    wait(C + ' && !' + C + '._loading && ' + C + '._items.every(i=>i.detail?.record.image)')
    assert action('return c._items.every(i=>i.id===i.detail.record.id && i.detail.generation===c._summary.generation);')
    wait(C + '.renderRoot.querySelector("library-product-image")?.renderRoot.querySelector("img")?.naturalWidth>0')
    shot('compare-1440')
    size(390)
    assert ev('document.documentElement.scrollWidth<=innerWidth')
    assert action('const r=c.renderRoot.querySelector("dialog").getBoundingClientRect();return r.left>=0 && r.right<=innerWidth;')
    shot('compare-390')
    call('Page.reload', {'ignoreCache': True})
    ready()
    wait(C + ' && !' + C + '._loading && ' + C + '._items.length===2 && ' + C + '._items.every(i=>i.detail)')
    call('Input.dispatchKeyEvent', {'type': 'keyDown', 'key': 'Escape', 'code': 'Escape', 'windowsVirtualKeyCode': 27})
    call('Input.dispatchKeyEvent', {'type': 'keyUp', 'key': 'Escape', 'code': 'Escape', 'windowsVirtualKeyCode': 27})
    wait('!' + C)
    assert action('return v._compareIds.length===2;')
    report['comparisonImagesAndRefreshVerified'] = True
    report['listViewImageVerified'] = True
    assert action('return Array.from(v.renderRoot.querySelector("select[aria-label=平台]").options).some(o=>o.value==="tmall");')
    action('const s=v.renderRoot.querySelector("select[aria-label=平台]");s.value="tmall";s.dispatchEvent(new Event("change",{bubbles:true}));')
    wait(V + '.routeQuery.platform==="tmall" && !' + V + '._loading && ' + V + '._data')
    assert action('return !v.routeQuery.shopId && !v.routeQuery.cursor;')
    report['platformChangeClearsShopVerified'] = True

    size(1440)
    go('/library/products/jd%3A100017664018?panel=reviews')
    product_ready()
    wait(P + '._data?.totalMatched===10')
    assert action('return p.reviewQuality==="audit" && p.renderRoot.querySelectorAll("article.review").length===10;')
    action('const s=p.renderRoot.querySelector("select[aria-label=商品评价范围]");s.value="analysis";s.dispatchEvent(new Event("change",{bubbles:true}));')
    wait(P + '.query.relatedQuality==="analysis" && !' + P + '._loading && ' + P + '._data?.totalMatched===0')
    action('const s=p.renderRoot.querySelector("select[aria-label=商品评价范围]");s.value="audit";s.dispatchEvent(new Event("change",{bubbles:true}));')
    wait(P + '.query.relatedQuality==="audit" && !' + P + '._loading && ' + P + '._data?.totalMatched===10')
    call('Page.reload', {'ignoreCache': True})
    ready()
    wait(P + '._data?.totalMatched===10')
    go('/library/products/jd%3A100005702440?panel=reviews')
    product_ready()
    wait(P + '._data?.totalMatched===18')
    assert action('return p.reviewQuality==="analysis";')
    report['publishedReviewHotfixPreserved'] = True

    if not args.production:
        # These two presentation states are private fixtures, kept separate from real-data checks.
        go('/library/products/jd%3A100005702440')
        product_ready()
        action('const image=v.renderRoot.querySelector(".product-hero>library-product-image");image.image=null;')
        wait(V + '.renderRoot.querySelector(".product-hero>library-product-image")?.renderRoot.querySelector(".placeholder")')
        assert action('return !!v.renderRoot.querySelector(".product-facts").textContent && !!v.renderRoot.querySelector("h2").textContent;')
        action('v.renderRoot.querySelector(".product-hero>library-product-image").unavailable=true;')
        wait(V + '.renderRoot.querySelector(".product-hero>library-product-image")?.renderRoot.textContent.includes("图片暂时无法加载")')
        report['fixtures'] = ['missing-image-keeps-facts', 'image-read-error-distinguished-from-missing']
    exceptions = [e['params']['exceptionDetails'].get('text') for e in events if e.get('method') == 'Runtime.exceptionThrown']
    assert not exceptions, exceptions
    assert set(overrides) == set(payload) if not args.production else not overrides
    api({'op': 'version', 'generation': generation})
    report.update(ok=True, consoleExceptions=0, overriddenModules=overrides, filesVerified=len(payload),
                  publicBrowserVerified=args.production, upperRightFactsVerified=True,
                  mobileImageAboveFactsVerified=True)
finally:
    if ws:
        ws.close()
    process.terminate()
    try:
        process.wait(timeout=8)
    except subprocess.TimeoutExpired:
        process.kill()
    name = 'production-browser-verification.json' if args.production else 'browser-verification.json'
    (WORK / name).write_text(json.dumps(report, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
print(json.dumps(report, ensure_ascii=False, indent=2))