ykxmz/_build/rp_pc_long.py

# -*- coding: utf-8 -*-
r'''PC 侧 2 小时长测 —— "真实用户"视角。

为什么必须两边同时测：
  设备自己发起的请求 = 设备 4G → 边缘 → 隧道 → 设备 4G
                      ^^^^^^^^^ 第一段           ^^^^^^^^^ 第二段
  两段压在同一个 4G 上，数字偏悲观。

  这台 PC（WiFi/宽带）发起 = PC → 边缘（好网络）→ 隧道 → 设备 4G
                            第一段好            第二段才是 4G

  只有第二段是真实用户也会遇到的。

用法: python _build/rp_pc_long.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 = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote\_build\rp_pc_long.tsv")
EVERY = 15


def hit(timeout: float = 25.0):
    r = urllib.request.Request(BASE + "/v1/health",
                               headers={"User-Agent": UA})
    t0 = time.perf_counter()
    try:
        with urllib.request.urlopen(r, timeout=timeout) as resp:
            resp.read()
            code = resp.status
    except urllib.error.HTTPError as exc:
        try:
            exc.read()
        except Exception:                              # noqa: BLE001
            pass
        code = exc.code
    except Exception:                                  # noqa: BLE001
        code = 0
    return code, int((time.perf_counter() - t0) * 1000)


def pct(vals, q):
    if not vals:
        return 0
    s = sorted(vals)
    return int(s[min(int(len(s) * q), len(s) - 1)])


def main() -> int:
    hours = float(sys.argv[1]) if len(sys.argv) > 1 else 2.0
    dur = int(hours * 3600)
    t0 = time.time()

    print("  PC 侧长测开始: %s  持续 %.1f 小时" %
          (time.strftime("%Y-%m-%d %H:%M:%S"), hours))
    print("  每 %d 秒一次，结果写入 %s" % (EVERY, OUT.name))

    rows = []
    with OUT.open("w", encoding="utf-8") as fh:
        fh.write("time\tcode\tms\n")
        while time.time() - t0 < dur:
            code, ms = hit()
            stamp = time.strftime("%Y-%m-%d %H:%M:%S")
            fh.write("%s\t%d\t%d\n" % (stamp, code, ms))
            fh.flush()
            rows.append((stamp, code, ms))
            time.sleep(EVERY)

    total = len(rows)
    ok = [r for r in rows if r[1] == 200]
    ms = [r[2] for r in ok]

    print()
    print("=" * 58)
    print("  PC 侧长测结束 · %s" % time.strftime("%Y-%m-%d %H:%M:%S"))
    print("=" * 58)
    print("  样本 %d  成功 %d = %.1f%%"
          % (total, len(ok), len(ok) * 100 / total if total else 0))
    if ms:
        print("  耗时 最小 %d  中位 %d  P90 %d  P99 %d  最大 %d ms"
              % (min(ms), int(statistics.median(ms)), pct(ms, 0.9),
                 pct(ms, 0.99), max(ms)))
    bad = [r for r in rows if r[1] != 200]
    if bad:
        print("  失败 %d 次:" % len(bad))
        for stamp, code, m in bad[:40]:
            print("    %s  code=%s  %dms" % (stamp, code, m))
    else:
        print("  ✓ 全程零失败")
    return 0


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