kangshifu-board/scripts/verify-single-sku-validation.py

"""Check the offline handoff, responsive layout, unknown states and JSON export."""

import argparse
import base64
from datetime import datetime, timezone
import hashlib
from html.parser import HTMLParser
import json
from pathlib import Path
import subprocess
import tempfile
import time
from urllib.parse import unquote, urlparse
from urllib.request import urlopen

import websocket


ROOT = Path(__file__).resolve().parents[1]


class Links(HTMLParser):
    def __init__(self):
        super().__init__()
        self.hrefs = []
        self.ids = []

    def handle_starttag(self, tag, attrs):
        attrs = dict(attrs)
        if 'id' in attrs:
            self.ids.append(attrs['id'])
        if tag == 'a' and 'href' in attrs:
            self.hrefs.append(attrs['href'])


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--package', type=Path,
                        default=ROOT / 'artifacts/ksf-single-sku-validation-20261107-r1/package')
    args = parser.parse_args()
    package = args.package.resolve()
    parsed = Links()
    page_text = (package / 'index.html').read_text('utf-8')
    parsed.feed(page_text)
    assert len(parsed.ids) == len(set(parsed.ids)), 'DUPLICATE_HTML_IDS'
    for href in parsed.hrefs:
        assert not urlparse(href).scheme, 'EXTERNAL_LINK_IN_OFFLINE_PACKAGE'
        assert (package / unquote(href)).is_file(), href

    evidence = json.loads((package / 'evidence.json').read_text('utf-8'))
    template = json.loads((package / 'current-label-response.template.json').read_text('utf-8'))
    assert all(v is None for v in template['currentLabel'].values())
    assert evidence['historicalPhotoLabel']['preparation']['completeMinutes'] is None
    assert evidence['historicalPhotoLabel']['nutrition']['netWeightG'] is None
    assert evidence['historicalPhotoLabel']['packagingCodeVisible'] == 'WXFC-1A'
    assert evidence['historicalPhotoLabel']['standardCodeVisible'] == 'GB 17400'
    assert len(evidence['otherMissingSkuIds']) == 7
    for photo in evidence['photos']:
        assert hashlib.sha256((package / photo['path']).read_bytes()).hexdigest() == photo['sha256']

    work = ROOT / 'work/single-sku-validation-20261107-r1'
    work.mkdir(exist_ok=False)
    downloads = work / 'downloads'
    downloads.mkdir()
    profile = Path(tempfile.mkdtemp(prefix='edge-', dir=work))
    screenshots = package / 'verification'
    screenshots.mkdir(exist_ok=False)
    browser = Path('C:/Program Files (x86)/Microsoft/Edge/Application/msedge.exe')
    assert browser.is_file(), 'EDGE_NOT_FOUND'
    process = subprocess.Popen([
        str(browser), '--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': datetime.now(timezone.utc).isoformat(),
              'scope': 'local offline artifact; no product study performed',
              'layout': [], 'screenshots': [], 'checks': {},
              'paidCalls': 0, 'productionWrites': 0, 'participantsObserved': 0}
    try:
        deadline = time.monotonic() + 20
        while not (profile / 'DevToolsActivePort').exists():
            assert time.monotonic() < deadline and process.poll() is None, 'BROWSER_START_TIMEOUT'
            time.sleep(.1)
        port = (profile / 'DevToolsActivePort').read_text().splitlines()[0]
        pages = json.load(urlopen('http://127.0.0.1:' + port + '/json/list', timeout=5))
        page = next(p for p in pages if p['type'] == 'page')
        ws = websocket.create_connection(page['webSocketDebuggerUrl'], timeout=30,
                                         suppress_origin=True, http_no_proxy=['127.0.0.1', 'localhost'])

        def call(method, params=None):
            nonlocal serial
            serial += 1
            ws.send(json.dumps({'id': serial, 'method': method, 'params': params or {}}))
            while True:
                result = json.loads(ws.recv())
                if result.get('id') == serial:
                    assert 'error' not in result, result
                    return result.get('result', {})
                events.append(result)

        def evaluate(expression):
            result = call('Runtime.evaluate', {'expression': expression,
                                             'returnByValue': True, 'awaitPromise': True})
            assert 'exceptionDetails' not in result, result
            return result['result'].get('value')

        def screenshot(name):
            result = call('Page.captureScreenshot', {'format': 'png', 'captureBeyondViewport': False})
            path = screenshots / name
            path.write_bytes(base64.b64decode(result['data']))
            report['screenshots'].append(path.relative_to(package).as_posix())

        call('Page.enable')
        call('Runtime.enable')
        call('Network.enable')
        call('Log.enable')
        call('Network.emulateNetworkConditions', {'offline': True, 'latency': 0,
                                                'downloadThroughput': 0, 'uploadThroughput': 0})
        call('Browser.setDownloadBehavior', {'behavior': 'allow', 'downloadPath': str(downloads)})
        call('Page.navigate', {'url': (package / 'index.html').as_uri()})
        evaluate('new Promise((resolve,reject)=>{const end=Date.now()+10000;'
                 'const tick=()=>{if(document.readyState==="complete" && typeof readResponse==="function")'
                 'resolve(true);else if(Date.now()>end)reject(Error("load timeout"));'
                 'else setTimeout(tick,40)};tick()})')
        evaluate('Promise.all([...document.images].map(img=>img.decode()))')
        images = evaluate('[...document.images].map(i=>({width:i.naturalWidth,height:i.naturalHeight}))')
        assert images == [{'width': 608, 'height': 1080}] * 3, images
        report['checks']['allThreeOriginalImagesDecodedOffline'] = True

        blank = evaluate('readResponse()')
        assert blank == template, 'BLANK_EXPORT_CHANGES_UNKNOWN_STATES'
        assert evaluate('[...document.querySelectorAll("input,textarea")].every(x=>x.closest("label"))')
        report['checks']['emptyFormPreservesAllUnknowns'] = True
        report['checks']['allInputsHaveLabels'] = True
        test_values = {'providedBy': '仅用于离线导出验收', 'providedAt': '2026-11-07',
                       'netWeightG': '96', 'servingSizeG': '96', 'servingsPerPackage': '1',
                       'sodiumMg': '0', 'sodiumNrvPercent': '0',
                       'preparationMethodsVerbatim': '仅用于导出测试\n第二行 <原文> & "引号"'}
        evaluate('(()=>{const values=' + json.dumps(test_values, ensure_ascii=False) +
                 ';for(const [name,value] of Object.entries(values))'
                 'document.getElementById("response-form").elements.namedItem(name).value=value;return true})()')
        filled = evaluate('readResponse()')
        assert filled['currentLabel']['sodiumMg'] == 0
        assert filled['currentLabel']['sodiumNrvPercent'] == 0
        assert filled['currentLabel']['netWeightG'] == 96
        assert filled['currentLabel']['preparationMethodsVerbatim'] == test_values['preparationMethodsVerbatim']
        assert filled['currentLabel']['version'] is None
        assert filled['verification']['status'] == 'NOT_REVIEWED'
        assert filled['participantTest']['status'] == 'NOT_STARTED'
        assert filled['participantTest']['participantsObserved'] == 0
        evaluate('document.querySelector("#response-form button[type=submit]").click()')
        download = downloads / 'jd-100297147639-current-label-response.json'
        deadline = time.monotonic() + 10
        while not download.exists():
            assert time.monotonic() < deadline, 'DOWNLOAD_TIMEOUT'
            time.sleep(.05)
        assert json.loads(download.read_text('utf-8')) == filled
        report['checks']['realJsonDownloadPreservesNumbersUnicodeAndNulls'] = True
        report['checks']['exportDoesNotClaimVerificationOrStudyCompletion'] = True
        evaluate('document.getElementById("response-form").reset();'
                 'document.getElementById("export-feedback").textContent="";true')
        assert evaluate('readResponse()') == template

        for width in [1440, 1024, 768, 390, 320]:
            call('Emulation.setDeviceMetricsOverride', {'width': width, 'height': 1100,
                                                       'deviceScaleFactor': 1, 'mobile': False})
            evaluate('window.scrollTo(0,0);new Promise(r=>requestAnimationFrame(()=>requestAnimationFrame(r)))')
            metrics = evaluate('({width:innerWidth,scrollWidth:document.documentElement.scrollWidth,'
                               'overflowing:[...document.body.querySelectorAll("*")].filter(e=>{'
                               'const r=e.getBoundingClientRect();return r.width>0 && '
                               '(r.left<-.5 || r.right>innerWidth+.5)}).map(e=>e.tagName+"."+e.className)})')
            assert metrics['scrollWidth'] <= metrics['width'], metrics
            assert not metrics['overflowing'], metrics
            report['layout'].append(metrics)
            if width == 1440:
                screenshot('desktop-1440.png')
            if width == 390:
                screenshot('mobile-390.png')
                evaluate('document.getElementById("response-title").scrollIntoView();true')
                screenshot('response-390.png')
        errors = [e for e in events if e['method'] == 'Runtime.exceptionThrown' or
                  e['method'] == 'Log.entryAdded' and e['params']['entry'].get('level') == 'error']
        remote = [e['params']['request']['url'] for e in events
                  if e['method'] == 'Network.requestWillBeSent'
                  and e['params']['request']['url'].startswith(('http:', 'https:'))]
        assert not errors, errors
        assert not remote, remote
        report['checks']['consoleErrors'] = 0
        report['checks']['httpRequestsByPage'] = 0
        report['checks']['allOfflinePackageLinksResolve'] = True
        report['checks']['historicalFactsAndCurrentUnknownsKeptSeparate'] = True
        report['checks']['sourcePhotoBytesUnchanged'] = True
        report['ok'] = True
    finally:
        if ws is not None:
            ws.close()
        process.terminate()
        try:
            process.wait(timeout=10)
        except subprocess.TimeoutExpired:
            process.kill()
            process.wait(timeout=10)
        (package / '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))


if __name__ == '__main__':
    main()