xiaoshu-app/perf/bench/admin-dashboard-return.py

#!/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())