xiaoshu-app/perf/bench/admin-frontend.py

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