kangshifu-board/scripts/verify-industry-workshop.py

"""Real-browser acceptance for workshop product analysis, actual filters and original evidence."""
import argparse
import base64
from datetime import datetime, timezone
import json
from pathlib import Path
import subprocess
import sys
import tempfile
import time
from urllib.request import Request, urlopen

import websocket

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


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--url', default='http://127.0.0.1:4183')
    parser.add_argument('--out', default='work/industry-workshop-20261107-r1/browser')
    args = parser.parse_args()
    out = (ROOT / args.out).resolve()
    out.mkdir(parents=True, exist_ok=True)
    profile = Path(tempfile.mkdtemp(prefix='innovation-edge-', dir=ROOT / 'work'))
    browser = subprocess.Popen(['C:/Program Files (x86)/Microsoft/Edge/Application/msedge.exe', '--headless=new',
        '--disable-gpu', '--disable-background-networking', '--disable-component-update', '--disable-extensions',
        '--no-first-run', '--remote-debugging-port=0', '--user-data-dir=' + str(profile), 'about:blank'],
        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, creationflags=subprocess.CREATE_NO_WINDOW)
    serial, ws, events = 0, None, []
    report = {'ok': False, 'checkedAt': datetime.now(timezone.utc).isoformat(), 'url': args.url,
              'checks': [], 'layouts': [], 'responseOverrides': 0, 'paidCalls': 0, 'productionWrites': 0}
    try:
        end = time.monotonic() + 20
        while not (profile / 'DevToolsActivePort').exists():
            assert time.monotonic() < end and browser.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')) if p['type'] == 'page')
        ws = websocket.create_connection(page['webSocketDebuggerUrl'], timeout=45, suppress_origin=True, http_no_proxy=['127.0.0.1'])

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

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

        view = 'document.querySelector("kf-app")?.renderRoot?.querySelector("view-industry")'
        def action(code): return ev('(()=>{const p=' + view + ';' + code + '})()')
        def ready(tab='insights'):
            ev('new Promise((resolve,reject)=>{let end=Date.now()+30000;function tick(){const p=' + view +
               ';if(p && p._tab===' + json.dumps(tab) + ' && !p._loading && p._data){p.updateComplete.then(()=>resolve(true));return}'
               'if(p?._error)reject(Error(p._error));else if(Date.now()>end)reject(Error("timeout"));else setTimeout(tick,70)}tick()})')
        def navigate(path, tab='insights'):
            call('Page.navigate', {'url': args.url + path})
            ready(tab)
        def change(field, value):
            action('const s=p.renderRoot.querySelector(' + json.dumps('select[data-filter=' + field + ']') + ');s.value=' + json.dumps(value) + ';s.dispatchEvent(new Event("change"));return true')
            ev('new Promise(r=>setTimeout(r,80))')
            ready()
        def shot(name):
            (out / name).write_bytes(base64.b64decode(call('Page.captureScreenshot', {'format': 'png', 'captureBeyondViewport': False})['data']))
        def check(name):
            report['checks'].append(name)
            (out/'in-progress.json').write_text(json.dumps(report,ensure_ascii=False,indent=2)+'\n',encoding='utf-8')
            print('PASS '+name,flush=True)
        def settle(): ev('new Promise(r=>setTimeout(r,120))')
        def click(selector, reload=True):
            point=action('const b=p.renderRoot.querySelector('+json.dumps(selector)+');if(!b)throw Error("missing selector");b.scrollIntoView({block:"center"});const r=b.getBoundingClientRect();return {x:r.x+r.width/2,y:r.y+r.height/2}')
            for kind in ['mousePressed','mouseReleased']:
                call('Input.dispatchMouseEvent',{'type':kind,**point,'button':'left','clickCount':1})
            settle()
            if reload:ready()
        def data():return action('return p._data')
        def dimensions():return action('return p._data.choices')
        def search(text):
            action('p.renderRoot.querySelector("[data-w-search]").value='+json.dumps(text)+';p.renderRoot.querySelector(".ws-product-search").requestSubmit();return true')
            settle();ready()
        def layout(section,width,height):
            action('p.renderRoot.querySelector("#workshop-'+section+'").scrollIntoView({block:"start"});return true');settle()
            metrics=ev('({innerWidth,clientWidth:document.documentElement.clientWidth,scrollWidth:document.documentElement.scrollWidth})')
            assert metrics['innerWidth']==width,metrics
            assert metrics['scrollWidth']<=metrics['clientWidth'],metrics
            bad=action('return [...p.renderRoot.querySelectorAll(".workshop-analysis *")].filter(e=>{const r=e.getBoundingClientRect();return r.width>0&&r.height>0&&(r.left < -1 || r.right>document.documentElement.clientWidth+1)}).slice(0,8).map(e=>({tag:e.tagName,cls:e.className,text:e.textContent.slice(0,80)}))')
            assert not bad,bad
            report['layouts'].append({'section':section,'width':width,'height':height,**metrics,'outOfBounds':len(bad)})

        # Independent readback of snapshot identifiers, raw values and provenance.
        snapshot=json.loads((ROOT/'work/industry-monitoring-20261107-r1/snapshot.json').read_text(encoding='utf-8'))
        entities=snapshot['entities'];products={p['record_id']:p for p in entities['products']};reviews={r['record_id']:r for r in entities['reviews']}
        sys.path.insert(0,str(ROOT/'server/voc-library'))
        from industry_research import analysis_eligible
        def verify_scope(result, selected):
            selected=set(selected);rs=[r for r in reviews.values() if r['product_id'] in selected]
            s=result['scope']
            assert s['products']==len(selected)
            assert s['reviewRows']==len(rs)
            assert s['reviewedProducts']==len({r['product_id'] for r in rs})
            assert sum(b['products'] for b in result['brands'])==len(selected)
            assert sum(b['reviewRows'] for b in result['brands'])==len(rs)
            assert s['eligibleRows']==sum(analysis_eligible(r,True) for r in rs)
            for p in result['products']['rows']:
                assert p['productId'] in selected and p['title']==products[p['productId']]['title']
                assert p['reviewRows']==sum(r['product_id']==p['productId'] for r in rs)
            for r in result['reviewEvidence']['rows']:
                original=reviews[r['reviewId']]
                assert r['productId'] in selected and r['productId']==original['product_id']
                assert r['content']==original.get('content','') and r['appendContent']==original.get('append_content','')
                assert r['sourceRecordId']==original.get('source_record_id')
                assert r['analysisEligible']==analysis_eligible(original,True)
            for group in [*result['motives'],*result['audiences'],*result['scenarios']]:
                for r in group['examples']:
                    original=reviews[r['reviewId']]
                    for hit in r['hits']:
                        assert hit['quote'] in original.get('content' if hit['field']=='main' else 'append_content','')
            return result

        call('Runtime.enable');call('Page.enable');call('Network.enable')
        call('Emulation.setFocusEmulationEnabled',{'enabled':True})
        call('Emulation.setDeviceMetricsOverride',{'width':1440,'height':1000,'deviceScaleFactor':1,'mobile':False})
        navigate('/industry')
        settle()
        base={pid for pid,p in products.items() if p['category']=='banmian'}
        baseline=verify_scope(data(),base);report['scope']=baseline['scope']
        assert baseline['generation']==snapshot['generation']
        assert action('return p.renderRoot.querySelectorAll(".ws-product").length')==12
        assert action('return p.renderRoot.querySelectorAll(".ws-review").length')==8
        requests=[json.loads(e['params']['request']['postData']) for e in events if e['method']=='Network.requestWillBeSent' and '/api/industry' in e['params']['request']['url'] and e['params']['request'].get('postData')]
        assert any(r['op']=='workshop' and r['generation']==snapshot['generation'] for r in requests)
        check('live-workshop-request-matches-independent-snapshot-products-raw-review-counts-identities-text-and-sources')

        click('[data-w-products-next]');second=data()
        assert second['products']['page']==2 and second['scope']==baseline['scope']
        assert abs(action('return p.renderRoot.querySelector("#workshop-products").getBoundingClientRect().top')-18)<3
        assert not set(p['productId'] for p in second['products']['rows']) & set(p['productId'] for p in baseline['products']['rows'])
        click('[data-w-products-prev]')
        click('[data-w-reviews-next]');second=data()
        assert second['reviewEvidence']['page']==2 and second['scope']==baseline['scope']
        assert not set(r['reviewId'] for r in second['reviewEvidence']['rows']) & set(r['reviewId'] for r in baseline['reviewEvidence']['rows'])
        click('[data-w-reviews-prev]')
        check('products-and-deduplicated-reviews-paginate-independently-with-unchanged-aggregate-scope')

        click('[data-w-choice="format:bag"]')
        import re
        bag={pid for pid in base if re.search('袋|连包',products[pid]['title'])}
        verify_scope(data(),bag)
        click('[data-w-choice="format:bowl"]')
        packaging={pid for pid in base if re.search('袋|连包|碗|桶',products[pid]['title'])}
        verify_scope(data(),packaging)
        assert dimensions()['format']==['bag','bowl']
        click('[data-w-choice="flavor:spicy"]')
        combined={pid for pid in packaging if re.search('麻辣|香辣|火鸡|超辣|小龙虾|红油',products[pid]['title'])}
        verify_scope(data(),combined)
        call('Page.reload');ready();settle()
        assert dimensions()['format']==['bag','bowl'] and dimensions()['flavor']==['spicy']
        check('packaging-and-flavor-really-filter-all-statistics-with-or-within-groups-and-and-across-groups-and-survive-refresh')
        click('[data-w-choice="format:all"]');click('[data-w-choice="flavor:all"]')
        action('p.renderRoot.querySelector(".ws-more-filters").open=true;return true')
        click('[data-w-choice="audience:office"]')
        office=data();assert 0<office['scope']['products']<len(base)
        for p in office['products']['rows']:
            assert p['matchEvidence']['audience']
            for hit in p['matchEvidence']['audience']:
                assert hit['quote'] in reviews[hit['reviewId']].get('content','')+reviews[hit['reviewId']].get('append_content','')
        click('[data-w-choice="scenario:lunch"]')
        assert 0<data()['scope']['products']<=office['scope']['products']
        assert all('audience' in p['matchEvidence'] and 'scenario' in p['matchEvidence'] for p in data()['products']['rows'])
        check('audience-and-scenario-restrict-products-by-exact-source-mentions-and-show-matching-evidence')

        click('[data-w-reset]');verify_scope(data(),base)
        change('category','all');assert data()['scope']['products']==len(products)
        change('category','banmian');change('brand','ksf')
        ksf={pid for pid in base if products[pid]['brand']=='ksf'};verify_scope(data(),ksf)
        change('shopId','1000084104')
        assert action('return p._filters.platform')=='jd'
        primary={pid for pid in ksf if products[pid]['shop_id']=='1000084104' and products[pid]['platform']=='jd'}
        verify_scope(data(),primary)
        change('platform','tmall');assert not action('return p.routeQuery.shopId')
        verify_scope(data(),{pid for pid in ksf if products[pid]['platform']=='tmall'})
        check('category-platform-brand-and-shop-selects-update-data-and-shop-automatically-binds-its-platform')

        click('[data-w-reset]');search('jd:100000348106')
        one=verify_scope(data(),['jd:100000348106']);assert one['products']['total']==1
        search('根本不存在的商品XYZ');empty_data=data()
        assert empty_data['scope']['products']==0 and empty_data['scope']['reviewRows']==0
        assert action('return p.renderRoot.querySelector(".ws-empty").textContent.includes("没有匹配商品")')
        assert not action('return p.renderRoot.querySelector("#workshop-opportunities")')
        click('[data-w-reset]')
        check('keyword-search-constrains-aggregate-products-and-reviews-and-empty-scope-does-not-invent-findings')

        pid=data()['products']['rows'][0]['productId']
        click('[data-w-product-reviews="'+pid+'"]')
        assert data()['scope']==baseline['scope']
        assert all(r['productId']==pid for r in data()['reviewEvidence']['rows'])
        click('.ws-evidence-controls>button')
        click('[data-w-pain="seasoning"]')
        assert data()['scope']==baseline['scope'] and data()['reviewEvidence']['total']>0
        assert action('return p.renderRoot.querySelector("[data-w-review-theme]").value')=='seasoning'
        assert action('return p.renderRoot.querySelector("[data-w-review-kind]").value')=='problem'
        assert all(any(t['id']=='seasoning' and any(h['kind']=='problem' for h in t['hits']) for t in r['themes']) for r in data()['reviewEvidence']['rows'])
        click('[data-w-opportunity="seasoning"]')
        assert action('return p.routeQuery.w_theme')=='seasoning' and not action('return p.routeQuery.w_kind')
        ev('history.back()');settle();ready();assert action('return p.routeQuery.w_kind')=='problem'
        ev('history.forward()');settle();ready();assert not action('return p.routeQuery.w_kind')
        check('product-pain-and-opportunity-buttons-open-their-own-original-comments-without-changing-summary-and-history-restores-them')

        call('Browser.grantPermissions',{'origin':args.url,'permissions':['clipboardReadWrite','clipboardSanitizedWrite']})
        click('[data-copy-agent]',False)
        text=ev('navigator.clipboard.readText()')
        assert snapshot['generation'] in text and '原文查看限定' in text and 'seasoning' in text
        refs=[json.loads(line) for line in text.splitlines() if line.startswith('{') and '"database"' in line]
        assert any(r['entity']=='products' for r in refs) and any(r['entity']=='reviews' for r in refs)
        for r in refs:
            assert any(row['record_id']==r['id'] for row in entities[r['entity']]),r
        report['copy']={'bytes':len(text.encode()),'references':len(refs),'exactReferencesVerified':len(refs),'generation':snapshot['generation']}
        check('real-clipboard-copy-includes-scope-statistics-exact-originals-and-all-database-identifiers-verified-against-snapshot')

        # Source navigation runs the real library generation guard.
        click('.ws-review footer a',False)
        ev('new Promise((resolve,reject)=>{const end=Date.now()+30000;function tick(){const p=document.querySelector("kf-app")?.renderRoot?.querySelector("view-library");if(p?._detail&&!p._loading)resolve(true);else if(p?._error)reject(Error(p._error));else if(Date.now()>end)reject(Error("library source timeout"));else setTimeout(tick,70)}tick()})')
        assert '/library/reviews/' in ev('location.pathname')
        ev('history.back()');settle();ready()
        assert action('return p.routeQuery.w_theme')=='seasoning'
        check('original-review-link-checks-library-generation-and-opens-real-database-detail-with-back-restoring-analysis')

        navigate('/industry?tab=reviews&category=banmian', 'reviews')
        assert action('return !!p._data.corpus')
        navigate('/industry?tab=listings&category=banmian', 'listings')
        assert action('return !!p._data.inventory')
        check('existing-review-corpus-and-product-inventory-remain-available-under-maintenance')
        navigate('/industry');settle()
        for width,height in [(1440,1000),(1024,900),(768,1024),(390,844),(320,568)]:
            call('Emulation.setDeviceMetricsOverride',{'width':width,'height':height,'deviceScaleFactor':1,'mobile':False});settle()
            ev('scrollTo(0,0)');settle();shot('overview-'+str(width)+'.png')
            for section in ['products','needs','audience','map','opportunities','trend','evidence']:
                layout(section,width,height)
                if width in (1440,390,320):shot(section+'-'+str(width)+'.png')
            action('p.renderRoot.querySelector(".copy-menu").open=true;return true')
            click('[data-preview-agent]',False)
            assert action('return p.renderRoot.querySelector(".agent-copy-dialog").matches(":modal")')
            d=action('const x=p.renderRoot.querySelector(".agent-copy-dialog"),r=x.getBoundingClientRect();return {width:r.width,right:r.right,bottom:r.bottom,top:r.top,scrollWidth:x.scrollWidth,clientWidth:x.clientWidth}')
            assert d['right']<=width and d['top']>=0 and d['bottom']<=height and d['scrollWidth']<=d['clientWidth'],d
            report.setdefault('copyDialogs',[]).append({'width':width,'height':height,**d})
            if width==390:shot('copy-390.png')
            click('.copy-dialog-head>button',False)
        check('seven-sections-and-native-copy-dialog-fit-five-real-viewports-including-320px-without-horizontal-overflow')
        # Finalize screenshot evidence on the real page, not an error route.
        assert action('return p.renderRoot.querySelector(".p-head h2").textContent')=='产品创新洞察'
        errors=[e for e in events if e['method']=='Runtime.exceptionThrown']
        assert not errors,errors
        report['javascriptExceptions']=len(errors)
        check('no-uncaught-browser-exceptions-no-response-overrides-and-no-production-or-paid-operations')
        report['ok']=True
    except Exception:
        if ws:shot('failure.png')
        raise
    finally:
        if ws:ws.close()
        browser.terminate()
        try:browser.wait(timeout=10)
        except subprocess.TimeoutExpired:browser.kill();browser.wait(timeout=10)
        (out/'verification.json').write_text(json.dumps(report,ensure_ascii=False,indent=2)+'\n',encoding='utf-8')
    print(json.dumps({'ok':report['ok'],'checks':report['checks'],'layouts':len(report['layouts']),'copyDialogs':len(report.get('copyDialogs',[])),'receipt':str(out/'verification.json')},ensure_ascii=False,indent=2))

if __name__=='__main__':main()