1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147
| async def collect_all_web_vitals(ws, session_id): """ 一次性采集所有 Core Web Vitals """ observers = """ (() => { const results = {}; // LCP const lcpObserver = new PerformanceObserver((list) => { const entries = list.getEntries(); const last = entries[entries.length - 1]; results.lcp = { time: last.startTime, size: last.size, element: last.element?.tagName || '', url: last.url || '' }; }); lcpObserver.observe({type: 'largest-contentful-paint', buffered: true}); // CLS let clsScore = 0; const clsObserver = new PerformanceObserver((list) => { for (const entry of list.getEntries()) { if (!entry.hadRecentInput) clsScore += entry.value; } }); clsObserver.observe({type: 'layout-shift', buffered: true}); // FCP const paintObserver = new PerformanceObserver((list) => { for (const entry of list.getEntries()) { if (entry.name === 'first-contentful-paint') { results.fcp = entry.startTime; } } }); paintObserver.observe({type: 'paint', buffered: true}); // 5 秒后断开并收集 return new Promise((resolve) => { setTimeout(() => { lcpObserver.disconnect(); clsObserver.disconnect(); paintObserver.disconnect(); results.cls = clsScore; resolve(JSON.stringify(results)); }, 5000); }); })(); """ result = await cdp(ws, "Runtime.evaluate", { "expression": observers, "sessionId": session_id, "awaitPromise": True, "returnByValue": True }, session_id) import json as pyjson vitals = pyjson.loads(result["result"]["value"]) print("===== Core Web Vitals 综合报告 =====") print(f"FCP: {vitals.get('fcp', 'N/A')}") print(f"LCP: {vitals.get('lcp', {}).get('time', 'N/A')}") print(f"CLS: {vitals.get('cls', 0):.4f}") return vitals
async def full_page_lifecycle_performance(ws, session_id, url): """ 完整页面生命周期性能追踪 从导航开始到加载完成的全部指标 """ print(f"开始追踪页面性能: {url}\n") await enable_performance(ws, session_id) await cdp(ws, "Runtime.evaluate", { "expression": """ window.__vitals = {}; // LCP observer new PerformanceObserver((list) => { const entries = list.getEntries(); window.__vitals.lcp = entries[entries.length - 1].startTime; }).observe({type: 'largest-contentful-paint', buffered: true}); // CLS observer window.__clsScore = 0; new PerformanceObserver((list) => { for (const entry of list.getEntries()) { if (!entry.hadRecentInput) window.__clsScore += entry.value; } }).observe({type: 'layout-shift', buffered: true}); // FCP observer new PerformanceObserver((list) => { for (const entry of list.getEntries()) { if (entry.name === 'first-contentful-paint') window.__vitals.fcp = entry.startTime; } }).observe({type: 'paint', buffered: true}); """, "sessionId": session_id }, session_id) nav_start = asyncio.get_event_loop().time() await cdp(ws, "Page.navigate", {"url": url}, session_id) await cdp(ws, "Page.loadEventFired", {}, session_id) load_time = asyncio.get_event_loop().time() - nav_start print(f"页面加载完成: {load_time:.2f}s") await asyncio.sleep(3) cdp_metrics = await get_performance_metrics(ws, session_id) vitals = await cdp(ws, "Runtime.evaluate", { "expression": "JSON.stringify({...window.__vitals, cls: window.__clsScore})", "sessionId": session_id, "returnByValue": True }, session_id) import json as pyjson web_vitals = pyjson.loads(vitals["result"]["value"]) print(f"\n===== 完整性能报告 =====") print(f"导航耗时: {load_time:.2f}s") print(f"FCP: {web_vitals.get('fcp', 'N/A')}") print(f"LCP: {web_vitals.get('lcp', 'N/A')}") print(f"CLS: {web_vitals.get('cls', 'N/A')}") return { "load_time": load_time, "cdp_metrics": cdp_metrics, "web_vitals": web_vitals }
|