"""
网络检测：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,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,
            },
        }