#!/usr/bin/env python3 """Paired local dashboard first-open and return benchmark with a slow fixed API. The mock session and gateway are synthetic. The marker is the real dashboard welcome and chart, not the title or loading placeholder. """ import argparse import asyncio import json import statistics 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 SESSION = {"objectId": "perf-test", "username": "perf-test", "sessionToken": "local-perf-test", "displayName": "测速账号", "roles": [], "operationsRole": "ops-manager", "isSuperAdmin": False} TREND = {"period": "week", "periodLabel": "近 7 天", "dateFrom": "2026-09-20", "dateTo": "2026-09-26", "points": [{"key": "2026-09-26", "label": "9月26日", "shortLabel": "周六", "courseCount": 2, "studentCount": 2, "completedCount": 1, "isToday": True}], "todayCourseCount": 2, "todayStudentCount": 2, "totalCourseCount": 2, "totalStudentCount": 2, "maxValue": 2, "refreshedAt": "2026-09-26T06:00:00.000Z", "liveSourceAvailable": True, "liveSourceMessage": "本地模拟数据"} DASHBOARD = {"identity": {"objectId": "perf-test", "username": "perf-test", "displayName": "运营经理", "roles": [], "operationsRole": "ops-manager", "isSuperAdmin": False}, "date": "2026-09-26", "metrics": [], "alerts": [], "scheduleTrend": TREND, "refreshedAt": "2026-09-26T06:00:00.000Z", "liveSourceAvailable": True} class AppHandler(SimpleHTTPRequestHandler): def do_GET(self): if not (Path(self.directory) / urlsplit(self.path).path.lstrip("/")).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, n): return round(statistics.quantiles(sorted(values), n=100, method="inclusive")[n - 1], 1) async def measure(browser, url, from_login=False): context = await browser.new_context(viewport={"width": 1440, "height": 900}) if not from_login: 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, }) dashboard_requests = 0 async def mock_gateway(route): nonlocal dashboard_requests operation = (route.request.post_data_json or {}).get("params", {}).get("operation", "") if route.request.url.endswith('/xiaoshu/app/gateway'): await asyncio.sleep(0.3) await route.fulfill(status=200, content_type="application/json", body=json.dumps({"retcode": 0, "result": {"objectId": SESSION["objectId"], "userName": SESSION["username"], "sessionToken": SESSION["sessionToken"]}})) return if operation == "meta": await asyncio.sleep(0.5) payload = {"identity": DASHBOARD["identity"]} elif operation == "ops/dashboard/summary": dashboard_requests += 1 await asyncio.sleep(1.2) payload = DASHBOARD elif operation == "ops/live/versions": payload = {"version": "fixed-v1", "scopeVersions": {}, "refreshedAt": "2026-09-26T06:00:00.000Z", "syncStatus": {"state": "current"}} else: payload = {} await route.fulfill(status=200, content_type="application/json", body=json.dumps({"success": True, "data": payload})) await page.route("**/cloud-functions/**", mock_gateway) try: if from_login: await page.goto(url + "/admin/login", wait_until="commit") await page.locator('input[name="username"]').fill("perf-test") await page.locator('input[name="password"]').fill("test-password") before = await page.evaluate("performance.now()") await page.get_by_role("button", name="登录管理后台").click() await page.get_by_role("heading", name="运营经理，欢迎回来").wait_for(timeout=30000) await page.locator(".trend-stat.courses strong").wait_for(timeout=30000) return {"loginMs": round(await page.evaluate("performance.now()") - before, 1), "dashboardRequests": dashboard_requests} await page.goto(url + "/admin/dashboard", wait_until="commit") await page.get_by_role("heading", name="运营经理，欢迎回来").wait_for(timeout=30000) await page.locator(".trend-stat.courses strong").wait_for(timeout=30000) first_ms = round(await page.evaluate("performance.now()"), 1) await page.locator('a[href="/admin/schedule"]').first.click() await page.get_by_role("heading", name="排课中心", exact=True).wait_for(timeout=30000) before = await page.evaluate("performance.now()") await page.locator('#admin-sidebar a[href="/admin/dashboard"]').last.click() await page.get_by_role("heading", name="运营经理，欢迎回来").wait_for(timeout=30000) await page.locator(".trend-stat.courses strong").wait_for(timeout=30000) returned_ms = round(await page.evaluate("performance.now()") - before, 1) return {"firstMs": first_ms, "returnMs": returned_ms, "dashboardRequests": dashboard_requests} 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("--runs", type=int, default=10) parser.add_argument("--out", type=Path) parser.add_argument("--from-login", action="store_true") args = parser.parse_args() 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.from_login)) finally: await browser.close() finally: for server in servers.values(): server.shutdown() keys = ["loginMs"] if args.from_login else ["firstMs", "returnMs"] summary = {name: {"n": len(items), **{key + "P75": percentile([item[key] for item in items], 75) for key in keys}, "dashboardRequests": max(item["dashboardRequests"] for item in items), "runs": items} for name, items in runs.items()} print(json.dumps({name: {k: v for k, v in item.items() if k != "runs"} 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") if __name__ == "__main__": asyncio.run(main())