ykxmz/_build/bench_from_pc.py

# -*- coding: utf-8 -*-
r'''从这台 Windows 机器测中转 —— 这才是"真实用户"的视角。

== 为什么要换测量端 ==

之前所有测量都是在设备自己上跑的 curl。那等于：
    设备 4G --> Cloudflare 边缘        （第一段）
    边缘 --> 隧道 --> 设备 4G 源站      （第二段）
**两段都压在同一个 4G 链路上**，量出来的数字是双重打击。

真实用户（手机/电脑在 WiFi、宽带、5G 上）：
    用户 --> Cloudflare 边缘            （第一段，好网络）
    边缘 --> 隧道 --> 设备 4G 源站      （第二段，只有这段是 4G）

所以从这台机器量出来的，才是用户实际会遇到的延迟和可用率。

用法: python _build/bench_from_pc.py [标签]
'''
from __future__ import annotations

import json
import statistics
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path

BASE = "https://relay.xxmxxs.top"
UA = "LanRemote/1.8.2"
OUT_DIR = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote\_build")


def req(path: str, method: str = "GET", data: bytes | None = None,
        headers: dict | None = None, timeout: float = 150.0):
    """发一个请求，返回 (状态码, 字节数, 秒数, 响应头)。"""
    hdrs = {"User-Agent": UA}
    hdrs.update(headers or {})
    r = urllib.request.Request(BASE + path, data=data, headers=hdrs,
                               method=method)
    t0 = time.perf_counter()
    try:
        with urllib.request.urlopen(r, timeout=timeout) as resp:
            body = resp.read()
            return resp.status, len(body), time.perf_counter() - t0, resp.headers
    except urllib.error.HTTPError as exc:
        try:
            exc.read()
        except Exception:                              # noqa: BLE001
            pass
        return exc.code, 0, time.perf_counter() - t0, exc.headers
    except Exception:                                  # noqa: BLE001
        return 0, 0, time.perf_counter() - t0, {}


def pct(values: list, q: float) -> int:
    if not values:
        return 0
    s = sorted(values)
    idx = min(int(len(s) * q), len(s) - 1)
    return int(s[idx])


def main() -> int:
    label = sys.argv[1] if len(sys.argv) > 1 else "pc"
    lines: list[str] = []

    def say(text: str = "") -> None:
        print(text)
        lines.append(text)

    say("=" * 62)
    say("  从 PC 量中转 · %s · %s" % (label, time.strftime("%Y-%m-%d %H:%M:%S")))
    say("=" * 62)

    # ---------------- A. 公网可用率 ----------------
    say()
    say("── A. 公网可用率（100 次 /v1/health）──")
    ok, times, fails = 0, [], 0
    for i in range(100):
        code, _n, secs, _h = req("/v1/health", timeout=25)
        if code == 200:
            ok += 1
            times.append(int(secs * 1000))
        else:
            fails += 1
        time.sleep(0.3)
    say("  可用率 %d/100 = %d%%   失败 %d 次" % (ok, ok, fails))
    if times:
        say("  耗时 最小 %d  中位 %d  P90 %d  P99 %d  最大 %d ms"
            % (min(times), int(statistics.median(times)), pct(times, 0.9),
               pct(times, 0.99), max(times)))

    # ---------------- E. 静态资源 ----------------
    say()
    say("── E. 静态资源（各 10 次）──")
    for path in ("/", "/shot/pc-main.png", "/favicon.ico"):
        o, ts = 0, []
        for _ in range(10):
            code, _n, secs, _h = req(path, timeout=25)
            if code == 200:
                o += 1
                ts.append(int(secs * 1000))
            time.sleep(0.2)
        med = int(statistics.median(ts)) if ts else 0
        say("  %-22s %2d/10   中位 %d  最大 %d ms"
            % (path, o, med, max(ts) if ts else 0))

    # ---------------- C. 下载 ----------------
    say()
    say("── C. 三个成品下载（完整下载）──")
    dl_results = {}
    for name in ("ykxmz-setup-1.8.2.exe", "ykxmz-server-1.8.2.zip",
                 "ykxmz-watch-1.8.2.apk"):
        code, size, secs, hdrs = req("/download/" + name, timeout=300)
        kbps = (size / secs / 1024) if secs > 0 else 0
        cf = (hdrs.get("cf-cache-status") or "?") if hdrs else "?"
        say("  %-30s HTTP %-4s %9d 字节  %6.1fs  %7.0f KB/s  cf=%s"
            % (name, code, size, secs, kbps, cf))
        if code == 200 and size > 0:
            dl_results[name] = kbps

    # ---------------- B. 长轮询 ----------------
    say()
    say("── B. 长轮询（6 次真实 poll，每次挂 60 秒）──")
    stamp = int(time.time())
    dev = "pcbench-%s-%d" % (label, stamp)
    code6 = "%06d" % (stamp % 1000000)
    payload = json.dumps({
        "deviceId": dev, "code": code6, "name": "pcbench",
        "version": "1.8.2",
    }, ensure_ascii=False).encode("utf-8")
    status, _n, _s, _h = req("/v1/agent/register", method="POST",
                             data=payload, timeout=30,
                             headers={"Content-Type": "application/json"})
    token = ""
    if status == 200:
        # 重新取一次拿 body
        r = urllib.request.Request(
            BASE + "/v1/agent/register", data=payload,
            headers={"User-Agent": UA, "Content-Type": "application/json"},
            method="POST")
        try:
            with urllib.request.urlopen(r, timeout=30) as resp:
                token = json.loads(resp.read().decode("utf-8")).get("token", "")
        except Exception:                              # noqa: BLE001
            pass
    say("  探针注册: %s" % ("成功" if token else "失败"))

    if token:
        okb, btimes, bfails = 0, [], []
        for i in range(6):
            code, _n, secs, _h = req(
                "/v1/agent/poll", method="POST", data=b"",
                headers={"X-Token": token}, timeout=150)
            ms = int(secs * 1000)
            if code == 200:
                okb += 1
                btimes.append(ms)
                say("    %d: 200  %6dms" % (i + 1, ms))
            else:
                bfails.append("%d:%s" % (i + 1, code))
                say("    %d: %-4s **失败**" % (i + 1, code))
        say("  长轮询成功 %d/6 = %d%%   失败: %s"
            % (okb, okb * 100 // 6, " ".join(bfails) or "无"))
        if btimes:
            say("  耗时 最小 %d  中位 %d  P90 %d  最大 %d ms（理想 ≈ 61000）"
                % (min(btimes), int(statistics.median(btimes)),
                   pct(btimes, 0.9), max(btimes)))

    say()
    say("=" * 62)
    out = OUT_DIR / ("bench_pc_%s.txt" % label)
    out.write_text("\n".join(lines), encoding="utf-8")
    say("  存档: %s" % out)
    return 0


if __name__ == "__main__":
    sys.exit(main())