""" 网络检测：Google（海外）与抖音（国内）可达性、解析 IP/地区、简易上下行 Mbps。 说明：上行依赖对端是否接受 POST，失败时返回 null 与原因，属正常现象。 """ from __future__ import annotations import asyncio import socket import time from typing import Any, Optional import httpx # ---------- 目标 ---------- 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,message,query" DOWNLOAD_CAP_BYTES = 600_000 UPLOAD_BYTES = 256 * 1024 CONNECT_TIMEOUT = 8.0 READ_TIMEOUT = 25.0 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 _country_for_ip(client: httpx.AsyncClient, ip: str) -> dict[str, Any]: if not ip: return {"country": None, "country_code": None, "error": "no ip"} try: r = await client.get(IP_API_TEMPLATE.format(ip=ip), timeout=6.0) r.raise_for_status() data = r.json() except Exception as e: return {"country": None, "country_code": None, "error": str(e)} if data.get("status") != "success": return { "country": None, "country_code": None, "error": data.get("message") or "lookup failed", } return { "country": data.get("country"), "country_code": data.get("countryCode"), "error": None, } 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 的国家码；若 API 失败则给 likely / unknown。 """ 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() ( (g_ok, g_ms, g_err), (d_ok, d_ms, d_err), g_geo, d_geo, ) = await asyncio.gather( _latency_get(client, GOOGLE_CHECK_URL), _latency_get(client, DOUYIN_CHECK_URL), g_country_task, d_country_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} g_code = g_geo.get("country_code") d_code = d_geo.get("country_code") routes_different: Optional[bool] = None routes_note = "" if g_code and d_code: routes_different = _is_probably_china_country(g_code) != _is_probably_china_country(d_code) routes_note = "依据解析 IP 的国家码：海外站点与国内站点出口不一致" if routes_different else ( "依据解析 IP：两国别相同或未区分境内外（若您期望分流，请检查代理/路由）" ) else: routes_note = "未能查询到双方 IP 国家，无法严格判断线路是否分流" if g_ip and d_ip and g_ip != d_ip: routes_note += "；两站点解析 IP 不同，通常走不同目标。" if g_res_err: routes_note += f" Google DNS: {g_res_err}" if d_res_err: routes_note += f" 抖音 DNS: {d_res_err}" # 测速（并行） 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(), }, "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, }, }