kangshifu-board/scripts/verify-product-images.py

"""Real browser checks for product images; does not write business data."""
import base64
import json
from pathlib import Path
import subprocess
import tempfile
import time
from urllib.request import urlopen

import websocket

ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / 'docs/industry-product-images-20261106'
SHOTS = OUT / 'screenshots'
SHOTS.mkdir(parents=True, exist_ok=True)
profile = Path(tempfile.mkdtemp(prefix='images-browser-', 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-component-update', '--disable-sync', '--disable-extensions',
    '--no-first-run', '--no-default-browser-check', '--user-data-dir=' + str(profile), 'about:blank'],
    stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, creationflags=subprocess.CREATE_NO_WINDOW)
ws = None
serial = 0
events = []
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(f'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
        rid = serial
        ws.send(json.dumps(dict(id=rid, method=method, params=params or {})))
        while True:
            response = json.loads(ws.recv())
            if response.get('id') == rid:
                assert 'error' not in response, response
                return response.get('result', {})
            events.append(response)
    def evaluate(expression):
        result = call('Runtime.evaluate', dict(expression=expression, returnByValue=True, awaitPromise=True))
        assert 'exceptionDetails' not in result, result
        return result['result'].get('value')
    def wait(expression):
        evaluate('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("Timed out: "+' + json.dumps(expression) + '));else setTimeout(tick,70)};tick()})')
    V = 'document.querySelector("kf-app")?.renderRoot.querySelector("view-library")'
    def action(body): return evaluate('(()=>{const v=' + V + ';' + body + '})()')
    def ready(): wait(V + '?._summary && !' + V + '._loading && (' + V + '._data || ' + V + '._detail)')
    def go(path):
        call('Page.navigate', {'url': 'http://127.0.0.1:4181' + path})
        ready()
    def shot(name):
        data = call('Page.captureScreenshot', dict(format='png', captureBeyondViewport=False))
        (SHOTS / (name + '.png')).write_bytes(base64.b64decode(data['data']))
    call('Page.enable'); call('Runtime.enable'); call('Network.enable')
    layouts = []
    for width in [1440, 1024, 390, 320]:
        call('Emulation.setDeviceMetricsOverride', dict(width=width, height=1000, deviceScaleFactor=1, mobile=width < 600))
        go('/library?type=products&platform=jd&shopId=1000084104&sort=reviews_desc')
        wait(V + '.renderRoot.querySelector("library-product-image")?.renderRoot.querySelector("img")?.complete')
        layout = action('return {width:innerWidth,documentWidth:document.documentElement.scrollWidth,loaded:v._data.records.filter(r=>r.image).length,cards:v.renderRoot.querySelectorAll(".product-choice").length,columns:getComputedStyle(v.renderRoot.querySelector(".record-list")).gridTemplateColumns.split(" ").length};')
        assert layout['documentWidth'] <= width, layout
        assert layout['loaded'] >= 10, layout
        assert not action('return !!v.renderRoot.querySelector("a a,a button");')
        layouts.append(layout)
        shot('catalog-' + str(width))
        go('/library/products/jd%3A100005702440')
        wait(V + '._detail.record.image && ' + V + '.renderRoot.querySelector("library-product-image")?.renderRoot.querySelector("img")?.naturalWidth > 0')
        assert evaluate('document.documentElement.scrollWidth <= innerWidth')
        shot('detail-' + str(width))
    call('Emulation.setDeviceMetricsOverride', dict(width=1440, height=1000, deviceScaleFactor=1, mobile=False))
    go('/library?compare=%5B%22jd%3A100005702440%22%2C%22jd%3A2848752%22%5D&comparing=1')
    C = V + '.renderRoot.querySelector("library-product-compare")'
    wait(C + ' && !' + C + '._loading && ' + C + '._items.every(i=>i.detail?.record.image)')
    wait(C + '.renderRoot.querySelector("library-product-image")?.renderRoot.querySelector("img")?.naturalWidth > 0')
    shot('compare-1440')
    call('Emulation.setDeviceMetricsOverride', dict(width=390, height=1000, deviceScaleFactor=1, mobile=True))
    assert evaluate('document.documentElement.scrollWidth <= innerWidth')
    shot('compare-390')
    # Explicitly injected read error: product text remains available.
    go('/library?type=products')
    evaluate('window.__imageFetch=window.fetch;window.fetch=(url,options)=>String(url).includes("/classes/KsfProductImage")?Promise.reject(new Error("image fixture offline")):window.__imageFetch(url,options);')
    action('v._data.records.forEach(r=>r.image=null);v.requestUpdate();')
    wait(V + '.renderRoot.querySelector("library-product-image")?.renderRoot.querySelector(".placeholder")')
    assert action('return v._data.records.length > 0 && !!v.renderRoot.querySelector("a[data-record-id] h4").textContent;')
    shot('missing-placeholder')
    action('v._data={...v._data,imagesUnavailable:true};v.requestUpdate();')
    wait(V + '.renderRoot.querySelector("library-product-image")?.renderRoot.textContent.includes("图片暂时无法加载")')
    shot('image-read-error')
    errors = [e for e in events if e.get('method') == 'Runtime.exceptionThrown']
    assert not errors, errors
    report = {'ok': True, 'layouts': layouts, 'consoleExceptions': 0, 'businessWrites': 0,
              'checks': ['production-image-bytes-rendered', 'four-widths-no-document-overflow', 'normal-product-navigation',
                         'full-size-detail-image', 'comparison-images', 'valid-card-html', 'missing-image-placeholder-keeps-text'],
              'fixtures': ['missing-image-placeholder']}
    (OUT / 'browser-acceptance.json').write_text(json.dumps(report, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    print(json.dumps(report, ensure_ascii=False, indent=2))
finally:
    if ws: ws.close()
    process.terminate()
    try: process.wait(timeout=8)
    except subprocess.TimeoutExpired: process.kill()