674 lines
22 KiB
Python
674 lines
22 KiB
Python
"""
|
||
网络检测：Google（海外）与抖音（国内）可达性、解析 IP/地区、简易上下行 Mbps。
|
||
说明：上行依赖对端是否接受 POST，失败时返回 null 与原因，属正常现象。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import socket
|
||
import platform
|
||
import subprocess
|
||
import time
|
||
import shutil
|
||
from pathlib import Path
|
||
from typing import Any, Optional
|
||
|
||
import httpx
|
||
import psutil
|
||
|
||
from web2_host_caps import machine_summary_payload
|
||
from web2_ip_quality import run_ip_quality
|
||
|
||
# ---------- 目标 ----------
|
||
GOOGLE_HOST = "www.google.com"
|
||
GOOGLE_CHECK_URL = "https://www.google.com/generate_204"
|
||
GOOGLE_DOWNLOAD_URL = (
|
||
"https://www.gstatic.com/images/branding/googlelogo/2x/googlelogo_color_272x92dp.png"
|
||
)
|
||
|
||
DOUYIN_HOST = "www.douyin.com"
|
||
DOUYIN_CHECK_URL = "https://www.douyin.com/"
|
||
# 首页 HTML 流足够大，便于采样测下行
|
||
DOUYIN_DOWNLOAD_URL = "https://www.douyin.com/"
|
||
|
||
OVERSEAS_UPLOAD_URL = "https://httpbin.org/post"
|
||
DOMESTIC_UPLOAD_URLS = (
|
||
"https://www.douyin.com/",
|
||
"https://www.163.com/",
|
||
)
|
||
|
||
IP_API_TEMPLATE = "http://ip-api.com/json/{ip}?fields=status,country,countryCode,regionName,city,isp,org,as,asname,message,query"
|
||
IP_API_SELF = (
|
||
"http://ip-api.com/json/?fields=status,message,query,country,countryCode,"
|
||
"regionName,city,isp,org,as,asname,proxy,hosting,mobile"
|
||
)
|
||
IPIFY_URL = "https://api.ipify.org?format=json"
|
||
|
||
DOWNLOAD_CAP_BYTES = 600_000
|
||
UPLOAD_BYTES = 256 * 1024
|
||
CONNECT_TIMEOUT = 8.0
|
||
READ_TIMEOUT = 25.0
|
||
NETWORK_LOGIC_VERSION = 2
|
||
|
||
|
||
def _is_probably_china_country(code: Optional[str]) -> bool:
|
||
return (code or "").upper() == "CN"
|
||
|
||
|
||
async def _resolve_ipv4(host: str) -> tuple[Optional[str], Optional[str]]:
|
||
def _sync() -> tuple[Optional[str], Optional[str]]:
|
||
try:
|
||
infos = socket.getaddrinfo(host, 443, type=socket.SOCK_STREAM)
|
||
except socket.gaierror as e:
|
||
return None, str(e)
|
||
for fam, _, _, _, sockaddr in infos:
|
||
if fam == socket.AF_INET:
|
||
return sockaddr[0], None
|
||
for fam, _, _, _, sockaddr in infos:
|
||
if fam == socket.AF_INET6:
|
||
return sockaddr[0], None
|
||
return None, "no address"
|
||
|
||
return await asyncio.to_thread(_sync)
|
||
|
||
|
||
async def _geo_for_ip(client: httpx.AsyncClient, ip: Optional[str] = None) -> dict[str, Any]:
|
||
url = IP_API_SELF if not ip else IP_API_TEMPLATE.format(ip=ip)
|
||
try:
|
||
r = await client.get(url, timeout=8.0)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
except Exception as e:
|
||
return {"error": str(e)}
|
||
if data.get("status") != "success":
|
||
return {
|
||
"error": data.get("message") or "lookup failed",
|
||
"ip": ip or data.get("query"),
|
||
}
|
||
return {
|
||
"ip": data.get("query") or ip,
|
||
"country": data.get("country"),
|
||
"country_code": data.get("countryCode"),
|
||
"region": data.get("regionName"),
|
||
"city": data.get("city"),
|
||
"isp": data.get("isp"),
|
||
"org": data.get("org"),
|
||
"asn": data.get("as"),
|
||
"asname": data.get("asname"),
|
||
"error": None,
|
||
}
|
||
|
||
|
||
async def _country_for_ip(client: httpx.AsyncClient, ip: str) -> dict[str, Any]:
|
||
geo = await _geo_for_ip(client, ip)
|
||
return {
|
||
"country": geo.get("country"),
|
||
"country_code": geo.get("country_code"),
|
||
"error": geo.get("error"),
|
||
}
|
||
|
||
|
||
async def _probe_default_egress(client: httpx.AsyncClient) -> dict[str, Any]:
|
||
"""本机默认出口：优先 ip-api 直连，失败则用 ipify + 反查。"""
|
||
geo = await _geo_for_ip(client, None)
|
||
if geo.get("ip") and not geo.get("error"):
|
||
return geo
|
||
fallback = await _sample_overseas_ip(client)
|
||
if fallback.get("ip") and not fallback.get("error"):
|
||
return fallback
|
||
return geo if geo.get("ip") else fallback
|
||
|
||
|
||
async def _sample_overseas_ip(client: httpx.AsyncClient) -> dict[str, Any]:
|
||
try:
|
||
r = await client.get(IPIFY_URL, timeout=8.0)
|
||
r.raise_for_status()
|
||
ip = (r.json().get("ip") or "").strip()
|
||
if not ip:
|
||
return {"error": "empty ip"}
|
||
return await _geo_for_ip(client, ip)
|
||
except Exception as e:
|
||
return {"error": str(e)}
|
||
|
||
|
||
def _format_isp_line(geo: dict[str, Any]) -> str:
|
||
isp = (geo.get("isp") or geo.get("org") or "").strip()
|
||
country = (geo.get("country") or "").strip()
|
||
city = (geo.get("city") or "").strip()
|
||
loc = " · ".join(x for x in (country, city) if x)
|
||
if isp and loc:
|
||
return f"{isp}（{loc}）"
|
||
return isp or loc or "未知运营商"
|
||
|
||
|
||
def _compute_routes(
|
||
egress: dict[str, Any],
|
||
overseas: dict[str, Any],
|
||
*,
|
||
google_ok: bool,
|
||
douyin_ok: bool,
|
||
) -> tuple[Optional[bool], str]:
|
||
ip = egress.get("ip")
|
||
o_ip = overseas.get("ip")
|
||
cc = (egress.get("country_code") or "").upper()
|
||
|
||
if egress.get("error") and not ip:
|
||
if google_ok and douyin_ok:
|
||
return None, "运营商未识别"
|
||
return None, "出口未识别，请重检"
|
||
|
||
if ip and o_ip and ip != o_ip:
|
||
return True, "已分流"
|
||
|
||
cn = _is_probably_china_country(cc)
|
||
if google_ok and douyin_ok:
|
||
if cn:
|
||
return True, "已分流"
|
||
return False, "未分流"
|
||
|
||
if douyin_ok and not google_ok:
|
||
return False, "Google 不通"
|
||
|
||
if google_ok and not douyin_ok:
|
||
return False, "抖音异常"
|
||
|
||
return False, "均不可达"
|
||
|
||
|
||
async def _latency_get(client: httpx.AsyncClient, url: str) -> tuple[bool, float, Optional[str]]:
|
||
t0 = time.monotonic()
|
||
try:
|
||
r = await client.get(url, follow_redirects=True)
|
||
ok = r.status_code < 400
|
||
ms = (time.monotonic() - t0) * 1000
|
||
return ok, ms, None if ok else f"HTTP {r.status_code}"
|
||
except Exception as e:
|
||
ms = (time.monotonic() - t0) * 1000
|
||
return False, ms, str(e)
|
||
|
||
|
||
async def _measure_download_mbps(
|
||
client: httpx.AsyncClient, url: str, cap: int = DOWNLOAD_CAP_BYTES
|
||
) -> tuple[Optional[float], Optional[str], int]:
|
||
total = 0
|
||
t0 = time.monotonic()
|
||
try:
|
||
async with client.stream("GET", url, follow_redirects=True) as r:
|
||
r.raise_for_status()
|
||
async for chunk in r.aiter_bytes():
|
||
total += len(chunk)
|
||
if total >= cap:
|
||
break
|
||
dt = time.monotonic() - t0
|
||
if dt < 0.001 or total == 0:
|
||
return None, "无数据或过快", total
|
||
mbps = (total * 8) / dt / 1_000_000
|
||
return round(mbps, 2), None, total
|
||
except Exception as e:
|
||
return None, str(e), total
|
||
|
||
|
||
async def _measure_upload_mbps(
|
||
client: httpx.AsyncClient, url: str, size: int = UPLOAD_BYTES
|
||
) -> tuple[Optional[float], Optional[str]]:
|
||
data = b"\x00" * size
|
||
t0 = time.monotonic()
|
||
try:
|
||
r = await client.post(
|
||
url,
|
||
content=data,
|
||
headers={"Content-Type": "application/octet-stream"},
|
||
)
|
||
dt = time.monotonic() - t0
|
||
if dt < 0.001:
|
||
return None, "耗时过短"
|
||
mbps = (size * 8) / dt / 1_000_000
|
||
return round(mbps, 2), None
|
||
except Exception as e:
|
||
return None, str(e)
|
||
|
||
|
||
async def _try_domestic_upload(client: httpx.AsyncClient) -> tuple[Optional[float], Optional[str], Optional[str]]:
|
||
last_err: Optional[str] = None
|
||
for url in DOMESTIC_UPLOAD_URLS:
|
||
up, err = await _measure_upload_mbps(client, url)
|
||
if up is not None:
|
||
return up, None, url
|
||
last_err = err
|
||
return None, last_err, None
|
||
|
||
|
||
async def run_network_dashboard() -> dict[str, Any]:
|
||
"""
|
||
返回结构供前端仪表盘使用。
|
||
routes_different：基于本机出口 IP/运营商 + Google/抖音可达性判断，不再用 CDN 解析 IP 比国家。
|
||
"""
|
||
timeout = httpx.Timeout(READ_TIMEOUT, connect=CONNECT_TIMEOUT)
|
||
limits = httpx.Limits(max_keepalive_connections=4, max_connections=8)
|
||
|
||
ua = (
|
||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||
"(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
|
||
)
|
||
async with httpx.AsyncClient(
|
||
timeout=timeout,
|
||
limits=limits,
|
||
verify=True,
|
||
headers={"User-Agent": ua},
|
||
) as client:
|
||
g_ip, g_res_err = await _resolve_ipv4(GOOGLE_HOST)
|
||
d_ip, d_res_err = await _resolve_ipv4(DOUYIN_HOST)
|
||
|
||
async def _empty_geo() -> dict[str, Any]:
|
||
return {"country": None, "country_code": None, "error": "no ip"}
|
||
|
||
g_country_task = _country_for_ip(client, g_ip) if g_ip else _empty_geo()
|
||
d_country_task = _country_for_ip(client, d_ip) if d_ip else _empty_geo()
|
||
egress_task = _probe_default_egress(client)
|
||
overseas_task = _sample_overseas_ip(client)
|
||
|
||
(
|
||
(g_ok, g_ms, g_err),
|
||
(d_ok, d_ms, d_err),
|
||
g_geo,
|
||
d_geo,
|
||
egress,
|
||
overseas,
|
||
) = await asyncio.gather(
|
||
_latency_get(client, GOOGLE_CHECK_URL),
|
||
_latency_get(client, DOUYIN_CHECK_URL),
|
||
g_country_task,
|
||
d_country_task,
|
||
egress_task,
|
||
overseas_task,
|
||
)
|
||
|
||
if not isinstance(g_geo, dict):
|
||
g_geo = {"country": None, "country_code": None, "error": None}
|
||
if not isinstance(d_geo, dict):
|
||
d_geo = {"country": None, "country_code": None, "error": None}
|
||
if not isinstance(egress, dict):
|
||
egress = {"error": "egress probe failed"}
|
||
if not isinstance(overseas, dict):
|
||
overseas = {"error": "overseas probe failed"}
|
||
|
||
g_code = g_geo.get("country_code")
|
||
d_code = d_geo.get("country_code")
|
||
|
||
routes_different, routes_note = _compute_routes(
|
||
egress,
|
||
overseas,
|
||
google_ok=g_ok,
|
||
douyin_ok=d_ok,
|
||
)
|
||
|
||
# 测速（并行）
|
||
od_mbps, od_err, od_bytes = await _measure_download_mbps(client, GOOGLE_DOWNLOAD_URL)
|
||
dd_mbps, dd_err, dd_bytes = await _measure_download_mbps(client, DOUYIN_DOWNLOAD_URL)
|
||
|
||
ou_mbps, ou_err = await _measure_upload_mbps(client, OVERSEAS_UPLOAD_URL)
|
||
du_mbps, du_err, du_url = await _try_domestic_upload(client)
|
||
|
||
return {
|
||
"google": {
|
||
"host": GOOGLE_HOST,
|
||
"reachable": g_ok,
|
||
"latency_ms": round(g_ms, 1),
|
||
"error": g_err or g_res_err,
|
||
"resolved_ip": g_ip,
|
||
"country": g_geo.get("country"),
|
||
"country_code": g_code,
|
||
"geo_error": g_geo.get("error"),
|
||
},
|
||
"douyin": {
|
||
"host": DOUYIN_HOST,
|
||
"reachable": d_ok,
|
||
"latency_ms": round(d_ms, 1),
|
||
"error": d_err or d_res_err,
|
||
"resolved_ip": d_ip,
|
||
"country": d_geo.get("country"),
|
||
"country_code": d_code,
|
||
"geo_error": d_geo.get("error"),
|
||
},
|
||
"routes": {
|
||
"different": routes_different,
|
||
"note": routes_note.strip(),
|
||
"isp": egress.get("isp") or egress.get("org"),
|
||
"country": egress.get("country"),
|
||
"country_code": egress.get("country_code"),
|
||
},
|
||
"egress": {
|
||
"default": egress,
|
||
"overseas_sample": overseas,
|
||
},
|
||
"speed": {
|
||
"overseas": {
|
||
"label": "海外（Google / gstatic）",
|
||
"download_mbps": od_mbps,
|
||
"download_error": od_err,
|
||
"download_bytes_sampled": od_bytes,
|
||
"upload_mbps": ou_mbps,
|
||
"upload_error": ou_err,
|
||
"upload_target": OVERSEAS_UPLOAD_URL,
|
||
},
|
||
"domestic": {
|
||
"label": "国内（字节 CDN / 门户）",
|
||
"download_mbps": dd_mbps,
|
||
"download_error": dd_err,
|
||
"download_bytes_sampled": dd_bytes,
|
||
"upload_mbps": du_mbps,
|
||
"upload_error": du_err,
|
||
"upload_target": du_url,
|
||
},
|
||
},
|
||
"meta": {
|
||
"at": time.time(),
|
||
"download_cap_bytes": DOWNLOAD_CAP_BYTES,
|
||
"upload_bytes": UPLOAD_BYTES,
|
||
"logic_version": NETWORK_LOGIC_VERSION,
|
||
},
|
||
}
|
||
|
||
|
||
def _cli_available(command: str) -> bool:
|
||
return shutil.which(command) is not None
|
||
|
||
|
||
def _run_cli(command: list[str], timeout: float = 10.0) -> dict[str, Any]:
|
||
kwargs: dict[str, Any] = {
|
||
"capture_output": True,
|
||
"text": True,
|
||
"timeout": timeout,
|
||
}
|
||
if hasattr(subprocess, "CREATE_NO_WINDOW"):
|
||
kwargs["creationflags"] = subprocess.CREATE_NO_WINDOW
|
||
try:
|
||
r = subprocess.run(command, **kwargs)
|
||
return {
|
||
"ok": r.returncode == 0,
|
||
"code": r.returncode,
|
||
"stdout": (r.stdout or "").strip(),
|
||
"stderr": (r.stderr or "").strip(),
|
||
}
|
||
except Exception as e:
|
||
return {"ok": False, "code": -1, "stdout": "", "stderr": str(e)}
|
||
|
||
|
||
def _safe_json_parse(text: str) -> Optional[dict[str, Any]]:
|
||
if not text:
|
||
return None
|
||
try:
|
||
data = json.loads(text)
|
||
if isinstance(data, dict):
|
||
return data
|
||
except Exception:
|
||
return None
|
||
return None
|
||
|
||
|
||
def _bytes_per_sec_to_mbps(value: Any) -> Optional[float]:
|
||
try:
|
||
n = float(value)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
if n <= 0:
|
||
return None
|
||
# Ookla speedtest: bandwidth 为 bytes/s；speedtest-cli: download 为 bits/s
|
||
if n > 50_000_000:
|
||
return round(n / 1_000_000, 2)
|
||
return round(n * 8 / 1_000_000, 2)
|
||
|
||
|
||
def _summarize_speedtest(raw: Any) -> dict[str, Any]:
|
||
if not isinstance(raw, dict):
|
||
return {}
|
||
download = raw.get("download")
|
||
upload = raw.get("upload")
|
||
ping = raw.get("ping")
|
||
down_mbps = None
|
||
up_mbps = None
|
||
ping_ms = None
|
||
if isinstance(download, dict):
|
||
down_mbps = _bytes_per_sec_to_mbps(download.get("bandwidth"))
|
||
elif download is not None:
|
||
down_mbps = _bytes_per_sec_to_mbps(download)
|
||
if isinstance(upload, dict):
|
||
up_mbps = _bytes_per_sec_to_mbps(upload.get("bandwidth"))
|
||
elif upload is not None:
|
||
up_mbps = _bytes_per_sec_to_mbps(upload)
|
||
if isinstance(ping, dict):
|
||
try:
|
||
ping_ms = round(float(ping.get("latency")), 1)
|
||
except (TypeError, ValueError):
|
||
ping_ms = None
|
||
elif ping is not None:
|
||
try:
|
||
ping_ms = round(float(ping), 1)
|
||
except (TypeError, ValueError):
|
||
ping_ms = None
|
||
return {
|
||
"download_mbps": down_mbps,
|
||
"upload_mbps": up_mbps,
|
||
"ping_ms": ping_ms,
|
||
}
|
||
|
||
|
||
def _summarize_mtr(raw: Any) -> dict[str, Any]:
|
||
if isinstance(raw, dict):
|
||
hubs = raw.get("report", {}).get("hubs") if isinstance(raw.get("report"), dict) else raw.get("hubs")
|
||
if not isinstance(hubs, list):
|
||
hubs = []
|
||
last_avg = None
|
||
if hubs:
|
||
try:
|
||
last_avg = float(hubs[-1].get("Avg", hubs[-1].get("avg", 0)) or 0)
|
||
except (TypeError, ValueError):
|
||
last_avg = None
|
||
return {"hops": len(hubs), "last_hop_avg_ms": last_avg}
|
||
if isinstance(raw, str) and raw.strip():
|
||
lines = [ln for ln in raw.splitlines() if ln.strip() and not ln.startswith("Start:")]
|
||
return {"hops": max(0, len(lines) - 1), "last_hop_avg_ms": None}
|
||
return {}
|
||
|
||
|
||
def _summarize_ipinfo(raw: Any) -> dict[str, Any]:
|
||
if not isinstance(raw, dict):
|
||
return {}
|
||
return {
|
||
"ip": raw.get("ip"),
|
||
"city": raw.get("city"),
|
||
"region": raw.get("region"),
|
||
"country": raw.get("country"),
|
||
"org": raw.get("org") or raw.get("asn"),
|
||
}
|
||
|
||
|
||
def _systeminformation_payload() -> dict[str, Any]:
|
||
vm = psutil.virtual_memory()
|
||
cpu = psutil.cpu_percent(interval=0.2)
|
||
machine = machine_summary_payload()
|
||
disks: list[dict[str, Any]] = []
|
||
try:
|
||
for part in psutil.disk_partitions(all=False)[:4]:
|
||
try:
|
||
usage = psutil.disk_usage(part.mountpoint)
|
||
disks.append(
|
||
{
|
||
"mount": part.mountpoint,
|
||
"percent": usage.percent,
|
||
"free_gb": round(usage.free / (1024**3), 2),
|
||
}
|
||
)
|
||
except (PermissionError, OSError):
|
||
continue
|
||
except Exception:
|
||
pass
|
||
net_io: dict[str, Any] = {}
|
||
try:
|
||
io = psutil.net_io_counters()
|
||
net_io = {
|
||
"bytes_sent_mb": round(io.bytes_sent / (1024**2), 1),
|
||
"bytes_recv_mb": round(io.bytes_recv / (1024**2), 1),
|
||
}
|
||
except Exception:
|
||
pass
|
||
return {
|
||
"provider": "sebhildebrandt/systeminformation",
|
||
"ok": True,
|
||
"host": platform.node(),
|
||
"platform": platform.platform(),
|
||
"machine": machine,
|
||
"cpu_logical": psutil.cpu_count(logical=True),
|
||
"cpu_percent": round(float(cpu), 1),
|
||
"memory_total_gb": round(vm.total / (1024**3), 2),
|
||
"memory_used_gb": round(vm.used / (1024**3), 2),
|
||
"memory_percent": round(float(vm.percent), 1),
|
||
"disks": disks,
|
||
"net_io": net_io,
|
||
}
|
||
|
||
|
||
def _speedtest_payload() -> dict[str, Any]:
|
||
if _cli_available("speedtest"):
|
||
result = _run_cli(
|
||
["speedtest", "--accept-license", "--accept-gdpr", "-f", "json"],
|
||
timeout=18.0,
|
||
)
|
||
parsed = _safe_json_parse(result.get("stdout", ""))
|
||
summary = _summarize_speedtest(parsed)
|
||
return {
|
||
"provider": "librespeed/speedtest",
|
||
"command": "speedtest",
|
||
"ok": result.get("ok", False),
|
||
"raw": parsed or result.get("stdout", ""),
|
||
"summary": summary,
|
||
"error": None if result.get("ok", False) else result.get("stderr", "") or "speedtest failed",
|
||
}
|
||
if _cli_available("speedtest-cli"):
|
||
result = _run_cli(["speedtest-cli", "--json"], timeout=18.0)
|
||
parsed = _safe_json_parse(result.get("stdout", ""))
|
||
summary = _summarize_speedtest(parsed)
|
||
return {
|
||
"provider": "librespeed/speedtest",
|
||
"command": "speedtest-cli",
|
||
"ok": result.get("ok", False),
|
||
"raw": parsed or result.get("stdout", ""),
|
||
"summary": summary,
|
||
"error": None if result.get("ok", False) else result.get("stderr", "") or "speedtest-cli failed",
|
||
}
|
||
return {
|
||
"provider": "librespeed/speedtest",
|
||
"command": None,
|
||
"ok": False,
|
||
"raw": None,
|
||
"error": "未检测到 speedtest 或 speedtest-cli",
|
||
}
|
||
|
||
|
||
def _mtr_payload() -> dict[str, Any]:
|
||
if not _cli_available("mtr"):
|
||
return {
|
||
"provider": "traviscross/mtr",
|
||
"ok": False,
|
||
"raw": None,
|
||
"error": "未检测到 mtr 命令",
|
||
}
|
||
result = _run_cli(["mtr", "--report", "--json", "-c", "5", "1.1.1.1"], timeout=18.0)
|
||
parsed = _safe_json_parse(result.get("stdout", ""))
|
||
summary = _summarize_mtr(parsed or result.get("stdout", ""))
|
||
return {
|
||
"provider": "traviscross/mtr",
|
||
"ok": result.get("ok", False),
|
||
"raw": parsed or result.get("stdout", ""),
|
||
"summary": summary,
|
||
"error": None if result.get("ok", False) else result.get("stderr", "") or "mtr failed",
|
||
}
|
||
|
||
|
||
def _ipinfo_payload() -> dict[str, Any]:
|
||
if not _cli_available("ipinfo"):
|
||
return {
|
||
"provider": "ipinfo/cli",
|
||
"ok": False,
|
||
"raw": None,
|
||
"error": "未检测到 ipinfo 命令",
|
||
}
|
||
result = _run_cli(["ipinfo", "--json"], timeout=10.0)
|
||
parsed = _safe_json_parse(result.get("stdout", ""))
|
||
summary = _summarize_ipinfo(parsed)
|
||
return {
|
||
"provider": "ipinfo/cli",
|
||
"ok": result.get("ok", False),
|
||
"raw": parsed or result.get("stdout", ""),
|
||
"summary": summary,
|
||
"error": None if result.get("ok", False) else result.get("stderr", "") or "ipinfo failed",
|
||
}
|
||
|
||
|
||
def _ipquality_toolkit_payload(quality: dict[str, Any]) -> dict[str, Any]:
|
||
script = Path(__file__).resolve().parent / "third_party" / "IPQuality" / "ip.sh"
|
||
err = quality.get("error")
|
||
info = quality.get("info") if isinstance(quality.get("info"), dict) else {}
|
||
summary = quality.get("summary") if isinstance(quality.get("summary"), dict) else {}
|
||
media = quality.get("media") if isinstance(quality.get("media"), dict) else {}
|
||
head = quality.get("head") if isinstance(quality.get("head"), dict) else {}
|
||
return {
|
||
"provider": "xykt/IPQuality",
|
||
"ok": not err,
|
||
"script": str(script),
|
||
"script_exists": script.is_file(),
|
||
"endpoint": "/ip_quality",
|
||
"summary": {
|
||
"ip": info.get("ip") or head.get("ip"),
|
||
"headline": summary.get("headline"),
|
||
"proxy_like": summary.get("proxy_like"),
|
||
"country": info.get("country"),
|
||
"organization": info.get("organization"),
|
||
"youtube": (media.get("Youtube") or {}).get("status_label"),
|
||
"tiktok": (media.get("TikTok") or {}).get("status_label"),
|
||
},
|
||
"error": err,
|
||
}
|
||
|
||
|
||
async def run_network_toolkit() -> dict[str, Any]:
|
||
timeout = httpx.Timeout(READ_TIMEOUT, connect=CONNECT_TIMEOUT)
|
||
async with httpx.AsyncClient(timeout=timeout, verify=True) as client:
|
||
egress, quality = await asyncio.gather(_probe_default_egress(client), run_ip_quality())
|
||
country_code = str(egress.get("country_code") or "").upper()
|
||
speed = _speedtest_payload()
|
||
mtr = _mtr_payload()
|
||
ipinfo = _ipinfo_payload()
|
||
return {
|
||
"providers": {
|
||
"systeminformation": _systeminformation_payload(),
|
||
"librespeed_speedtest": speed,
|
||
"mtr": mtr,
|
||
"ipinfo_cli": ipinfo,
|
||
"ipquality": _ipquality_toolkit_payload(quality),
|
||
"is_china_user": {
|
||
"provider": "yArna/isChinaUser",
|
||
"scope": "server_egress",
|
||
"ok": bool(country_code == "CN"),
|
||
"country_code": country_code or None,
|
||
"reason": "服务端出口 IP 国家码（前端 is-china-user 为浏览器侧信号）",
|
||
},
|
||
"webrtc": {
|
||
"provider": "pion/webrtc",
|
||
"ok": True,
|
||
"note": "前端执行 WebRTC candidate 泄漏检测",
|
||
},
|
||
"fingerprintjs": {
|
||
"provider": "fingerprintjs/fingerprintjs",
|
||
"ok": True,
|
||
"note": "前端 @fingerprintjs/fingerprintjs 生成 visitorId",
|
||
},
|
||
},
|
||
"meta": {
|
||
"at": time.time(),
|
||
"logic_version": 2,
|
||
},
|
||
}
|