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