#!/usr/bin/env python3 """Compare cold admin page loads with a fixed, local API response. Requires Python Playwright and its Chromium browser. No real account or backend is contacted. Run with --baseline DIST/browser [--candidate DIST/browser]. """ import argparse import asyncio import json import statistics import sys import threading from functools import partial from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import urlsplit from playwright.async_api import async_playwright MEMBERS = { "page": 1, "pageSize": 20, "total": 0, "items": [], "agents": [], "summary": {"total": 0, "todayNew": 0, "sevenDayNew": 0, "studying": 0, "followUp": 0}, "refreshedAt": "2026-09-26T00:00:00.000Z", } REPORTS = { "page": 1, "pageSize": 20, "total": 0, "items": [], "summary": {"reportCount": 0, "studentCount": 0, "learnedCount": 0, "masteredCount": 0, "reviewDueCount": 0, "masteryRate": 0}, "sourceStatus": {"state": "synced", "label": "已同步", "message": "本地模拟数据", "lastSuccessAt": "", "lagSeconds": 0}, } VOCABULARY_TREE = {"items": [], "total": 0, "totalWords": 123, "refreshedAt": "2026-09-26T00:00:00.000Z"} VOCABULARY_WORDS = {"page": 1, "pageSize": 30, "total": 7, "items": [], "refreshedAt": "2026-09-26T00:00:00.000Z"} SESSION = { "objectId": "perf-test", "username": "perf-test", "sessionToken": "local-perf-test", "displayName": "测速账号", "roles": [], "operationsRole": "ops-manager", "isSuperAdmin": False, } class AppHandler(SimpleHTTPRequestHandler): def do_GET(self): path = Path(self.directory) / urlsplit(self.path).path.lstrip("/") if not path.is_file(): self.path = "/index.html" return super().do_GET() def log_message(self, *_args): pass def serve(directory): server = ThreadingHTTPServer(("127.0.0.1", 0), partial(AppHandler, directory=str(directory))) threading.Thread(target=server.serve_forever, daemon=True).start() return server def percentile(values, percentage): return round(statistics.quantiles(sorted(values), n=100, method="inclusive")[percentage - 1], 1) async def measure(browser, base_url, page_name): context = await browser.new_context(viewport={"width": 1440, "height": 900}) await context.add_init_script( "sessionStorage.setItem('xiaoshu_admin_session_v1', " + json.dumps(json.dumps(SESSION)) + ");" ) page = await context.new_page() cdp = await context.new_cdp_session(page) await cdp.send("Network.enable") await cdp.send("Network.setCacheDisabled", {"cacheDisabled": True}) await cdp.send("Network.emulateNetworkConditions", { "offline": False, "latency": 20, "downloadThroughput": 4 * 1024 * 1024 / 8, "uploadThroughput": 3 * 1024 * 1024 / 8, }) async def mock_gateway(route): await asyncio.sleep(0.15) if page_name == "vocabulary": operation = (route.request.post_data_json or {}).get("params", {}).get("operation") payload = VOCABULARY_TREE if operation == "ops/vocabulary/tree" else VOCABULARY_WORDS else: payload = REPORTS if page_name == "learning-reports" else MEMBERS await route.fulfill(status=200, content_type="application/json", body=json.dumps({"success": True, "data": payload})) await page.route("**/cloud-functions/**", mock_gateway) try: await page.goto(base_url + "/admin/" + page_name, wait_until="commit") if page_name == "vocabulary": await page.get_by_text("完整管理 123 条词汇", exact=False).first.wait_for(timeout=30000) await page.get_by_text("共 7 条符合条件的词汇", exact=False).first.wait_for(timeout=30000) else: ready_text = "当前筛选范围暂无学习报表" if page_name == "learning-reports" else "没有符合条件的会员" await page.get_by_text(ready_text, exact=True).first.wait_for(timeout=30000) result = await page.evaluate("""() => { const js = performance.getEntriesByType('resource').filter(item => item.name.endsWith('.js')); return { readyMs: performance.now(), jsRequests: js.length, jsTransferBytes: js.reduce((total, item) => total + item.transferSize, 0), jsNames: js.map(item => new URL(item.name).pathname.split('/').pop()) }; }""") return result finally: await context.close() async def main(): parser = argparse.ArgumentParser() parser.add_argument("--baseline", type=Path, required=True) parser.add_argument("--candidate", type=Path) parser.add_argument("--page", choices=["members", "learning-reports", "vocabulary"], default="members") parser.add_argument("--runs", type=int, default=10) parser.add_argument("--out", type=Path) parser.add_argument("--max-js-bytes", type=int, help="Fail if the candidate loads more JavaScript bytes") parser.add_argument("--ceilings", type=Path, help="JSON map from page name to JavaScript byte ceiling") args = parser.parse_args() if args.runs < 2: parser.error("--runs must be at least 2") roots = {"A": args.baseline} if args.candidate: roots["B"] = args.candidate servers = {name: serve(root) for name, root in roots.items()} urls = {name: f"http://127.0.0.1:{server.server_port}" for name, server in servers.items()} runs = {name: [] for name in roots} try: async with async_playwright() as playwright: browser = await playwright.chromium.launch() try: for index in range(args.runs): for name in (list(roots) if index % 2 == 0 else list(reversed(roots))): runs[name].append(await measure(browser, urls[name], args.page)) finally: await browser.close() finally: for server in servers.values(): server.shutdown() summary = { name: { "n": len(items), "readyP50Ms": percentile([item["readyMs"] for item in items], 50), "readyP75Ms": percentile([item["readyMs"] for item in items], 75), "jsRequests": max(item["jsRequests"] for item in items), "jsTransferBytes": max(item["jsTransferBytes"] for item in items), "jsNames": items[-1]["jsNames"], "runs": items, } for name, items in runs.items() } print(json.dumps({name: {key: value for key, value in item.items() if key not in ("runs", "jsNames")} for name, item in summary.items()}, ensure_ascii=False, indent=2)) if args.out: args.out.parent.mkdir(parents=True, exist_ok=True) args.out.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") ceiling = args.max_js_bytes if args.ceilings: ceiling = json.loads(args.ceilings.read_text(encoding="utf-8"))[args.page] if ceiling is not None: measured = summary["B" if args.candidate else "A"]["jsTransferBytes"] if measured > ceiling: print(f"JavaScript transfer ceiling exceeded: {measured} > {ceiling}", file=sys.stderr) raise SystemExit(1) if __name__ == "__main__": asyncio.run(main())