kangshifu-board/docs/workshop-decision-quality-20261109/verify_quality_preview.py

"""Check actual local preview text, navigation and desktop/mobile layout in Edge."""
from pathlib import Path
import base64
import functools
import http.server
import json
import subprocess
import tempfile
import threading
import time
from urllib.parse import quote, urlparse
from urllib.request import urlopen
import websocket

WORK = Path(__file__).resolve().parent
OUT = WORK / 'browser'
OUT.mkdir(exist_ok=True)
source = json.loads((WORK / 'probe-outputs.json').read_text(encoding='utf-8'))


class Handler(http.server.SimpleHTTPRequestHandler):
    def log_message(self, *args):
        pass

    def list_directory(self, path):
        self.send_error(404)
        return None


server = http.server.ThreadingHTTPServer(('127.0.0.1', 0), functools.partial(Handler, directory=str(WORK)))
threading.Thread(target=server.serve_forever, daemon=True).start()
base = 'http://127.0.0.1:' + str(server.server_port)
profile = Path(tempfile.mkdtemp(prefix='ksf-quality-browser-'))
edge = 'C:/Program Files (x86)/Microsoft/Edge/Application/msedge.exe'
proc = subprocess.Popen([edge, '--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)
seq = 0
ws = None
errors = []
requests = []
checks = []
try:
    deadline = time.monotonic() + 20
    while not (profile / 'DevToolsActivePort').exists():
        assert time.monotonic() < deadline and proc.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=20, suppress_origin=True, http_no_proxy=['127.0.0.1', 'localhost'])

    def call(method, params=None):
        global seq
        seq += 1
        rid = seq
        ws.send(json.dumps({'id': rid, 'method': method, 'params': params or {}}))
        while True:
            msg = json.loads(ws.recv())
            if msg.get('id') == rid:
                assert 'error' not in msg, msg
                return msg.get('result', {})
            if msg.get('method') == 'Runtime.exceptionThrown':
                errors.append(msg['params'])
            if msg.get('method') == 'Network.requestWillBeSent':
                requests.append(msg['params']['request']['url'])

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

    def settle():
        ev('new Promise(r=>requestAnimationFrame(()=>requestAnimationFrame(r)))', True)

    call('Page.enable')
    call('Runtime.enable')
    call('Network.enable')
    call('Network.setBlockedURLs', {'urls': ['https://*']})
    call('Page.navigate', {'url': base + '/' + quote('分析与成果样张.html')})
    ev('new Promise((resolve,reject)=>{const end=Date.now()+15000;const tick=()=>{if(document.querySelectorAll("[data-select]").length===5){resolve(true);return;}if(Date.now()>end)reject(new Error("preview timeout"));else setTimeout(tick,60)};tick()})', True)

    for width, height, name in [(1440, 1000, 'desktop'), (390, 844, 'mobile'), (360, 800, 'narrow')]:
        call('Emulation.setDeviceMetricsOverride', {'width': width, 'height': height, 'deviceScaleFactor': 1, 'mobile': width < 500})
        settle()
        for i, sample in enumerate(source['samples']):
            ev('document.querySelector(' + json.dumps(f'[data-select="{i}"]') + ').click()')
            for mode in ('reply', 'report'):
                ev('document.querySelector(' + json.dumps(f'[data-mode="{mode}"]') + ').click()')
                settle()
                result = ev('({text:document.querySelector("#content").innerText,width:innerWidth,scroll:document.documentElement.scrollWidth,minFont:Math.min(...[...document.querySelectorAll("#content p,#content td,#content th")].filter(e=>e.getClientRects().length).map(e=>parseFloat(getComputedStyle(e).fontSize))),reportHidden:document.querySelector("#report").hidden,selected:document.querySelector("[aria-current=true]").dataset.select})')
                assert result['scroll'] <= width and result['minFont'] >= 15, (name, sample['id'], result)
                assert int(result['selected']) == i and result['reportHidden'] == (mode == 'reply')
                for part in ('title', 'answer', 'meaning', 'evidence'):
                    assert sample[part] in result['text'], (sample['id'], part)
                if sample['next']:
                    assert sample['next'] in result['text'], sample['id']
                if mode == 'report':
                    for cell in sample['columns'] + [c for row in sample['rows'] for c in row]:
                        assert cell in result['text'], (sample['id'], cell)
                    scrollable = ev('(()=>{let e=document.querySelector(".table-wrap");if(!e)return null;let before=e.scrollLeft;e.scrollLeft=99999;let changed=e.scrollLeft>before;e.scrollLeft=0;return {needed:e.scrollWidth>e.clientWidth,changed};})()')
                    if width < 500 and scrollable:
                        assert scrollable['needed'] and scrollable['changed'], scrollable
                checks.append({k:v for k,v in result.items() if k != 'text'} | {'sample':sample['id'],'viewport':name,'mode':mode,'fullContentMatches':True})
                if (name, sample['id'], mode) in [('desktop','rd','report'),('desktop','trend','report'),('mobile','consumer','reply'),('mobile','revision','report')]:
                    ev('scrollTo(0,document.querySelector("#content").getBoundingClientRect().top+scrollY-12)')
                    settle()
                    shot = call('Page.captureScreenshot', {'format':'png','captureBeyondViewport':False})
                    (OUT / f'{name}-{sample["id"]}-{mode}.png').write_bytes(base64.b64decode(shot['data']))
                ev('scrollTo(0,0)')
    assert not errors, errors
    external = [u for u in requests if not u.startswith(base + '/') and not u.startswith('data:')]
    assert not external, external
    receipt = {'schema':'ksf.decision-quality-browser.v1','ok':True,'scope':'Local standalone preview only; not target Agent, production board or semantic quality validation','checks':checks,'renderChecks':len(checks),'samples':len(source['samples']),'viewportWidths':[1440,390,360],'minimumMainTextFont':15,'bodyHorizontalOverflow':False,'exceptions':errors,'externalPageRequests':external,'screenshotsManuallyReviewed':False}
    (WORK / 'browser-checks.json').write_text(json.dumps(receipt,ensure_ascii=False,indent=2)+'\n',encoding='utf-8')
    print(json.dumps({k:v for k,v in receipt.items() if k != 'checks'},ensure_ascii=False,indent=2))
finally:
    if ws:
        ws.close()
    proc.terminate()
    proc.wait(timeout=10)
    server.shutdown()
    server.server_close()