""", encoding="utf-8")
return html_path, json_path
def public_report_payload(report: dict[str, Any]) -> dict[str, Any]:
"""Convert RC4 native data to the public site's established flat schema."""
carrier_names = {"CT": "中国电信", "CU": "中国联通", "CM": "中国移动"}
carrier_marks = {"CT": "◈", "CU": "∞", "CM": "▰"}
def average(values: list[Any]) -> float | None:
numeric = [float(x) for x in values if isinstance(x, (int, float))]
return round(statistics.mean(numeric), 1) if numeric else None
def flatten_forward(item: dict[str, Any]) -> dict[str, Any]:
s = item["stats"]
return {
"region": item.get("region", "中国"),
"label": f"{item.get('region', '中国')}远端实测 → VPS",
"access": item["access"], "publicIp": "",
"verified": item["verified"],
"requestedCity": item.get("requestedCity", ""),
"actualProbeCity": item.get("actualProbeCity", ""),
"selectionScope": item.get("selectionScope", ""),
"regionalVerified": bool(item.get("regionalVerified")),
"sourceAsn": item.get("sourceAsn", 0),
"nationalProbeReused": bool(item.get("nationalProbeReused")),
"probeFingerprint": item.get("probeFingerprint", ""),
"route": item["class"], "evidence": item["evidence"],
"avg": s.get("avg"), "min": s.get("minimum"), "max": s.get("maximum"),
"p95": s.get("p95"), "jitter": s.get("jitter"), "stddev": None,
"loss": s.get("loss"),
"success": f"{s.get('success', 0)}/{s.get('expected', 0)}",
"routeHops": item.get("routeHops"),
"timeoutHops": item.get("timeoutHops"),
"backboneTags": item.get("backboneTags", []),
"routeNote": item.get("routeNote", ""),
"probeCapital": item.get("probeCapital", CAPITALS.get(item.get("region", ""), "")),
"probeHealth": item.get("probeHealth", ""),
"reachability": item.get("reachability", "INCONCLUSIVE"),
"latencyVisual": item.get("latencyVisual", {}),
}
carriers: list[dict[str, Any]] = []
for carrier in ("CT", "CU", "CM"):
forward_items = [x for x in report["forward"] if x["carrier"] == carrier]
reached_forward_items = [
x for x in forward_items
if int(x.get("rank", 0)) > 0 and x.get("verified") and x.get("targetReached")
]
regional_forward_items = [
x for x in reached_forward_items if not national_reference(x)
]
reference_forward_items = [
x for x in reached_forward_items
if national_reference(x) and not x.get("nationalProbeReused")
]
shared_reference_items = [
x for x in reached_forward_items
if national_reference(x) and x.get("nationalProbeReused")
]
valid_forward_items = regional_forward_items + reference_forward_items
forward_probes = [flatten_forward(x) for x in forward_items]
grade_item = next(x for x in report["grades"] if x["carrier"] == carrier)
return_items = [x for x in report["returns"] if x["carrier"] == carrier]
probes: list[dict[str, Any]] = []
for item in return_items:
s = item["stats"]
probes.append({
"city": item["city"], "host": item["host"], "ip": item["probeIp"],
"targetPort": item.get("targetPort", 443),
"nodeSource": item.get("nodeSource", "STATIC+NETQUALITY_FALLBACK"),
"fallbackUsed": bool(item.get("fallbackUsed")),
"route": item["class"], "evidence": item["evidence"],
"observedClasses": item.get("observedClasses", []),
"avg": s.get("avg"), "min": s.get("minimum"), "max": s.get("maximum"),
"p95": s.get("p95"), "jitter": s.get("jitter"), "stddev": None,
"loss": s.get("loss"), "success": f"{s.get('success', 0)}/{s.get('expected', 0)}",
"routeHops": item.get("routeHops", 0),
"timeoutHops": 0,
"backboneTags": item.get("backboneTags", []),
"routeNote": item.get("routeNote", ""),
"probeCapital": item.get("probeCapital", CAPITALS.get(item.get("city", ""), "")),
"probeHealth": item.get("probeHealth", ""),
"reachability": item.get("reachability", "INCONCLUSIVE"),
"latencyVisual": item.get("latencyVisual", {}),
})
forward_flat = {
"region": f"{MATRIX_LABEL}汇总",
"label": f"{MATRIX_LABEL}远端实测 → VPS",
"access": MATRIX_CITIES,
"publicIp": "",
"verified": len(regional_forward_items) == len(forward_items),
"route": grade_item["forwardRoute"],
"evidence": f"{MATRIX_LABEL}去程共 {len(forward_items)} 组",
"avg": average([x["stats"].get("avg") for x in valid_forward_items]),
"min": average([x["stats"].get("minimum") for x in valid_forward_items]),
"max": average([x["stats"].get("maximum") for x in valid_forward_items]),
"p95": average([x["stats"].get("p95") for x in valid_forward_items]),
"jitter": average([x["stats"].get("jitter") for x in valid_forward_items]),
"stddev": None, "loss": None,
"success": f"{len(regional_forward_items)}/{len(forward_items)}",
"regionalSuccess": f"{len(regional_forward_items)}/{len(forward_items)}",
"nationalReference": len(reference_forward_items),
"sharedReference": len(shared_reference_items),
"routeHops": None, "timeoutHops": None,
"backboneTags": sorted({
tag for x in valid_forward_items for tag in x.get("backboneTags", [])
}),
"routeNote": (
"各地区分别列示骨干标签与中文路由注释；"
"全国同运营商参考不会冒充指定地区证据。"
),
"reachability": (
f"指定地区 {len(regional_forward_items)}/{len(forward_items)}"
f"｜全国独立参考 {len(reference_forward_items)}"
f"｜复用参考 {len(shared_reference_items)}"
),
}
carriers.append({
"id": carrier, "name": carrier_names[carrier],
"brand": {
"mark": carrier_marks[carrier],
"short": carrier,
"color": CARRIER_HEX[carrier],
},
"route": grade_item["returnRoute"], "probeCount": len(probes),
"routeTypes": len({
x["class"] for x in return_items if int(x.get("rank", 0)) > 0
}),
"forward": forward_flat, "forwardRoute": grade_item["forwardRoute"],
"forwardProbes": forward_probes,
"evidenceCoverage": grade_item["evidenceCoverage"],
"measurementCoverage": grade_item["measurementCoverage"],
"scorableCoverage": grade_item["scorableCoverage"],
"matrixStatus": grade_item["matrixStatus"],
"matrixComplete": grade_item["matrixComplete"],
"forwardMeasured": grade_item["forwardMeasured"],
"forwardRegional": grade_item["forwardRegional"],
"forwardReference": grade_item["forwardReference"],
"forwardSharedReference": grade_item["forwardSharedReference"],
"forwardReached": grade_item["forwardReached"],
"forwardFailures": grade_item["forwardFailures"],
"returnMeasured": grade_item["returnMeasured"],
"returnValid": grade_item["returnValid"],
"returnFailures": grade_item["returnFailures"],
"bidirectional": grade_item["bidirectionalPremium"],
"routeGrade": grade_item["routeGrade"],
"probes": probes,
})
matrix_complete = all(x["matrixComplete"] for x in report["grades"])
forward_regional_total = sum(
int(str(x["forwardRegional"]).split("/", 1)[0])
for x in report["grades"]
)
route_classification = report["routeClassification"]
final_title = (
f"{route_classification['overallLabel']}｜"
+ (
"三网完整矩阵"
if matrix_complete else
f"矩阵 PARTIAL｜指定地区去程 {forward_regional_total}/{EXPECTED_PER_DIRECTION}"
)
)
return {
"version": report["version"], "generated": report["generated"],
"target": report["target"]["host"], "targetPort": report["target"]["port"],
"returnSshHost": report["exit"]["host"], "selfTest": report["selfTest"],
"mode": report["mode"],
"matrix": (
f"{MATRIX_CITIES} × 三网去程＋回程（{TOTAL_MATRIX_GROUPS} 组）"
),
"methodology": report["methodology"],
"speedMethodology": report.get("speedMethodology", ""),
"bgp": {"asn": report["dedicatedLine"]["exitAsn"],
"provider": report["exit"]["identity"], "location": ""},
"final": {
"matrixComplete": matrix_complete,
"matrixStatus": "COMPLETE" if matrix_complete else "PARTIAL",
"title": final_title,
"elapsed": "N/A",
"routeGrade": route_classification,
},
"carriers": carriers,
"routeClassification": route_classification,
"singleThreadSpeed": report.get("singleThreadSpeed", {}),
"improvements": report.get("improvements", []),
"latencyHeatmap": report.get("latencyHeatmap", {}),
"matrixAssessment": report.get("matrixAssessment", {}),
"dedicatedLine": report["dedicatedLine"],
}
def public_url_from_response(result: dict[str, Any]) -> str:
candidates = [result]
candidates.extend(
value for value in result.values() if isinstance(value, dict)
)
for candidate in candidates:
if not isinstance(candidate, dict):
continue
for key in (
"url", "reportUrl", "report_url", "publicUrl", "public_url"
):
if candidate.get(key):
value = str(candidate[key])
return (
value
if value.startswith("http")
else PUBLIC_REPORT_ROOT + value
)
if candidate.get("id"):
return f"{PUBLIC_REPORT_ROOT}/r/{candidate['id']}"
return ""
def show_public_report_links(public_url: str) -> None:
"""Print synchronized light and dark report URLs after upload."""
dark_url = public_url.rstrip("/")
if dark_url.endswith("/readable"):
dark_url = dark_url[:-len("/readable")]
readable_url = f"{dark_url}/readable"
field("浅色报告", readable_url, GREEN)
field("深色报告", dark_url, CYAN)
field(
"页面切换",
"网页内可随时切换浅色／深色；NodeSeek 提供浅色高清与深色高清两套格式",
GRAY,
)
def post_report_json(
payload: dict[str, Any],
timeout: int = 30,
) -> dict[str, Any]:
"""Use the same urllib POST shape and headers as IX v1.2.2."""
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
request = urllib.request.Request(
PUBLIC_REPORT_API,
data=body,
method="POST",
headers={
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": "ix-route/0.1",
},
)
with urllib.request.urlopen(request, timeout=timeout) as response:
parsed = json.loads(response.read().decode("utf-8"))
if not isinstance(parsed, dict):
raise RuntimeError("标准 POST 未返回有效 JSON")
return parsed
def cli_sync_payload(payload: dict[str, Any]) -> str:
"""Compress and URL-safe encode one public report for GET chunk upload."""
raw = json.dumps(
payload,
ensure_ascii=False,
separators=(",", ":"),
).encode("utf-8")
compressed = gzip.compress(raw, compresslevel=9, mtime=0)
if len(compressed) > 192 * 1024:
raise RuntimeError("压缩报告超过 192 KB 上限；本地 HTML／JSON 已保留")
return base64.urlsafe_b64encode(compressed).rstrip(b"=").decode("ascii")
def curl_get_json(
url: str,
params: dict[str, Any],
timeout: int = 45,
) -> dict[str, Any]:
command = [
"curl",
"--silent",
"--show-error",
"--compressed",
"--connect-timeout", "12",
"--max-time", str(timeout),
"--retry", "3",
"--retry-delay", "1",
"--request", "GET",
"--header", "Accept: application/json",
"--header", "Cache-Control: no-cache",
"--user-agent", "Mozilla/5.0 3net-route-cli/1.0.0",
"--get",
]
for name, value in params.items():
command.extend(["--data-urlencode", f"{name}={value}"])
command.extend(["--write-out", "\n%{http_code}", url])
try:
completed = subprocess.run(
command,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
timeout=timeout + 8,
check=False,
)
except subprocess.TimeoutExpired as exc:
raise RuntimeError("GET 分段上传连接超时；本地 HTML／JSON 已保留") from exc
if completed.returncode != 0:
detail = completed.stderr.decode("utf-8", "replace").strip()
detail = re.sub(r"\s+", " ", detail)[:160]
raise RuntimeError(
f"GET 分段上传网络失败（curl {completed.returncode}）"
+ (f"｜{detail}" if detail else "")
)
try:
response_body, status_text = completed.stdout.rsplit(b"\n", 1)
status = int(status_text.strip())
except (ValueError, TypeError) as exc:
raise RuntimeError("GET 分段接口返回格式异常；本地 HTML／JSON 已保留") from exc
result: dict[str, Any] = {}
if response_body.lstrip().startswith((b"{", b"[")):
try:
parsed = json.loads(response_body.decode("utf-8", "replace"))
if isinstance(parsed, dict):
result = parsed
except json.JSONDecodeError:
result = {}
if not 200 <= status < 300:
api_error = str(result.get("error") or "").strip()
if not api_error:
api_error = (
"站点边缘拒绝了 GET 请求"
if status == 403 else
"接口未返回可读 JSON"
)
raise RuntimeError(
f"HTTP {status}｜{api_error}；本地 HTML／JSON 已保留"
)
if not result:
raise RuntimeError("GET 分段接口未返回有效 JSON；本地 HTML／JSON 已保留")
return result
def curl_chunked_upload(
payload: dict[str, Any],
timeout: int = 45,
) -> dict[str, Any]:
encoded = cli_sync_payload(payload)
chunk_chars = 1500
chunks = [
encoded[index:index + chunk_chars]
for index in range(0, len(encoded), chunk_chars)
]
if not chunks or len(chunks) > 128:
raise RuntimeError("报告分段数量超过接口上限；本地 HTML／JSON 已保留")
upload_id = os.urandom(5).hex()
ping = curl_get_json(
PUBLIC_REPORT_SYNC_API,
{"action": "ping"},
timeout,
)
if ping.get("transport") != "get-chunks-v1":
raise RuntimeError("零配置上传通道尚未就绪；本地 HTML／JSON 已保留")
total = len(chunks)
for part, chunk in enumerate(chunks):
result = curl_get_json(
PUBLIC_REPORT_SYNC_API,
{
"action": "chunk",
"id": upload_id,
"part": part,
"total": total,
"data": chunk,
},
timeout,
)
if result.get("accepted") != part:
raise RuntimeError(
f"上传第 {part + 1}/{total} 段确认异常；本地 HTML／JSON 已保留"
)
return curl_get_json(
PUBLIC_REPORT_SYNC_API,
{
"action": "complete",
"id": upload_id,
"total": total,
},
timeout,
)
def publish(report: dict[str, Any]) -> str:
"""Try the IX-compatible POST, then the public zero-config GET transport."""
payload = public_report_payload(report)
errors: list[str] = []
try:
result = post_report_json(payload, 30)
public_url = public_url_from_response(result)
if public_url:
field("上传通道", "标准 POST（IX v1.2.2 同方式）", GREEN)
return public_url
errors.append("标准 POST 未返回报告网址")
except Exception as exc:
detail = re.sub(r"\s+", " ", str(exc)).strip()[:160]
errors.append(f"标准 POST：{detail}")
try:
result = curl_chunked_upload(payload, 45)
public_url = public_url_from_response(result)
if public_url:
field("上传通道", "公共 GET 分段通道（零配置）", GREEN)
return public_url
errors.append("GET 分段上传未返回报告网址")
except Exception as exc:
detail = re.sub(r"\s+", " ", str(exc)).strip()[:160]
errors.append(f"GET 分段：{detail}")
field(
"公共报告",
"上传失败｜" + "；".join(errors)
+ "；本地 HTML／JSON 已保留",
YELLOW,
)
return ""
def publish_report_file(json_path: Path) -> str:
"""Read the saved report and use the same zero-config upload path."""
resolved_path = json_path.expanduser().resolve()
if not resolved_path.is_file():
raise RuntimeError(f"找不到已生成的 JSON：{resolved_path}")
try:
stored_report = json.loads(resolved_path.read_text(encoding="utf-8"))
except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
raise RuntimeError(f"无法读取已生成的 JSON：{exc}") from exc
if not isinstance(stored_report, dict):
raise RuntimeError("已生成的 JSON 不是有效报告对象")
return publish(stored_report)
def main() -> int:
logo()
if RETRY_UPLOAD_PATH:
retry_path = Path(RETRY_UPLOAD_PATH).expanduser().resolve()
banner("REPORT RETRY / 已测结果免重跑上传", MAGENTA)
field("JSON 数据", retry_path, GREEN)
field("处理方式", "只上传现有报告；不重新执行路由、延迟或速度测试", CYAN)
public_url = publish_report_file(retry_path)
if not public_url:
return 8
show_public_report_links(public_url)
return 0
if SELF_TEST:
exit_ip, exit_identity = "45.207.225.70", "SELF-TEST VPS"
else:
exit_ip, exit_identity = public_ip()
target, port = ask_target(exit_ip)
banner("TARGET / 双端检测模型", CYAN)
native_mode = target == exit_ip
field("被测 VPS", f"{mask_ip(target)}:{port}", CYAN)
field("执行位置", f"{mask_ip(exit_ip)}｜{exit_identity or '本机原生执行'}", GREEN)
field("认证方式", "VPS 本机原生执行｜不使用 SSH／密码／私钥", GREEN)
field("去程核对方式", "中国本地端主动证据优先；缺项才使用 Globalping，不以回程反推", CYAN)
field("测试矩阵", f"{MATRIX_LABEL}｜{MATRIX_CITIES}", MAGENTA)
client_forward, client_forward_status = load_client_forward_evidence(
FORWARD_EVIDENCE_PATH, target, port
)
field(
"真实去程证据",
client_forward_status,
GREEN if client_forward else YELLOW,
)
forward: list[dict[str, Any]] = []
forward_total = len(FORWARD_REGIONS) * 3
forward_index = 0
for carrier in ("CT", "CU", "CM"):
for region, probe_city in FORWARD_REGIONS:
forward_index += 1
if SELF_TEST:
item = self_test_forward(carrier, region, probe_city)
elif (carrier, region) in client_forward:
item = client_forward[(carrier, region)]
else:
field(
f"[{forward_index}/{forward_total}] 去程",
f"{region} AS{FORWARD_ASN[carrier]} → {mask_ip(target)}:{port}",
CARRIER_COLOR[carrier],
)
field("探针状态", "本地证据缺项，正在请求 Globalping 同省远端 TCP traceroute……", GRAY)
item = globalping_trace(target, port, carrier, region, probe_city)
item = add_route_labels(item)
forward.append(item)
show_forward(item, forward_index, forward_total)
true_forward_count = sum(
1 for item in forward
if item.get("evidenceSource") == "CHINA_CLIENT_ACTIVE"
)
if not SELF_TEST and true_forward_count < len(forward):
banner("TRUE FORWARD / 中国本地端补测提示", MAGENTA)
field(
"已导入真去程",
f"{true_forward_count}/{EXPECTED_PER_DIRECTION} 组",
GREEN if true_forward_count else YELLOW,
)
field(
"Windows 客户端",
"运行 cn3_client_probe.ps1，按当前网络填写 CT／CU／CM 与所在省份，生成 forward_evidence.json",
CYAN,
)
field(
"重新合并",
f"bash 3net-route.sh --target {target} --port {port} --forward-evidence /root/forward_evidence.json",
GRAY,
)
dedicated_line = dedicated_line_assessment(target, port, exit_ip, forward)
banner(
"VPS MODEL / 原生公网双向模型"
if native_mode else
"FORWARDING MODEL / 入口与 NAT 内段",
MAGENTA,
)
field("网络拓扑", dedicated_line["topology"], CYAN)
field("外部端口", dedicated_line["portStatus"],
GREEN if dedicated_line["reachedBy"] else YELLOW)
field("目标 ASN", dedicated_line["entryAsn"], CYAN)
field("出口 ASN", dedicated_line["exitAsn"], GREEN)
field(
"内段可见性",
(
"不适用｜目标与执行 VPS 为同一公网 IP"
if native_mode else
"NAT 隐藏｜不强判精品或普通线路"
),
GREEN if native_mode else YELLOW,
)
field("判断边界", dedicated_line["internalVerdict"], GRAY)
return_pool, return_pool_status = load_return_probe_pool()
return_total = sum(len(items) for items in return_pool.values())
banner(
f"RETURN PROBE / 出口 VPS 原生三网"
f"{EXPECTED_PER_DIRECTION}组回程",
CYAN,
)
field("执行位置", mask_ip(exit_ip), GREEN)
field("动态测点池", return_pool_status, GREEN if return_pool_status.startswith("ONLINE") else YELLOW)
field("执行方式", "真实 TCP 端口＋TCP／ICMP／UDP 交叉识别；证据不足再切 oneclickvirt 同省三网 ICMP 备用", GRAY)
returns: list[dict[str, Any]] = []
index = 0
for carrier in ("CT", "CU", "CM"):
for spec in return_pool[carrier]:
city, host = str(spec["city"]), str(spec["host"])
index += 1
item = self_test_return(carrier, city, host) if SELF_TEST else return_probe(carrier, city, spec)
item = add_route_labels(item)
returns.append(item)
show_return(item, index, return_total)
if SPEED_TEST:
banner("NEXT / 开始三网公网单线程测速", YELLOW)
field(
"预计耗时",
"约 4～12 分钟｜北上广三网 9 组｜会消耗实际测速流量",
YELLOW,
)
speed = single_thread_speed()
latency_heatmap = annotate_latency_visuals(forward, returns)
grades = [grade(c, forward, returns) for c in ("CT", "CU", "CM")]
route_classification = overall_route_grade(grades)
measured_total = sum(
int(item["forwardMeasured"].split("/", 1)[0])
+ int(item["returnMeasured"].split("/", 1)[0])
for item in grades
)
scorable_total = sum(
int(item["forwardValid"].split("/", 1)[0])
+ int(item["returnValid"].split("/", 1)[0])
for item in grades
)
matrix_complete = (
all(item["matrixComplete"] for item in grades)
and measured_total == TOTAL_MATRIX_GROUPS
)
matrix_assessment = {
"status": "COMPLETE" if matrix_complete else "PARTIAL",
"completed": f"{measured_total}/{TOTAL_MATRIX_GROUPS}",
"scorable": f"{scorable_total}/{TOTAL_MATRIX_GROUPS}",
"measurementCoverage": round(
measured_total / max(1, TOTAL_MATRIX_GROUPS), 3
),
"scorableCoverage": round(
scorable_total / max(1, TOTAL_MATRIX_GROUPS), 3
),
"matrixComplete": matrix_complete,
"coverageNote": (
"指定省份去程与回程全部完成"
if matrix_complete else
"指定省份矩阵未完成；全国参考不会补足省份覆盖"
),
}
if SELF_TEST:
self_test_regressions()
if not all(item["bidirectionalPremium"] for item in grades):
raise AssertionError("离线精品双程样本未通过严格双程判定")
if route_classification["overallLevel"] != "BIDIRECTIONAL_PREMIUM":
raise AssertionError("离线三网最终线路等级未判为双程精品")
banner("FINAL VERDICT / 三网双程线路判定", CYAN)
overall_color = (
GREEN
if route_classification["overallLevel"] == "BIDIRECTIONAL_PREMIUM"
else YELLOW
if route_classification["overallLevel"] == "ONE_WAY_OPTIMIZED_OR_MIXED"
else RED
if route_classification["overallLevel"] == "EVIDENCE_INSUFFICIENT"
else CYAN
)
field("最终线路等级", route_classification["overallLabel"], overall_color)
field("等级判定依据", route_classification["reason"], GRAY)
field("判定边界", route_classification["policy"], YELLOW)
for item in grades:
color = CARRIER_COLOR[item["carrier"]]
field(
carrier_label(item["carrier"]),
f"等级 {item['routeGrade']['label']}｜"
f"去程 {item['forwardRoute']}〔地区 {item['forwardRegional']}｜"
f"测量 {item['forwardMeasured']}｜有效 {item['forwardValid']}｜"
f"全国独立参考 {item['forwardReference']}｜"
f"复用参考 {item['forwardSharedReference']}｜失败 {item['forwardFailures']}〕｜"
f"回程 {item['returnRoute']}〔测量 {item['returnMeasured']}｜"
f"有效 {item['returnValid']}｜失败 {item['returnFailures']}〕｜"
f"测量覆盖 {round(item['measurementCoverage'] * 100)}%｜"
f"矩阵 {item['matrixStatus']}",
color,
)
field("判定说明", item["routeGrade"]["reason"], GRAY)
field(
"矩阵完整性",
f"{matrix_assessment['status']}｜完成 {matrix_assessment['completed']}｜"
f"可判定 {matrix_assessment['scorable']}｜"
+ ("指定省份完整" if matrix_assessment["matrixComplete"] else "指定省份未完整"),
GREEN if matrix_assessment["matrixComplete"] else YELLOW,
)
show_latency_heatmap(forward, returns)
show_single_thread_speed(speed)
improvements = build_improvements(grades, forward, returns, speed)
banner("IMPROVEMENTS / 最终判定建议改善", MAGENTA)
for index, suggestion in enumerate(improvements, start=1):
field(f"建议 {index}", suggestion, YELLOW)
# 评分只保留在运行期供旧分类回归使用；用户报告与公开载荷不输出
# 这些内部字段，避免无意义的兼容字段占用版面。
for item in forward + returns:
item.pop("score", None)
item.pop("stars", None)
for item in grades:
for key in ("forwardScore", "returnScore", "sampleScore", "score", "stars"):
item.pop(key, None)
report = {
"version": VERSION,
"generated": dt.datetime.now(dt.timezone.utc).astimezone().isoformat(timespec="seconds"),
"mode": "VPS_NATIVE_DUAL_ENDPOINT",
"matrixLabel": MATRIX_LABEL,
"matrix": (
f"{MATRIX_CITIES} × 三网去程＋回程（{TOTAL_MATRIX_GROUPS} 组）"
),
"selfTest": SELF_TEST,
"target": {"host": mask_ip(target), "port": port, "role": "被测 VPS 公网"},
"exit": {"host": mask_ip(exit_ip), "identity": exit_identity, "role": "脚本执行 VPS"},
"forwardEvidence": {
"status": client_forward_status,
"clientActive": true_forward_count,
"total": len(forward),
},
"methodology": "默认使用成熟的北上广三网主矩阵；--extended 才追加合肥、南京、杭州。去程优先读取 cn3-forward-evidence/v1：中国本地 Windows 客户端对目标业务端口执行真实 TCP connect 与 tracert，且证据目标必须与本次 IP／端口完全一致；未覆盖项先由 Globalping 请求同省真实外部探针。省会及同省没有探针时，一次收集中国境内同运营商名称及各省网 ASN 的多探针共享快照；北上广与扩展模式使用相同池规则，按实际城市和地区 ASN 优先分配并避免重复。全国参考只证明该运营商网络到入口的线路与可达性，不冒充指定省份，也不能补足指定省份矩阵。仍无探针时标记 NOT-TESTED，不得写成入口不通。完整矩阵必须由指定省份去程探针与对应省份回程目标实际执行；实际执行但未接通保留为真实失败，NOT-TESTED 与复用参考不冒充失败。最终结论以去回程骨干线路类型、各省实测 AVG／P95、相对热度与单线程吞吐为主，全国参考不进入省份延迟热图。net.sh 的 zstaticcdn 目标、TcpQuality 动态节点池以及 zhanghanyun／oneclickvirt backtrace 均为 VPS→中国回程，只用于回程稳定性，绝不冒充去程。回程先使用 TcpQuality 真实端口主备节点与 NetQuality 域名备用，TCP／ICMP／UDP 交叉取证；三协议仍无骨干证据时，才按 oneclickvirt/backtrace 的设计切换 spiritLHLS/icmp_targets 同省同运营商最多三个 ICMP 地址。若多协议同时观察到精品与普通线路，按动态混合保守降级。CN2 GIA 至少需要两个可见 CN2 跳点，单一 CN2 特征或多个 163 交付跳点只判混合／证据不足。",
"speedMethodology": (
"--speed 固定调用已锁定提交的 TcpQuality 北上广三网公共 TOS 单线程测试；"
"回程定义为当前 VPS→中国测速端，去程定义为中国测速端→当前 VPS。"
"测速结果作为独立实测指标，不补足省份路由矩阵。"
),
"forward": forward,
"returns": returns,
"grades": grades,
"routeClassification": route_classification,
"latencyHeatmap": latency_heatmap,
"singleThreadSpeed": speed,
"improvements": improvements,
"matrixAssessment": matrix_assessment,
"dedicatedLine": dedicated_line,
"privacy": "报告中的入口与出口 IPv4 均只保留前两段。",
}
if SELF_TEST:
public_payload = public_report_payload(report)
raw_upload_payload = json.dumps(
public_payload, ensure_ascii=False
).encode("utf-8")
upload_round_trip = json.loads(raw_upload_payload.decode("utf-8"))
encoded_sync_payload = cli_sync_payload(public_payload)
padded_sync_payload = encoded_sync_payload + (
"=" * (-len(encoded_sync_payload) % 4)
)
sync_round_trip = json.loads(
gzip.decompress(
base64.urlsafe_b64decode(padded_sync_payload)
).decode("utf-8")
)
speed_check = public_payload.get("singleThreadSpeed") or {}
serialized_payload = json.dumps(public_payload, ensure_ascii=False)
if upload_round_trip != public_payload:
raise AssertionError("IX v1.2.2 标准 JSON POST 载荷无法无损还原")
if sync_round_trip != public_payload:
raise AssertionError("公共 GET 分段载荷无法无损还原")
if (
not public_payload.get("latencyHeatmap", {}).get("forward")
or any(
token in serialized_payload
for token in ('"score"', '"stars"', '"sampleScore"', '"ratingEnabled"')
)
):
raise AssertionError("公开报告载荷必须使用延迟热图，且不得包含旧评分字段")
if (
len(speed_check.get("rows") or []) != 9
or int((speed_check.get("summary") or {}).get("complete", 0)) != 9
):
raise AssertionError("单线程速度必须包含北上广三网九组与完整摘要")
if any(
item.get("routeHops") == 0
for carrier in public_payload["carriers"]
for item in carrier.get("forwardProbes") or []
if item.get("verified")
):
raise AssertionError("已核对去程不得再硬编码显示 0 hops")
html_path, json_path = write_report(report)
banner("REPORT / 报告输出", MAGENTA)
field("HTML 报告", html_path, GREEN)
field("JSON 数据", json_path, GREEN)
field(
"页面功能",
"双程线路／延迟热图／单线程吞吐／改善建议／NodeSeek／JSON／PDF",
CYAN,
)
if not SELF_TEST:
public_url = publish_report_file(json_path)
if public_url:
show_public_report_links(public_url)
else:
field(
"SELF-TEST",
"PASS｜离线分类、延迟热图、单线程吞吐、跳数与报告生成完成",
GREEN,
)
field("隐私", f"入口 {mask_ip(target)}｜出口 {mask_ip(exit_ip)}；无 SSH 凭据", CYAN)
return 0
if __name__ == "__main__":
try:
raise SystemExit(main())
except KeyboardInterrupt:
print("\n" + YELLOW + "用户已中止。" + RESET)
raise SystemExit(130)
except urllib.error.HTTPError as exc:
detail = exc.read().decode("utf-8", "replace")[:500]
print(RED + f"\n[ERROR] HTTP {exc.code}：{detail}" + RESET)
raise SystemExit(6)
except Exception as exc:
print(RED + f"\n[ERROR] {type(exc).__name__}：{exc}" + RESET)
raise SystemExit(7)
PY