kangshifu-board/scripts/verify-library-cleanup-browser.py

"""Real production browser checks, without modifying responses or page code."""
import base64
from datetime import datetime,timezone
import json
from pathlib import Path
import subprocess
import tempfile
import time
from urllib.parse import quote
from urllib.request import urlopen
import websocket

ROOT=Path(__file__).resolve().parents[1]
WORK=ROOT/'work/library-cleanup-20261111'
final_dir=WORK/'round-02' if (WORK/'round-02/apply-result.json').exists() else WORK
applied=json.loads((final_dir/'apply-result.json').read_bytes())['body']['result']
OUT=WORK/('browser-final' if final_dir!=WORK else 'browser')
OUT.mkdir(parents=True,exist_ok=True)
profile=Path(tempfile.mkdtemp(prefix='cleanup-browser-',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=0;ws=None;events=[]
report={'ok':False,'checkedAt':datetime.now(timezone.utc).isoformat(),'scenes':[],'responseOverrides':0}
try:
    end=time.monotonic()+30
    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=150,suppress_origin=True,http_no_proxy=['127.0.0.1'])
    def call(method,params=None):
        global 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(expression):
        r=call('Runtime.evaluate',{'expression':expression,'returnByValue':True,'awaitPromise':True})
        assert 'exceptionDetails' not in r,r
        return r['result'].get('value')
    view='document.querySelector("kf-app")?.renderRoot?.querySelector("view-library")'
    call('Page.enable');call('Runtime.enable');call('Network.enable')
    plan=json.loads((WORK/'sealed-plan.json').read_bytes())['plan']
    removed_final={r['id'] for r in json.loads((final_dir/'sealed-plan.json').read_bytes())['plan']['removals']}
    repaired=next(r for r in plan['repairs'] if r['id'] not in removed_final)
    scenarios=[('products','/library?type=products&quality=audit',1440,900),
        ('social-mobile','/library?type=social&quality=audit',390,844),
        ('removed-social','/library/social/bilibili%3ABV11ENX6BE6p?type=social&quality=audit',1440,900),
        ('removed-product','/library/products/jd%3A100000283767?type=products&quality=audit',1440,900),
        ('corrected-body','/library/social/'+quote(repaired['id'],safe='')+'?type=social&quality=audit',1440,900),
        ('retained-article','/library/social/smzdm%3Aag5rd99w?type=social&quality=audit',390,844)]
    for name,path,width,height in scenarios:
        call('Emulation.setDeviceMetricsOverride',{'width':width,'height':height,'deviceScaleFactor':1,'mobile':False})
        call('Page.navigate',{'url':'https://kangshifu.fmode.cn'+path})
        ev('new Promise((resolve,reject)=>{const end=Date.now()+100000;function tick(){const p='+view+';if(p&&!p._loading&&(p._error||p._data||p._detail)){p.updateComplete.then(()=>resolve(true));return}if(Date.now()>end)reject(Error("render timeout"));else setTimeout(tick,150)}tick()})')
        state=ev('(()=>{const p='+view+';return {error:p._error,generation:p._summary?.generation,counts:p._summary?.counts,total:p._data?.totalMatched,detailId:p._detail?.record?.id,fulltext:p._detail?.record?.fulltext_available,content:p._detail?.record?.content,renderedText:p.renderRoot.textContent,innerWidth,clientWidth:document.documentElement.clientWidth,scrollWidth:document.documentElement.scrollWidth}})()')
        assert state['generation']==applied['generation'],state
        assert state['innerWidth']==width and state['scrollWidth']<=state['clientWidth'],state
        if name in ('removed-social','removed-product'):
            assert state['error']=='未找到这条资料',state
            assert 'detailId' not in state,state
        else:
            assert not state['error'],state
        if name=='products':assert state['total']==applied['plannedCounts']['products'],state
        if name=='social-mobile':assert state['total']==applied['plannedCounts']['social_posts'],state
        if name=='corrected-body':assert state['detailId']==repaired['id'] and state['fulltext']==0 and not state.get('content'),state
        if name=='retained-article':
            expected=next(r['content'] for r in json.loads((ROOT/'work/database-enrichment-20261110/batches/five-topics-public-03/expected-records.json').read_bytes()) if r['id']=='smzdm:ag5rd99w')
            assert state['detailId']=='smzdm:ag5rd99w' and state['content']==expected,state
        (OUT/(name+'.png')).write_bytes(base64.b64decode(call('Page.captureScreenshot',{'format':'png','captureBeyondViewport':False})['data']))
        state.pop('content',None);state['renderedText']=state['renderedText'][-1200:]
        report['scenes'].append({'name':name,'path':path,'width':width,'height':height,**state})
        (OUT/'report.json').write_text(json.dumps(report,ensure_ascii=False,indent=2),encoding='utf8')
        print(json.dumps({'scene':name,'passed':True},ensure_ascii=False),flush=True)
    report['uncaughtExceptions']=[e['params'].get('exceptionDetails',{}) for e in events if e['method']=='Runtime.exceptionThrown']
    assert not report['uncaughtExceptions'],report['uncaughtExceptions']
    report['ok']=True
finally:
    (OUT/'report.json').write_text(json.dumps(report,ensure_ascii=False,indent=2),encoding='utf8')
    if ws:ws.close()
    browser.terminate()
    try:browser.wait(timeout=10)
    except subprocess.TimeoutExpired:browser.kill()
print(json.dumps({'ok':report['ok'],'scenes':len(report['scenes'])},ensure_ascii=False))