[Python] 纯文本查看 复制代码
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
netspeed_guiV2.py —— 多源并发 · 多线程循环下载测速（无控制台窗口 + 图形界面）
设计要点
* 多线程分段下载（HTTP Range），读到的数据直接丢弃：
不落盘、不合并、不占用磁盘，测出来的就是纯网络下行速度。
* 连续循环：一个线程持续把「分段任务」投进队列，N 个工作线程不停取任务；
一轮下完立刻接下一轮，段与段之间没有等待，读数更平稳。
* 实时统计：当前速度、平均速度、峰值、上一轮速度、累计下载量、完成轮数、
运行时长、错误数，以及一条滚动的速度曲线。
* 保存为 .pyw，双击运行不会弹出黑色控制台窗口。
多源并发（V2 新增）
* 节点列表第一列是复选框，打钩可同时选中任意多个测速源；点「开始测速」后
所有打钩的源【同时】下载，速度累加，用来把带宽顶到极限。
* 每个源各自使用「线程数」指定的线程数（勾 5 个源 × 16 线程 = 80 条连接），
界面会实时显示总连接数，便于控制内存占用（实测约 0.7~1 MB/线程）。
* 某个源连不上（403/418 拦截、DNS 失败、404 等）只把该源标记为失败并跳过，
不影响其余源继续测速；全部失败时才会报错。
* 各源文件大小不同，「完成轮数」已取消：顶部大数字只报【合计速度】与
【累计下载量】，每个源自己的速度和本轮进度显示在列表里。
* 一个源都不打钩时，退回原来的单源模式，用「下载地址」框里那一个地址测速。
测速节点管理（本版新增）
* 「测速节点」列表统一管理三类节点：
- 预设 ：首次运行自动写入的内置测速源（来自《测速源清单.md》）
- 手动 ：自己新建并永久保存的节点，不受「清空历史」影响
- 历史 ：每次成功开始测速后自动记录，可一键清空
* 节点连同使用次数、最佳/最近速度、文件大小一起持久化。
* 配置文件为本文件同目录下的 netspeed_nodes.json，任何改动即时写盘；
没有配置文件时自动生成，损坏时自动备份为 .bak 并重建。
运行方式
双击本文件即可；或命令行:
pythonw netspeed_guiV1.pyw
需要 Python 自带 tkinter（Windows 官方版和大多数发行版都自带）。
注意
速度单位 Mbps = 10^6 bit/s（运营商口径），括号内是 MB/s（MiB/s）。
「当前速度」是 200ms 瞬时采样，抖动是正常的；「平均速度」最能代表实际带宽。
"""
from __future__ import annotations
import collections
import datetime
import json
import math
import os
import queue
import sys
import threading
import time
import tkinter as tk
import urllib.parse
import urllib.request
from tkinter import messagebox, ttk
DEFAULT_URL = "https://download.deepseek.com/desktop/dsh-latest-windows-x64.exe"
UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36")
READ_SIZE = 256 * 1024 # 每次 socket 读取块大小
MIN_SEGMENT = 1024 * 1024 # 每段最小 1 MiB，避免分段过碎
SAMPLE_MS = 200 # 界面刷新 / 速度采样间隔（毫秒）
HISTORY = 200 # 曲线保留的采样点数（200 × 200ms = 40 秒）
FONT_UI = "Microsoft YaHei UI"
FONT_NUM = "Segoe UI"
CONFIG_NAME = "netspeed_nodes.json"
CONFIG_VERSION = 2 # 2 起：新增 selected_urls（多源勾选状态）
KIND_LABEL = {"manual": "手动", "preset": "预设", "history": "历史"}
KIND_ORDER = {"manual": 0, "preset": 1, "history": 2}
# 首次运行写入的内置测速源（取自同目录《测速源清单.md》的实测结论）
PRESET_NODES = (
{"name": "DeepSeek 桌面客户端",
"url": "https://download.deepseek.com/desktop/dsh-latest-windows-x64.exe",
"note": "276 MB · latest 永久有效"},
{"name": "CUDA Toolkit 12.6.3",
"url": "https://developer.download.nvidia.cn/compute/cuda/12.6.3/local_installers/"
"cuda_12.6.3_561.17_windows.exe",
"note": "3.0 GB · 含版本号，适合长时间测"},
{"name": "微信 Windows x64",
"url": "https://dldir1v6.qq.com/weixin/Windows/WeChatSetup.exe",
"note": "270 MB · 腾讯 CDN，永久有效"},
{"name": "PyTorch 2.5.0+cu124",
"url": "https://download.pytorch.org/whl/cu124/"
"torch-2.5.0%2Bcu124-cp311-cp311-win_amd64.whl",
"note": "2.4 GB · CloudFront，含版本号"},
{"name": "Anaconda3 2025.06",
"url": "https://repo.anaconda.com/archive/Anaconda3-2025.06-0-Windows-x86_64.exe",
"note": "914 MB · 含版本号"},
{"name": "Docker Desktop",
"url": "https://desktop.docker.com/win/main/amd64/Docker%20Desktop%20Installer.exe",
"note": "599 MB · 永久有效，偶发失败"},
)
# 用完整的浏览器请求头，避免被镜像站/云厂商的 WAF 当成爬虫拦截（常见 403 / 418）
BASE_HEADERS = {
"User-Agent": UA,
"Accept": "*/*",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Accept-Encoding": "identity",
}
def req_headers(**extra: str) -> dict:
h = dict(BASE_HEADERS)
h.update(extra)
return h
def is_http_url(url: str) -> bool:
return url.strip().lower().startswith(("http://", "https://"))
def auto_name(url: str) -> str:
"""给自动记录的历史节点起一个好读的名字：主机名 · 文件名。"""
try:
parts = urllib.parse.urlsplit(url)
base = os.path.basename(urllib.parse.unquote(parts.path)) or parts.netloc
name = f"{parts.netloc} · {base}"
return name if len(name) <= 70 else name[:67] + "..."
except Exception: # noqa: BLE001
return url
# --------------------------------------------------------------------------- #
# 格式化
# --------------------------------------------------------------------------- #
def human_bytes(n: float) -> str:
for unit in ("B", "KiB", "MiB", "GiB", "TiB"):
if abs(n) < 1024 or unit == "TiB":
return f"{n:.0f} B" if unit == "B" else f"{n:.2f} {unit}"
n /= 1024
return f"{n:.2f} TiB"
def mbps(bps: float) -> float:
return bps * 8 / 1e6
def speed_text(bps: float) -> str:
return f"{mbps(bps):.1f} Mbps ({bps / 1024 / 1024:.1f} MB/s)"
def human_time(sec: float) -> str:
sec = int(sec)
if sec < 60:
return f"{sec} 秒"
if sec < 3600:
return f"{sec // 60} 分 {sec % 60} 秒"
return f"{sec // 3600} 时 {(sec % 3600) // 60} 分"
def int_var(var: tk.IntVar, default: int, lo: int, hi: int) -> int:
"""读取 Spinbox/Entry 的整数；输入框被清空或填了非法值时退回默认值。
.pyw 没有控制台，IntVar.get() 抛出的 TclError 会静默消失，所以必须兜住。
"""
try:
return max(lo, min(hi, int(var.get())))
except Exception: # noqa: BLE001
return default
def nice_max(v: float) -> float:
"""把曲线纵轴上限调成 1/2/2.5/5/10 × 10^k，刻度好看且不会溢出。"""
if v <= 0:
return 10.0
base = 10.0 ** math.floor(math.log10(v))
for m in (1.0, 2.0, 2.5, 5.0, 10.0):
if v <= m * base:
return m * base
return 10.0 * base
def now_text() -> str:
return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# --------------------------------------------------------------------------- #
# 节点配置持久化
# --------------------------------------------------------------------------- #
def config_dir() -> str:
"""配置文件所在目录：优先脚本所在目录（双击运行时就是「当前文件夹」）。"""
# 打包成单文件 exe 后，__file__ 指向临时解包目录（_MEIxxxx，每次运行都不同），
# 配置写到那里会在退出时被清掉。冻结运行时固定用 exe 自身所在目录。
if getattr(sys, "frozen", False):
return os.path.dirname(os.path.abspath(sys.executable))
for candidate in (globals().get("__file__"), sys.argv[0] if sys.argv else None):
if not candidate:
continue
try:
path = os.path.abspath(candidate)
except Exception: # noqa: BLE001
continue
if os.path.isfile(path):
return os.path.dirname(path)
return os.getcwd()
def config_path() -> str:
return os.path.join(config_dir(), CONFIG_NAME)
class NodeStore:
"""测速节点集合 + JSON 配置文件的读写（原子写，防止写坏）。"""
def __init__(self, path: str) -> None:
self.path = path
self.nodes: list[dict] = []
self.selected_urls: list[str] = [] # 多源并发：被勾选的节点地址
self.current_url = DEFAULT_URL
self.load()
# ---------------- 读写 ---------------- #
def load(self) -> None:
raw = None
if os.path.isfile(self.path):
try:
with open(self.path, "r", encoding="utf-8") as fh:
raw = json.load(fh)
except Exception: # noqa: BLE001
# 配置损坏：备份后按首次运行处理，绝不因为配置文件而启动失败
try:
os.replace(self.path, self.path + ".bak")
except Exception: # noqa: BLE001
pass
raw = None
loaded = isinstance(raw, dict) and isinstance(raw.get("nodes"), list)
if loaded:
# 手工编辑过的配置可能含重复地址；Treeview 用 URL 当 iid，必须去重
seen: set[str] = set()
for item in raw["nodes"]:
node = self._coerce(item)
if node and node["url"] not in seen:
seen.add(node["url"])
self.nodes.append(node)
cur = str(raw.get("current_url") or "").strip()
self.current_url = cur if is_http_url(cur) else DEFAULT_URL
# 勾选状态：只保留确实存在的节点，配置被手工改坏也不会出错
sel = raw.get("selected_urls")
if isinstance(sel, list):
known = {n["url"] for n in self.nodes}
self.selected_urls = [str(u) for u in sel
if isinstance(u, str) and u in known]
return
# 首次运行（或配置损坏）：写入预设节点，并立刻落盘
self.current_url = DEFAULT_URL
self.selected_urls = []
for preset in PRESET_NODES:
node = self.blank(preset["name"], preset["url"], "preset", preset.get("note", ""))
self.nodes.append(node)
try:
self.save()
except Exception: # noqa: BLE001
pass
def save(self) -> None:
payload = {
"version": CONFIG_VERSION,
"updated_at": now_text(),
"current_url": self.current_url,
"selected_urls": self.selected_urls,
"nodes": self.nodes,
}
tmp = self.path + ".tmp"
with open(tmp, "w", encoding="utf-8") as fh:
json.dump(payload, fh, ensure_ascii=False, indent=2)
os.replace(tmp, self.path)
# ---------------- 记录 ---------------- #
@staticmethod
def blank(name: str, url: str, kind: str, note: str = "") -> dict:
return {
"name": name.strip() or url, "url": url.strip(), "kind": kind,
"note": note or "", "created_at": now_text(), "last_used_at": "",
"uses": 0, "best_mbps": 0.0, "last_mbps": 0.0, "size": 0,
}
@staticmethod
def _coerce(raw: object) -> dict | None:
if not isinstance(raw, dict):
return None
url = str(raw.get("url") or "").strip()
if not is_http_url(url):
return None
kind = str(raw.get("kind") or "manual")
if kind not in KIND_LABEL:
kind = "manual"
def as_int(key: str) -> int:
try:
return max(0, int(raw.get(key) or 0))
except Exception: # noqa: BLE001
return 0
def as_float(key: str) -> float:
try:
return max(0.0, float(raw.get(key) or 0.0))
except Exception: # noqa: BLE001
return 0.0
return {
"name": str(raw.get("name") or "").strip() or url,
"url": url, "kind": kind, "note": str(raw.get("note") or ""),
"created_at": str(raw.get("created_at") or now_text()),
"last_used_at": str(raw.get("last_used_at") or ""),
"uses": as_int("uses"), "best_mbps": as_float("best_mbps"),
"last_mbps": as_float("last_mbps"), "size": as_int("size"),
}
# ---------------- 查询 ---------------- #
def find(self, url: str) -> dict | None:
target = (url or "").strip()
for node in self.nodes:
if node["url"] == target:
return node
return None
def sorted_nodes(self) -> list[dict]:
return sorted(
self.nodes,
key=lambda n: (KIND_ORDER.get(n["kind"], 9), -(n["uses"] or 0),
(n["name"] or "").lower()),
)
def history_count(self) -> int:
return sum(1 for n in self.nodes if n["kind"] == "history")
# ---------------- 变更 ---------------- #
def add(self, name: str, url: str, note: str, kind: str) -> dict | None:
"""新增节点；地址已存在时返回 None（由调用方决定是否提示）。"""
if self.find(url) is not None:
return None
node = self.blank(name, url, kind, note)
self.nodes.append(node)
return node
def touch(self, url: str, size: int = 0) -> dict:
"""一次成功启动 = 使用一次；没见过的地址自动记入历史。"""
node = self.find(url)
if node is None:
node = self.blank(auto_name(url), url, "history")
self.nodes.append(node)
node["uses"] = int(node.get("uses") or 0) + 1
node["last_used_at"] = now_text()
if size:
node["size"] = int(size)
return node
def record_result(self, url: str, mbps_avg: float) -> None:
node = self.find(url)
if node is None:
return
value = round(float(mbps_avg), 1)
node["last_mbps"] = value
if value > float(node.get("best_mbps") or 0.0):
node["best_mbps"] = value
def clear_history(self) -> int:
before = len(self.nodes)
self.nodes = [n for n in self.nodes if n["kind"] != "history"]
self.prune_selected()
return before - len(self.nodes)
# ---------------- 多源勾选 ---------------- #
def prune_selected(self) -> None:
"""把已经不存在的节点从勾选集合里剔除。"""
known = {n["url"] for n in self.nodes}
self.selected_urls = [u for u in self.selected_urls if u in known]
def is_selected(self, url: str) -> bool:
return url in self.selected_urls
def toggle_selected(self, url: str) -> bool:
"""切换勾选状态，返回切换后的状态（节点不存在时返回 False）。"""
if url in self.selected_urls:
self.selected_urls.remove(url)
return False
if self.find(url) is None:
return False
self.selected_urls.append(url)
return True
def selected_nodes(self) -> list[dict]:
"""按列表的显示顺序返回被勾选的节点。"""
return [n for n in self.sorted_nodes() if n["url"] in self.selected_urls]
def set_selected(self, urls: list[str]) -> None:
known = {n["url"] for n in self.nodes}
self.selected_urls = [u for u in urls if u in known]
# --------------------------------------------------------------------------- #
# 目标探测
# --------------------------------------------------------------------------- #
def preflight(url: str, timeout: float) -> dict:
"""用一个 Range: bytes=0-0 请求探测文件大小和是否支持分段（不读 body）。"""
req = urllib.request.Request(url, headers=req_headers(Range="bytes=0-0"))
with urllib.request.urlopen(req, timeout=timeout) as resp:
status = resp.status
ctype = resp.headers.get("Content-Type") or ""
content_range = resp.headers.get("Content-Range")
content_length = resp.headers.get("Content-Length")
ranges = status == 206 and bool(content_range)
total = None
if ranges and content_range and "/" in content_range:
tail = content_range.rsplit("/", 1)[-1]
if tail.isdigit():
total = int(tail)
if total is None and content_length and content_length.isdigit():
total = int(content_length)
return {"ranges": ranges, "total": total, "ctype": ctype}
# --------------------------------------------------------------------------- #
# 测速引擎：N 个工作线程 + 1 个分段投递线程，数据全部丢弃
# --------------------------------------------------------------------------- #
class RangeUnsupported(Exception):
pass
class Engine:
def __init__(self, url: str, info: dict, threads: int,
timeout: float = 8.0, retries: int = 3) -> None:
self.url = url
self.info = info
self.total = info.get("total")
self.timeout = timeout
self.retries = retries
self.use_range = bool(info.get("ranges") and self.total)
self._stop = threading.Event()
self._lock = threading.Lock()
self._bytes = 0
self._segments = 0
self._errors = 0
self._last_error = ""
if self.use_range:
# 分段不细于 MIN_SEGMENT，避免小文件开太多线程
self.nseg = max(1, min(threads, (self.total + MIN_SEGMENT - 1) // MIN_SEGMENT))
self.per = (self.total + self.nseg - 1) // self.nseg
else:
# 不支持分段（或不知道大小）：退化为单连接顺序下载
self.nseg = 1
self.per = 0
self.threads = self.nseg
self.running = False
self.t_start = 0.0
self._queue: queue.Queue = queue.Queue(maxsize=max(4, self.nseg * 2))
self._workers: list[threading.Thread] = []
self._coord: threading.Thread | None = None
self._reaper: threading.Thread | None = None
# ---------------- 对外接口 ---------------- #
def start(self) -> None:
self.running = True
self.t_start = time.perf_counter()
self._coord = threading.Thread(target=self._coordinator, daemon=True)
self._coord.start()
self._workers = [threading.Thread(target=self._worker, daemon=True)
for _ in range(self.threads)]
for t in self._workers:
t.start()
def stop(self) -> None:
"""请求停止。
这里立刻把 running 置 False，让界面即时响应（按钮恢复、「已停止」），
线程回收丢给后台线程慢慢做：worker 在 256KiB 粒度上检查停止标志，通常
毫秒级退出；但如果它正好卡在建立新连接上，最坏要等 socket 超时才返回
（实测过 1.2 秒仍未 join 完），不能让界面一直等它。
"""
self.running = False
self._stop.set()
if self._reaper is None or not self._reaper.is_alive():
self._reaper = threading.Thread(target=self._reap, daemon=True)
self._reaper.start()
def _reap(self) -> None:
for t in self._workers:
t.join(timeout=self.timeout + 5)
if self._coord is not None:
self._coord.join(timeout=2)
self.running = False # 兜底，保证最终状态一致
def snapshot(self) -> dict:
with self._lock:
return {"bytes": self._bytes, "segments": self._segments,
"errors": self._errors, "last_error": self._last_error}
def elapsed(self) -> float:
return time.perf_counter() - self.t_start if self.t_start else 0.0
def avg_bps(self) -> float:
elapsed = self.elapsed()
return self.snapshot()["bytes"] / elapsed if elapsed > 0 else 0.0
# ---------------- 内部 ---------------- #
def _add_bytes(self, n: int) -> None:
with self._lock:
self._bytes += n
def _error(self, msg: str) -> None:
with self._lock:
self._errors += 1
self._last_error = msg
def _coordinator(self) -> None:
"""不停地按轮投递分段任务；队列满就等，保证工作线程不空转也不积压。"""
while not self._stop.is_set():
for i in range(self.nseg):
if self._stop.is_set():
return
if self.use_range:
start = i * self.per
end = min(self.total - 1, start + self.per - 1)
if start > end:
continue
else:
start, end = 0, None
while not self._stop.is_set():
try:
self._queue.put((start, end), timeout=0.2)
break
except queue.Full:
continue
def _worker(self) -> None:
while not self._stop.is_set():
try:
start, end = self._queue.get(timeout=0.2)
except queue.Empty:
continue
try:
self._fetch(start, end)
finally:
with self._lock:
self._segments += 1
self._queue.task_done()
def _fetch(self, start: int, end: int | None) -> None:
"""下载并丢弃 [start, end]；失败时从已收位置续传重试。"""
pos = start
attempt = 0
while not self._stop.is_set():
attempt += 1
try:
h = req_headers()
if end is not None:
h["Range"] = f"bytes={pos}-{end}"
req = urllib.request.Request(self.url, headers=h)
with urllib.request.urlopen(req, timeout=self.timeout) as resp:
if resp.status not in (200, 206):
raise RuntimeError(f"HTTP {resp.status}")
if resp.status == 200 and end is not None and pos != 0:
raise RangeUnsupported()
while not self._stop.is_set():
if end is None:
want = READ_SIZE
else:
want = min(READ_SIZE, end - pos + 1)
if want <= 0:
break
buf = resp.read(want)
if not buf:
break
if end is not None:
pos += len(buf)
self._add_bytes(len(buf))
# 到这儿连接已正常结束，判断这一段是否收全
if end is None:
# 不支持分段：读到 EOF 即算完成；若能拿到总大小则校验
if not self.total or pos >= self.total:
return
self._error("顺序下载未收全（服务端不支持分段续传）")
return
if pos > end:
return
if attempt > self.retries:
self._error(f"分段 {start}-{end} 重试 {self.retries} 次仍未收全")
return
except RangeUnsupported:
self._error("服务端返回完整文件而非分段，已停止")
self._stop.set()
return
except Exception as exc: # noqa: BLE001
if self._stop.is_set():
return
if attempt > self.retries:
self._error(f"{type(exc).__name__}: {exc}")
return
time.sleep(min(2 ** (attempt - 1), 3))
# --------------------------------------------------------------------------- #
# 界面
# --------------------------------------------------------------------------- #
class App(tk.Tk):
def __init__(self) -> None:
super().__init__()
self.title("网络下载测速 · 多线程循环（不落盘）")
self.geometry("1120x900")
self.minsize(1000, 820)
self.store = NodeStore(config_path())
self.engines: list[Engine] = [] # 多源并发：每个源一个独立引擎
self.msg_q: queue.Queue = queue.Queue()
self.hist: collections.deque = collections.deque(maxlen=HISTORY)
self.peak = 0.0
self.prev_bytes = 0
self.prev_t = 0.0
self.seen_errors = 0
# 每个源的实时状态（键为 url）
self.live_speed: dict[str, float] = {} # 当前速度 bps
self.live_prog: dict[str, float] = {} # 本轮进度 0~1
self.live_prev: dict[str, tuple[int, float]] = {} # 上次采样 (字节, 时刻)
self.node_status: dict[str, str] = {} # 失败原因（空 = 正常）
self.session_engines: list[Engine] = []
self.session_done = True
self._build()
self._refresh_nodes()
self.url_var.set(self.store.current_url)
self.url_box.set(self.store.current_url)
self._refresh_url_info()
self.protocol("WM_DELETE_WINDOW", self._on_close)
self.after(SAMPLE_MS, self._tick)
# ---------------- 构建界面 ---------------- #
def _build(self) -> None:
style = ttk.Style(self)
try:
style.theme_use("vista")
except tk.TclError:
pass
style.configure("Big.TLabel", font=(FONT_NUM, 30, "bold"), foreground="#1565c0")
style.configure("Stat.TLabel", font=(FONT_UI, 10))
style.configure("Val.TLabel", font=(FONT_NUM, 12, "bold"))
style.configure("Hint.TLabel", font=(FONT_UI, 9), foreground="#777")
root = ttk.Frame(self, padding=10)
root.pack(fill="both", expand=True)
root.columnconfigure(0, weight=1)
# --- 配置区 ---
cfg = ttk.LabelFrame(root, text=" 测速目标 ", padding=10)
cfg.grid(row=0, column=0, sticky="ew")
cfg.columnconfigure(1, weight=1)
ttk.Label(cfg, text="下载地址", style="Stat.TLabel").grid(row=0, column=0, sticky="w")
self.url_var = tk.StringVar(value=DEFAULT_URL)
self.url_box = ttk.Combobox(cfg, textvariable=self.url_var, values=[])
self.url_box.grid(row=0, column=1, columnspan=5, sticky="ew", padx=(8, 0))
self.url_box.bind("<<ComboboxSelected>>", lambda _e: self._refresh_url_info())
self.url_box.bind("<KeyRelease>", lambda _e: self._refresh_url_info())
self.url_box.bind("<Return>", lambda _e: self._on_start())
self.node_info_var = tk.StringVar(value="")
ttk.Label(cfg, textvariable=self.node_info_var, style="Hint.TLabel").grid(
row=1, column=0, columnspan=6, sticky="w", pady=(4, 0))
ttk.Label(cfg, text="线程数", style="Stat.TLabel").grid(
row=2, column=0, sticky="w", pady=(8, 0))
self.threads_var = tk.IntVar(value=16)
ttk.Spinbox(cfg, from_=1, to=128, width=6, textvariable=self.threads_var,
command=self._update_sel_hint).grid(
row=2, column=1, sticky="w", padx=(8, 0), pady=(8, 0))
ttk.Label(cfg, text="超时(秒)", style="Stat.TLabel").grid(
row=2, column=2, sticky="e", pady=(8, 0))
self.timeout_var = tk.IntVar(value=8)
ttk.Spinbox(cfg, from_=3, to=60, width=6, textvariable=self.timeout_var).grid(
row=2, column=3, sticky="w", padx=(8, 0), pady=(8, 0))
ttk.Label(cfg, text="重试", style="Stat.TLabel").grid(
row=2, column=4, sticky="e", pady=(8, 0))
self.retries_var = tk.IntVar(value=3)
ttk.Spinbox(cfg, from_=0, to=10, width=6, textvariable=self.retries_var).grid(
row=2, column=5, sticky="w", padx=(8, 0), pady=(8, 0))
# --- 节点管理区 ---
nodes = ttk.LabelFrame(root, text=" 测速节点（手动保存 / 自动历史） ", padding=8)
nodes.grid(row=1, column=0, sticky="ew", pady=(10, 0))
nodes.columnconfigure(0, weight=1)
columns = ("sel", "kind", "name", "cur", "prog", "size",
"best", "last", "uses", "url", "used_at", "note")
headings = (("sel", "选", 34, False), ("kind", "来源", 52, False),
("name", "名称", 170, False), ("cur", "当前Mbps", 84, False),
("prog", "进度", 54, False), ("size", "大小", 76, False),
("best", "最佳Mbps", 76, False), ("last", "最近Mbps", 76, False),
("uses", "次数", 46, False), ("url", "地址", 240, True),
("used_at", "最近使用", 130, False), ("note", "备注", 140, True))
self.tree = ttk.Treeview(nodes, columns=columns, show="headings",
height=6, selectmode="browse")
for key, text, width, stretch in headings:
self.tree.heading(key, text=text)
self.tree.column(key, width=width, minwidth=40, anchor="w", stretch=stretch)
self.tree.grid(row=0, column=0, sticky="ew")
vsb = ttk.Scrollbar(nodes, orient="vertical", command=self.tree.yview)
vsb.grid(row=0, column=1, sticky="ns")
hsb = ttk.Scrollbar(nodes, orient="horizontal", command=self.tree.xview)
hsb.grid(row=1, column=0, sticky="ew")
self.tree.configure(yscrollcommand=vsb.set, xscrollcommand=hsb.set)
self.tree.bind("<Double-1>", lambda _e: self._use_selected(silent=True))
# 复选框：只拦截「选」这一列的点击，其余列的选中/编辑行为完全不变
self.tree.bind("<Button-1>", self._on_tree_click)
self.tree.tag_configure("failed", foreground="#c62828")
self.tree.tag_configure("running", foreground="#1565c0")
bar = ttk.Frame(nodes)
bar.grid(row=2, column=0, columnspan=2, sticky="ew", pady=(8, 0))
ttk.Button(bar, text="新建节点", command=self._dlg_new).pack(side="left")
ttk.Button(bar, text="编辑", command=self._dlg_edit).pack(side="left", padx=(6, 0))
ttk.Button(bar, text="删除", command=self._del_selected).pack(side="left", padx=(6, 0))
ttk.Button(bar, text="设为当前", command=self._use_selected).pack(side="left", padx=(6, 0))
ttk.Button(bar, text="检测可用性", command=self._check_selected).pack(
side="left", padx=(6, 0))
ttk.Button(bar, text="清空历史", command=self._clear_history).pack(
side="left", padx=(6, 0))
ttk.Button(bar, text="打开配置目录", command=self._open_config_dir).pack(side="right")
ttk.Button(bar, text="清空勾选", command=self._clear_sel).pack(
side="right", padx=(0, 6))
ttk.Button(bar, text="全选", command=self._select_all).pack(
side="right", padx=(0, 6))
self.cfg_hint_var = tk.StringVar(value=f"配置文件：{self.store.path}")
ttk.Label(nodes, textvariable=self.cfg_hint_var, style="Hint.TLabel").grid(
row=3, column=0, columnspan=2, sticky="w", pady=(6, 0))
self.sel_hint_var = tk.StringVar(value="")
ttk.Label(nodes, textvariable=self.sel_hint_var, style="Hint.TLabel").grid(
row=4, column=0, columnspan=2, sticky="w", pady=(2, 0))
# --- 控制区 ---
ctl = ttk.Frame(root)
ctl.grid(row=2, column=0, sticky="ew", pady=(10, 0))
self.btn_start = ttk.Button(ctl, text="开始测速", command=self._on_start)
self.btn_start.pack(side="left")
self.btn_stop = ttk.Button(ctl, text="停止", command=self._on_stop, state="disabled")
self.btn_stop.pack(side="left", padx=(8, 0))
self.status_var = tk.StringVar(value="就绪")
ttk.Label(ctl, textvariable=self.status_var, style="Hint.TLabel").pack(
side="left", padx=(14, 0))
# --- 速度区 ---
box = ttk.LabelFrame(root, text=" 实时速度 ", padding=12)
box.grid(row=3, column=0, sticky="ew", pady=(10, 0))
box.columnconfigure(0, weight=1)
self.now_var = tk.StringVar(value="0.0 Mbps (0.0 MB/s)")
ttk.Label(box, textvariable=self.now_var, style="Big.TLabel").grid(
row=0, column=0, columnspan=4, sticky="w")
ttk.Label(box, text="当前速度", style="Hint.TLabel").grid(
row=1, column=0, columnspan=4, sticky="w", pady=(0, 8))
stats = ttk.Frame(box)
stats.grid(row=2, column=0, columnspan=4, sticky="ew")
for c in range(4):
stats.columnconfigure(c, weight=1, minsize=150)
self.stat_vars: dict[str, tk.StringVar] = {}
fields = [("avg", "合计平均速度"), ("peak", "合计峰值"), ("total", "累计下载"),
("uptime", "运行时长"), ("sources", "活动源数"), ("conns", "总连接数"),
("errors", "错误数"), ("segspeed", "累计分段")]
for i, (key, label) in enumerate(fields):
r, c = divmod(i, 4)
cell = ttk.Frame(stats)
cell.grid(row=r, column=c, sticky="w", pady=(6, 0))
ttk.Label(cell, text=label, style="Hint.TLabel").pack(anchor="w")
var = tk.StringVar(value="—")
self.stat_vars[key] = var
ttk.Label(cell, textvariable=var, style="Val.TLabel").pack(anchor="w")
prog = ttk.Frame(box)
prog.grid(row=3, column=0, columnspan=4, sticky="ew", pady=(12, 0))
prog.columnconfigure(0, weight=1)
self.prog_var = tk.DoubleVar(value=0.0)
ttk.Progressbar(prog, variable=self.prog_var, maximum=100).grid(
row=0, column=0, sticky="ew")
self.round_var = tk.StringVar(value="等待开始")
ttk.Label(prog, textvariable=self.round_var, style="Hint.TLabel").grid(
row=1, column=0, sticky="w", pady=(4, 0))
# --- 曲线区 ---
graph = ttk.LabelFrame(root, text=" 速度曲线（最近 40 秒） ", padding=6)
graph.grid(row=4, column=0, sticky="nsew", pady=(10, 0))
root.rowconfigure(4, weight=1)
graph.columnconfigure(0, weight=1)
graph.rowconfigure(0, weight=1)
self.canvas = tk.Canvas(graph, height=150, background="#fafafa",
highlightthickness=1, highlightbackground="#dddddd")
self.canvas.grid(row=0, column=0, sticky="nsew")
self.err_var = tk.StringVar(value="")
ttk.Label(root, textvariable=self.err_var, style="Hint.TLabel",
wraplength=930, justify="left").grid(
row=5, column=0, sticky="w", pady=(8, 0))
# ---------------- 节点列表 ---------------- #
def _after_change(self, quiet: bool = False) -> None:
"""任何节点改动之后：先落盘，再刷新界面。"""
try:
self.store.save()
self.cfg_hint_var.set(f"配置文件：{self.store.path}（改动已自动保存）")
except Exception as exc: # noqa: BLE001
text = f"{type(exc).__name__}: {exc}"
if quiet:
self.cfg_hint_var.set(f"⚠ 保存失败：{text}")
else:
messagebox.showerror(
"保存失败", f"无法写入配置文件：\n{self.store.path}\n\n{text}")
self._refresh_nodes()
def _row_values(self, node: dict) -> tuple:
"""一行 Treeview 的显示内容：勾选框 + 实时速度/进度 + 持久化统计。"""
url = node["url"]
if self.node_status.get(url):
cur, prog = "失败", "—"
else:
bps = self.live_speed.get(url) or 0.0
cur = f"{mbps(bps):.1f}" if bps else "—"
prog = f"{self.live_prog.get(url, 0.0) * 100:.0f}%" if bps else "—"
return ("☑" if self.store.is_selected(url) else "☐",
KIND_LABEL.get(node["kind"], node["kind"]), node["name"], cur, prog,
human_bytes(node["size"]) if node["size"] else "—",
f"{node['best_mbps']:.1f}" if node["best_mbps"] else "—",
f"{node['last_mbps']:.1f}" if node["last_mbps"] else "—",
node["uses"] or "", url, node["last_used_at"] or "—",
node["note"] or "")
def _row_tag(self, node: dict) -> str:
url = node["url"]
if self.node_status.get(url):
return "failed"
return "running" if url in self.live_speed else ""
def _refresh_nodes(self) -> None:
yview = self.tree.yview() # 保留滚动位置，重填时不跳回顶部
self.tree.delete(*self.tree.get_children())
for node in self.store.sorted_nodes():
tag = self._row_tag(node)
self.tree.insert("", "end", iid=node["url"],
values=self._row_values(node),
tags=(tag,) if tag else ())
self.url_box.configure(values=[n["url"] for n in self.store.sorted_nodes()])
self._refresh_url_info()
self._update_sel_hint()
if yview and yview[0] > 0:
self.tree.yview_moveto(yview[0])
def _refresh_row(self, node: dict) -> None:
"""只更新一行（勾选切换、实时刷新用），避免整表重建造成闪烁。"""
iid = node["url"]
if not self.tree.exists(iid):
return
tag = self._row_tag(node)
self.tree.item(iid, values=self._row_values(node), tags=(tag,) if tag else ())
# ---------------- 多源勾选 ---------------- #
def _on_tree_click(self, event) -> str | None:
"""点「选」列切换勾选；其它列交给 Treeview 原本的选中逻辑。"""
if self.tree.identify("region", event.x, event.y) != "cell":
return None
if self.tree.identify_column(event.x) != "#1":
return None
iid = self.tree.identify_row(event.y)
if not iid:
return None
self._toggle_sel(iid)
return "break"
def _toggle_sel(self, url: str) -> None:
node = self.store.find(url)
if node is None:
return
self.store.toggle_selected(url)
self._refresh_row(node)
self._update_sel_hint()
self._save_quiet()
def _select_all(self) -> None:
self.store.set_selected([n["url"] for n in self.store.sorted_nodes()])
self._refresh_nodes()
self._save_quiet()
def _clear_sel(self) -> None:
self.store.set_selected([])
self._refresh_nodes()
self._save_quiet()
def _save_quiet(self) -> None:
try:
self.store.save()
self.cfg_hint_var.set(f"配置文件：{self.store.path}（改动已自动保存）")
except Exception as exc: # noqa: BLE001
self.cfg_hint_var.set(f"⚠ 保存失败：{type(exc).__name__}: {exc}")
def _update_sel_hint(self) -> None:
picked = self.store.selected_nodes()
n = len(picked)
threads = int_var(self.threads_var, 16, 1, 128)
if n:
names = "、".join(x["name"] for x in picked[:3])
if n > 3:
names += f" 等 {n} 个"
conns = n * threads
warn = (f" ⚠ 共 {conns} 条连接，内存约 {conns} MB，建议调小线程数"
if conns > 256 else "")
self.sel_hint_var.set(
f"已勾选 {n} 个源：{names} → 合计 {conns} 条连接（{n} × {threads} 线程）{warn}")
self.btn_start.config(text=f"开始测速（{n} 个源并发）")
else:
self.sel_hint_var.set(
"未勾选任何源 → 只按「下载地址」测单个源（勾选多个源可并发顶满带宽）")
self.btn_start.config(text="开始测速")
def _refresh_url_info(self) -> None:
url = self.url_var.get().strip()
node = self.store.find(url)
if node is None:
self.node_info_var.set(
"当前节点：未保存（测速成功后会自动记入历史）" if url else "")
return
bits = [f"当前节点：{node['name']}（{KIND_LABEL.get(node['kind'], node['kind'])}）"]
if node["size"]:
bits.append(f"大小 {human_bytes(node['size'])}")
if node["best_mbps"]:
bits.append(f"最佳 {node['best_mbps']:.1f} Mbps")
if node["last_mbps"]:
bits.append(f"最近 {node['last_mbps']:.1f} Mbps")
if node["uses"]:
bits.append(f"已测 {node['uses']} 次")
if node["note"]:
bits.append(node["note"])
self.node_info_var.set(" · ".join(bits))
def _selected_node(self, silent: bool = False) -> dict | None:
sel = self.tree.selection()
if not sel:
if not silent:
messagebox.showinfo("提示", "请先在下面的列表里选中一个节点")
return None
return self.store.find(sel[0])
def _use_selected(self, silent: bool = False) -> None:
node = self._selected_node(silent=silent)
if node is None:
return
self.url_var.set(node["url"])
self.url_box.set(node["url"])
self._refresh_url_info()
self.status_var.set(f"已选择节点：{node['name']}")
def _del_selected(self) -> None:
node = self._selected_node()
if node is None:
return
if not messagebox.askyesno("确认删除", f"删除节点「{node['name']}」？\n\n{node['url']}"):
return
self.store.nodes.remove(node)
self._after_change()
def _clear_history(self) -> None:
count = self.store.history_count()
if count == 0:
messagebox.showinfo("提示", "没有自动记录的历史节点")
return
if not messagebox.askyesno(
"确认清空",
f"删除 {count} 条自动记录的历史节点？\n\n手动新建和预设节点不受影响。"):
return
self.store.clear_history()
self._after_change()
def _open_config_dir(self) -> None:
folder = os.path.dirname(self.store.path)
try:
os.startfile(folder) # noqa: S606 (Windows 专用)
except Exception: # noqa: BLE001
messagebox.showinfo("配置文件", self.store.path)
# ---------------- 新建 / 编辑节点 ---------------- #
def _node_dialog(self, node: dict | None) -> dict | None:
editing = node is not None
win = tk.Toplevel(self)
win.title("编辑测速节点" if editing else "新建测速节点")
win.transient(self)
win.resizable(False, False)
body = ttk.Frame(win, padding=12)
body.pack(fill="both", expand=True)
body.columnconfigure(1, weight=1)
name_var = tk.StringVar(value=node["name"] if editing else "")
url_var = tk.StringVar(value=node["url"] if editing else self.url_var.get().strip())
note_var = tk.StringVar(value=node["note"] if editing else "")
keep_var = tk.BooleanVar(value=(not editing) or node["kind"] != "history")
rows = (("名称", name_var), ("地址", url_var), ("备注", note_var))
for i, (label, var) in enumerate(rows):
pad = (0 if i == 0 else 8, 0)
ttk.Label(body, text=label).grid(row=i, column=0, sticky="w", pady=pad)
ttk.Entry(body, textvariable=var, width=66).grid(
row=i, column=1, sticky="ew", padx=(8, 0), pady=pad)
actions = ttk.Frame(body)
actions.grid(row=1, column=2, sticky="w", padx=(6, 0))
def paste_url() -> None:
try:
text = self.clipboard_get().strip()
except Exception: # noqa: BLE001
text = ""
if text:
url_var.set(text)
ttk.Button(actions, text="粘贴", width=6, command=paste_url).pack(side="top")
ttk.Checkbutton(body, text="永久保存（不受「清空历史」影响）",
variable=keep_var).grid(
row=3, column=1, sticky="w", pady=(10, 0))
ttk.Label(body, text="留空备注即可；地址必须是 http:// 或 https:// 开头的直链。",
style="Hint.TLabel").grid(row=4, column=1, sticky="w", pady=(4, 0))
result: dict = {"ok": False}
def on_ok() -> None:
url = url_var.get().strip()
if not is_http_url(url):
messagebox.showwarning("提示", "地址必须以 http:// 或 https:// 开头",
parent=win)
return
result.update(ok=True, name=name_var.get().strip() or auto_name(url),
url=url, note=note_var.get().strip(),
keep=bool(keep_var.get()))
win.destroy()
buttons = ttk.Frame(body)
buttons.grid(row=5, column=1, sticky="e", pady=(14, 0))
ttk.Button(buttons, text="取消", command=win.destroy).pack(side="right")
ttk.Button(buttons, text="确定", command=on_ok).pack(side="right", padx=(0, 8))
win.bind("<Return>", lambda _e: on_ok())
win.bind("<Escape>", lambda _e: win.destroy())
win.update_idletasks()
win.geometry(f"+{self.winfo_rootx() + 90}+{self.winfo_rooty() + 120}")
win.grab_set()
self.wait_window(win)
return result if result["ok"] else None
def _dlg_new(self) -> None:
data = self._node_dialog(None)
if not data:
return
kind = "manual" if data["keep"] else "history"
existing = self.store.find(data["url"])
if existing is not None:
existing["name"] = data["name"]
existing["note"] = data["note"]
if existing["kind"] == "history" and data["keep"]:
existing["kind"] = "manual"
self.status_var.set(f"地址已存在，已更新节点：{existing['name']}")
else:
self.store.add(data["name"], data["url"], data["note"], kind)
self.status_var.set(f"已新建节点：{data['name']}")
self.url_var.set(data["url"])
self.url_box.set(data["url"])
self._after_change()
def _dlg_edit(self) -> None:
node = self._selected_node()
if node is None:
return
data = self._node_dialog(node)
if not data:
return
other = self.store.find(data["url"])
if other is not None and other is not node:
messagebox.showwarning("提示", "该地址已被另一个节点使用")
return
node["name"] = data["name"]
node["url"] = data["url"]
node["note"] = data["note"]
if data["keep"]:
node["kind"] = "preset" if node["kind"] == "preset" else "manual"
else:
node["kind"] = "history"
self._after_change()
def _check_selected(self) -> None:
"""检测勾选的源（一个都没勾时，检测列表里高亮的那一行）。"""
picked = self.store.selected_nodes()
if not picked:
node = self._selected_node()
if node is None:
return
picked = [node]
urls = [n["url"] for n in picked]
timeout = int_var(self.timeout_var, 8, 3, 60)
self.status_var.set(f"正在并发检测 {len(urls)} 个源 ...")
def work() -> None:
lock = threading.Lock()
out: list[tuple[str, object]] = []
def one(u: str) -> None:
try:
res: object = preflight(u, timeout)
except Exception as exc: # noqa: BLE001
res = f"{type(exc).__name__}: {exc}"
with lock:
out.append((u, res))
ts = [threading.Thread(target=one, args=(u,), daemon=True) for u in urls]
for t in ts:
t.start()
for t in ts:
t.join()
self.msg_q.put(("checkedbatch", out))
threading.Thread(target=work, daemon=True).start()
# ---------------- 事件 ---------------- #
def _on_start(self) -> None:
# 勾选了就用勾选的多个源；一个都没勾就退回单地址模式
urls = [n["url"] for n in self.store.selected_nodes()]
if not urls:
url = self.url_var.get().strip()
if not url:
messagebox.showwarning("提示", "请先勾选测速节点，或填写一个下载地址")
return
if not is_http_url(url):
messagebox.showwarning("提示", "地址必须以 http:// 或 https:// 开头")
return
urls = [url]
threads = int_var(self.threads_var, 16, 1, 128)
timeout = int_var(self.timeout_var, 8, 3, 60)
retries = int_var(self.retries_var, 3, 0, 10)
self.btn_start.config(state="disabled")
self.btn_stop.config(state="normal")
self.status_var.set(f"正在并发探测 {len(urls)} 个源...")
self._reset_ui()
self.store.current_url = urls[0]
self._save_quiet()
threading.Thread(target=self._probe_then_start_many,
args=(urls, threads, timeout, retries), daemon=True).start()
def _probe_then_start_many(self, urls: list[str], threads: int, timeout: float,
retries: int) -> None:
"""后台并发探测所有源；成功的立刻全部启动，失败的只记录、不阻塞其余源。"""
lock = threading.Lock()
ready: list[tuple[str, dict]] = []
broken: list[tuple[str, str]] = []
def probe(u: str) -> None:
try:
info = preflight(u, timeout)
except Exception as exc: # noqa: BLE001
with lock:
broken.append((u, f"{type(exc).__name__}: {exc}"))
return
with lock:
ready.append((u, info))
probes = [threading.Thread(target=probe, args=(u,), daemon=True) for u in urls]
for t in probes:
t.start()
for t in probes:
t.join()
if not ready:
self.msg_q.put(("allfailed", broken))
return
engines: list[Engine] = []
for u, info in ready:
eng = Engine(u, info, threads, timeout=timeout, retries=retries)
eng.start()
engines.append(eng)
self.msg_q.put(("started", (engines, ready, broken)))
def _on_stop(self) -> None:
if any(e.running for e in self.engines):
self.status_var.set("正在停止...")
self.btn_stop.config(state="disabled")
for eng in self.engines:
eng.stop()
def _on_close(self) -> None:
if any(e.running for e in self.engines):
self._finish_session(closing=True)
for eng in self.engines:
eng.stop()
current = self.url_var.get().strip()
if is_http_url(current):
self.store.current_url = current
try:
self.store.save()
except Exception: # noqa: BLE001
pass
self.destroy()
def _reset_ui(self) -> None:
self.hist.clear()
self.peak = 0.0
self.prev_bytes = 0
self.prev_t = time.perf_counter()
self.seen_errors = 0
self.live_speed.clear()
self.live_prog.clear()
self.live_prev.clear()
self.node_status.clear()
self.engines = []
self.session_engines = []
self.session_done = True
self.now_var.set("0.0 Mbps (0.0 MB/s)")
for v in self.stat_vars.values():
v.set("—")
self.prog_var.set(0.0)
self.round_var.set("等待开始")
self.err_var.set("")
self.canvas.delete("all")
self._refresh_nodes()
# ---------------- 定时刷新 ---------------- #
def _tick(self) -> None:
self._drain_messages()
now = time.perf_counter()
status = self.status_var.get()
if any(e.running for e in self.engines):
self._update_live(now)
elif self.engines and (status == "正在停止..." or status.startswith("运行中")):
self._finish_session()
self.status_var.set("已停止" if status == "正在停止..."
else "已结束（引擎自行停止）")
self.btn_start.config(state="normal")
self.btn_stop.config(state="disabled")
self.after(SAMPLE_MS, self._tick)
def _drain_messages(self) -> None:
while True:
try:
kind, payload = self.msg_q.get_nowait()
except queue.Empty:
return
if kind == "allfailed":
self.status_var.set("全部源都不可用")
self.btn_start.config(state="normal")
self.btn_stop.config(state="disabled")
for u, m in payload:
self.node_status[u] = m
self._refresh_nodes()
detail = "\n".join(f"· {self._node_name(u)}\n {m}" for u, m in payload[:4])
self.err_var.set(f"所有源都探测失败：{detail}")
messagebox.showerror("无法开始",
f"{len(payload)} 个源全部探测失败：\n\n{detail}")
elif kind == "started":
engines, ready, broken = payload
self.engines = engines
self.session_engines = list(engines)
self.session_done = False
self.prev_t = time.perf_counter()
self.prev_bytes = 0
self.live_speed.clear()
self.live_prog.clear()
self.live_prev.clear()
self.node_status.clear()
not_range = 0
for eng, (_u, info) in zip(engines, ready):
url = eng.url
self.live_prev[url] = (0, self.prev_t)
self.live_speed[url] = 0.0
self.node_status[url] = ""
self.store.touch(url, size=info.get("total") or 0)
if not eng.use_range:
not_range += 1
for url, msg in broken:
self.node_status[url] = msg
first = engines[0]
self.store.current_url = first.url
self.url_var.set(first.url)
self.url_box.set(first.url)
self._after_change(quiet=True)
conns = sum(e.threads for e in engines)
bits = [f"运行中 · {len(engines)} 个源并发 · 合计 {conns} 条连接"]
if broken:
bits.append(f"{len(broken)} 个源失败已跳过")
self.status_var.set(" · ".join(bits))
notes = []
if broken:
notes.append("失败源：" + "，".join(self._node_name(u) for u, _ in broken))
if not_range:
notes.append(f"{not_range} 个源不支持分段，已退化为单线程")
self.err_var.set("；".join(notes))
elif kind == "checkedbatch":
idle = self.session_done # 测速中就不要改动失败标记
rows = []
bad = 0
for url, res in payload:
name = self._node_name(url)
if isinstance(res, dict):
node = self.store.find(url)
if node is not None and res.get("total"):
node["size"] = int(res["total"])
total = (human_bytes(res["total"]) if res.get("total")
else "大小未知")
seg = "支持分段" if res.get("ranges") else "不支持分段→单线程"
rows.append(f"✅ {name}：{total} · {seg}")
if idle:
self.node_status.pop(url, None)
else:
bad += 1
rows.append(f"❌ {name}：{res}")
if idle:
self.node_status[url] = str(res)
self._after_change(quiet=True)
self.status_var.set(
f"检测完成：{len(payload) - bad} 个可用，{bad} 个失败")
body = "\n".join(rows[:12]) + ("\n…" if len(rows) > 12 else "")
messagebox.showinfo("检测结果", body)
def _node_name(self, url: str) -> str:
node = self.store.find(url)
return node["name"] if node else auto_name(url)
def _finish_session(self, closing: bool = False) -> None:
"""测速结束时把各源的平均速度写回对应节点。
多源并发时带宽是被瓜分的，各源速度必然低于它单独跑的成绩；把它写进
best/last 会污染「这个源单独能跑多快」的记录，所以只在单源模式下记录。
"""
if self.session_done or not self.session_engines:
return
self.session_done = True
if len(self.session_engines) > 1:
return
eng = self.session_engines[0]
snap = eng.snapshot()
elapsed = eng.elapsed()
if elapsed < 1.0 or snap["bytes"] < 1024 * 1024:
return # 太短/太小的运行不记
self.store.record_result(eng.url, mbps(snap["bytes"] / elapsed))
if not closing:
self._after_change(quiet=True)
def _update_live(self, now: float) -> None:
engines = self.engines
if not engines:
return
snaps = [(e, e.snapshot()) for e in engines]
total = sum(s["bytes"] for _e, s in snaps)
# ---- 合计速度：所有源的字节增量之和 ----
dt = now - self.prev_t
# 采样窗口太短会算出离谱的瞬时值（刚启动那一帧 dt 可能只有几微秒），
# 这时不取样，把窗口累积到下一次刷新，避免峰值被污染。
if dt >= 0.05:
inst = max(0.0, (total - self.prev_bytes) / dt)
self.hist.append(inst)
self.peak = max(self.peak, inst)
self.now_var.set(speed_text(inst))
self.prev_bytes = total
self.prev_t = now
# ---- 每个源自己的瞬时速度与本轮进度（喂给列表的「当前Mbps」「进度」列）----
active = 0
conns = 0
errs = 0
segs = 0
last_err = ""
prog_sum = 0.0
for eng, snap in snaps:
url = eng.url
prev_b, prev_t = self.live_prev.get(url, (0, now))
sdt = now - prev_t
if sdt >= 0.05:
self.live_speed[url] = max(0.0, (snap["bytes"] - prev_b) / sdt)
self.live_prev[url] = (snap["bytes"], now)
if eng.total:
self.live_prog[url] = (snap["bytes"] % eng.total) / eng.total
else:
self.live_prog[url] = 0.0
prog_sum += self.live_prog[url]
if eng.running:
active += 1
conns += eng.threads
errs += snap["errors"]
segs += snap["segments"]
if snap["last_error"]:
last_err = snap["last_error"]
elapsed = max(e.elapsed() for e in engines)
avg = total / elapsed if elapsed > 0 else 0.0
prog_avg = prog_sum / len(engines)
v = self.stat_vars
v["avg"].set(speed_text(avg))
v["peak"].set(f"{mbps(self.peak):.1f} Mbps")
v["total"].set(human_bytes(total))
v["uptime"].set(human_time(elapsed))
v["sources"].set(f"{active}/{len(engines)}")
v["conns"].set(str(conns))
v["errors"].set(str(errs))
v["segspeed"].set(f"{segs} 段")
self.prog_var.set(prog_avg * 100)
self.round_var.set(
f"{len(engines)} 个源并发 · 已下载 {human_bytes(total)} · "
f"各源本轮进度平均 {prog_avg * 100:.0f}%")
if errs != self.seen_errors:
self.seen_errors = errs
self.err_var.set(f"最近错误：{last_err}")
# 只刷新参与本次测速的行，避免整表重建
for eng, _snap in snaps:
node = self.store.find(eng.url)
if node is not None:
self._refresh_row(node)
self._draw_graph()
# ---------------- 曲线绘制 ---------------- #
def _draw_graph(self) -> None:
c = self.canvas
w, h = c.winfo_width(), c.winfo_height()
c.delete("all")
if w <= 20 or h <= 20:
return
pad = 8
vals = [mbps(v) for v in self.hist] # 直方图存的是字节/秒，纵轴换成 Mbps
if len(vals) < 2:
return
vmax = nice_max(max(vals))
for i in range(1, 4):
y = pad + (h - 2 * pad) * i / 4
c.create_line(pad, y, w - pad, y, fill="#ececec")
n = len(vals)
pts: list[float] = []
for i, val in enumerate(vals):
x = pad + (w - 2 * pad) * (i / (n - 1))
y = (h - pad) - (h - 2 * pad) * min(val / vmax, 1.0)
pts.extend((x, y))
c.create_polygon(*pts, w - pad, h - pad, pad, h - pad,
fill="#dceaf8", outline="")
c.create_line(*pts, fill="#1565c0", width=2, smooth=False)
c.create_text(pad + 4, pad + 2, anchor="nw", text=f"上限 {vmax:.0f} Mbps",
fill="#888888", font=(FONT_NUM, 8))
# --------------------------------------------------------------------------- #
def main() -> None:
# 高分屏下让界面不糊
try:
from ctypes import windll
windll.shcore.SetProcessDpiAwareness(1)
except Exception: # noqa: BLE001
pass
app = App()
app.mainloop()
if __name__ == "__main__":
try:
main()
except Exception as exc: # .pyw 没有控制台，出错必须弹窗告知
import traceback
try:
root = tk.Tk()
root.withdraw()
messagebox.showerror("启动失败",
f"{type(exc).__name__}: {exc}\n\n{traceback.format_exc()}")
except Exception: # noqa: BLE001
pass
[复制链接]
发表于 2026-10-3 09:43
分享到朋友圈
