#!/usr/bin/env python3
# Copyright (c) 2026 Petr Balvín <opensource@petrbalvin.org> (https://petrbalvin.org)
# SPDX-License-Identifier: MIT

"""Network diagnostics — latency, DNS, MTU, packet loss, dual-stack, and port checks.

No external dependencies — leverages system tools (ping, ip, ss, ifconfig,
netstat, sockstat) with fallback to pure stdlib where possible.

Supports Linux and FreeBSD (13+ recommended; on pre-13 FreeBSD the legacy
ping6 binary is used for IPv6 instead of the unified ``ping -6``).

Usage:
    network-diag.py                                    # full diagnostic
    network-diag.py --target cloudflare.com            # target a specific host
    network-diag.py --protocol v4                      # IPv4 only
    network-diag.py --protocol v6                      # IPv6 only
"""

from __future__ import annotations

import argparse
import ipaddress
import os
import re
import shutil
import socket
import statistics
import subprocess
import sys
import time
from concurrent.futures import Future, ThreadPoolExecutor, as_completed

BOLD = "\033[1m"
RED = "\033[31m"
GREEN = "\033[32m"
YELLOW = "\033[33m"
CYAN = "\033[36m"
DIM = "\033[2m"
RESET = "\033[0m"
VERSION = "2.1.1"

# Well-known hosts for connectivity tests
CONNECTIVITY_HOSTS = [
    ("cloudflare.com", 443),
    ("google.com", 443),
]

# Ping statistics differ between iputils (Linux) and FreeBSD:
#   Linux:   "5 packets transmitted, 5 received, 0% packet loss, time 4003ms"
#   FreeBSD: "5 packets transmitted, 5 packets received, 0.0% packet loss"
_PING_TIME_RE = re.compile(r"time=(\d+(?:\.\d+)?)")
_PING_RECEIVED_RE = re.compile(r"(\d+)\s+(?:packets?\s+)?received")


def detect_os() -> str:
    """Return 'linux' or 'freebsd' based on uname."""
    system = os.uname().sysname.lower()
    if system == "freebsd":
        return "freebsd"
    return "linux"  # default


def _status(msg: str) -> None:
    """Print progress message to stderr — stays on same line until ``_status_done``."""
    print(f"  {msg}...", end="", flush=True, file=sys.stderr)


def _status_done(msg: str = "done") -> None:
    """Complete the current progress line on stderr."""
    print(f" {msg}", file=sys.stderr)


def run(cmd: list[str], timeout: int = 15) -> tuple[str, str] | None:
    """Execute a command and return (stdout, stderr), or None if not installed.

    Sets LC_ALL=C (and LANG=C) for reliable English-language output parsing
    and decodes with ``errors="replace"`` so unexpected bytes never raise.
    Returns ("", "") on timeout and None only when the binary is missing, so
    callers can distinguish "tool unavailable" from "no output".
    """
    try:
        env = {**os.environ, "LANG": "C", "LC_ALL": "C"}
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            errors="replace",
            timeout=timeout,
            env=env,
        )
        return result.stdout, result.stderr
    except subprocess.TimeoutExpired:
        return "", ""
    except FileNotFoundError:
        return None


def _parse_ping_output(output: str) -> tuple[list[float], int | None]:
    """Parse ping output for per-reply latencies and the received counter.

    Returns (latencies in ms, received count).  The received count is None
    when no statistics line was recognised; callers should then fall back
    to len(latencies).
    """
    latencies = [float(m.group(1)) for m in _PING_TIME_RE.finditer(output)]
    received: int | None = None
    for line in output.split("\n"):
        if "transmitted" in line:
            match = _PING_RECEIVED_RE.search(line)
            if match:
                received = int(match.group(1))
            break
    return latencies, received


def _ping_result_to_dict(
    host: str,
    latencies: list[float],
    received: int,
    count: int,
    protocol: str,
) -> dict[str, object]:
    """Build a standardised ping result dictionary from parsed data.

    ``loss_pct`` is derived from the authoritative received counter, not
    from the number of parsed ``time=`` lines.
    """
    loss_pct = round((count - received) / count * 100, 1) if count > 0 else 100.0
    result: dict[str, object] = {
        "host": host,
        "protocol": protocol,
        "sent": count,
        "received": received,
        "loss_pct": loss_pct,
        "min_ms": 0.0,
        "avg_ms": 0.0,
        "max_ms": 0.0,
        "stddev_ms": 0.0,
    }
    if latencies:
        result["min_ms"] = round(min(latencies), 1)
        result["avg_ms"] = round(statistics.mean(latencies), 1)
        result["max_ms"] = round(max(latencies), 1)
        if len(latencies) > 1:
            result["stddev_ms"] = round(statistics.stdev(latencies), 1)
    return result


def _freebsd_major() -> int:
    """Parse the FreeBSD major version from uname release (99 if unsure)."""
    try:
        return int(os.uname().release.split("-")[0].split(".")[0])
    except (ValueError, IndexError):
        return 99  # assume a modern, unified ping when unparseable


def _ping_base_cmd(os_type: str, family: str) -> list[str] | None:
    """Return the ping command prefix for *family*, or None if unavailable.

    FreeBSD 13+ ships a unified ping understanding -4/-6; earlier releases
    need the legacy ping6 binary for IPv6.
    """
    if os_type == "freebsd" and _freebsd_major() < 13:
        if family == "v6":
            return ["ping6"] if shutil.which("ping6") else None
        return ["ping"]
    return ["ping", "-6" if family == "v6" else "-4"]


def _ping_wait_args(os_type: str, base_cmd: list[str], timeout: int) -> list[str]:
    """Per-reply wait flag for the given ping binary.

    Linux iputils ``-W`` takes seconds; FreeBSD ping ``-W`` takes
    milliseconds; legacy FreeBSD ping6 uses ``-x`` (milliseconds) instead.
    """
    if os_type == "freebsd":
        flag = "-x" if base_cmd[0] == "ping6" else "-W"
        return [flag, str(timeout * 1000)]
    return ["-W", str(timeout)]


def ping_host(
    host: str,
    os_type: str,
    count: int = 5,
    timeout: int = 5,
    protocol: str = "v4",
) -> dict[str, object]:
    """Ping a host over one address family (``"v4"`` or ``"v6"``).

    Uses the OS-appropriate binary and flags (see ``_ping_base_cmd``).
    When no suitable ping binary exists, the result carries a ``"note"``
    key explaining the gap instead of silently reporting zero replies.
    """
    base = _ping_base_cmd(os_type, protocol)
    if base is None:
        result = _ping_result_to_dict(host, [], 0, count, protocol)
        result["note"] = "no ping binary with IPv6 support found"
        return result
    cmd = [*base, "-c", str(count), *_ping_wait_args(os_type, base, timeout), "--", host]
    # The subprocess timeout must also cover ping's own DNS resolution time.
    res = run(cmd, timeout=count + timeout + 10)
    if res is None:
        result = _ping_result_to_dict(host, [], 0, count, protocol)
        result["note"] = f"'{base[0]}' not found"
        return result
    latencies, parsed_received = _parse_ping_output(res[0])
    received = parsed_received if parsed_received is not None else len(latencies)
    return _ping_result_to_dict(host, latencies, received, count, protocol)


def check_interface_info(os_type: str = "linux") -> dict[str, object]:
    """Collect network interface addresses and default routes."""
    info: dict[str, object] = {
        "interfaces": [],
        "default_route_v4": "",
        "default_route_v6": "",
        "error": "",
    }
    if os_type == "freebsd":
        return _check_interface_info_freebsd(info)
    return _check_interface_info_linux(info)


def _check_interface_info_linux(info: dict[str, object]) -> dict[str, object]:
    """Collect interface info on Linux using iproute2."""
    res = run(["ip", "-br", "addr", "show"])
    if res is None:
        info["error"] = "'ip' not found"
        return info
    for line in res[0].split("\n"):
        parts = line.split()
        if len(parts) >= 2 and parts[0] != "lo":
            info["interfaces"].append(
                {
                    "name": parts[0],
                    "state": parts[1],
                    "ips": parts[2:],
                }
            )

    route4 = run(["ip", "-4", "route", "show", "default"])
    if route4 and route4[0].strip():
        parts = route4[0].strip().split()
        for i, p in enumerate(parts):
            if p == "via" and i + 1 < len(parts):
                info["default_route_v4"] = parts[i + 1]
                break

    route6 = run(["ip", "-6", "route", "show", "default"])
    if route6 and route6[0].strip():
        parts = route6[0].strip().split()
        for i, p in enumerate(parts):
            if p == "via" and i + 1 < len(parts):
                info["default_route_v6"] = parts[i + 1]
                break
    return info


def _check_interface_info_freebsd(info: dict[str, object]) -> dict[str, object]:
    """Collect interface info on FreeBSD using ifconfig and netstat."""
    iflist = run(["ifconfig", "-l"])
    if iflist is None:
        info["error"] = "'ifconfig' not found"
        return info
    ifaces = iflist[0].strip().split()
    if not ifaces:
        return info

    for ifname in ifaces:
        if ifname == "lo0":
            continue
        iface = run(["ifconfig", ifname])
        if iface is None:
            continue
        lines = iface[0].split("\n")
        if not lines or not lines[0]:
            continue

        # Parse flags line: name: flags=...<UP,...>
        state = "DOWN"
        if "<UP," in lines[0] or "<UP>" in lines[0]:
            state = "UP"

        ips: list[str] = []
        for line in lines[1:]:
            stripped = line.strip()
            if stripped.startswith("inet "):
                # inet 192.168.1.100 netmask ...
                addr = stripped.split()[1]
                if addr != "127.0.0.1":
                    ips.append(addr)
            elif stripped.startswith("inet6 "):
                addr = stripped.split()[1].split("%")[0]  # strip %scope suffix
                # Skip link-local (fe80:) and loopback (::1)
                if addr.startswith("fe80:") or addr == "::1":
                    continue
                ips.append(addr)

        info["interfaces"].append({"name": ifname, "state": state, "ips": ips})

    # Default routes via netstat
    route = run(["netstat", "-rn"])
    if route is None:
        return info
    for line in route[0].split("\n"):
        if not line.startswith("default"):
            continue
        parts = line.split()
        if len(parts) < 2:
            continue
        gateway = parts[1].split("%")[0]  # strip %scope from link-local gateways
        # Determine v4 vs v6 by checking for colons
        if ":" in gateway:
            if not info["default_route_v6"]:
                info["default_route_v6"] = gateway
        elif not info["default_route_v4"]:
            info["default_route_v4"] = gateway

    return info


def check_connectivity() -> dict[str, object]:
    """TCP handshake probes to well-known hosts over both address families.

    This is the single place performing TCP probes; the per-family
    reachability sections reuse these results (see ``_reachability_report``).
    """
    info: dict[str, object] = {"v4": {}, "v6": {}}
    for host, port in CONNECTIVITY_HOSTS:
        info["v6"][host] = _tcp_probe(host, port, socket.AF_INET6)  # IPv6 first
        info["v4"][host] = _tcp_probe(host, port, socket.AF_INET)
    return info


def _tcp_probe(host: str, port: int, family: int) -> dict[str, object]:
    """Attempt a TCP connection and return timing info.

    Iterates all resolved candidates (in getaddrinfo order) and passes the
    full sockaddr tuple to connect(), preserving the IPv6 scope id.
    """
    try:
        addrs = socket.getaddrinfo(host, port, family=family, type=socket.SOCK_STREAM)
    except OSError:
        return {"reachable": False}
    for af, socktype, proto, _canonname, sockaddr in addrs:
        try:
            with socket.socket(af, socktype, proto) as sock:
                sock.settimeout(5)
                start = time.perf_counter()
                sock.connect(sockaddr)
                elapsed = round((time.perf_counter() - start) * 1000, 1)
                return {"reachable": True, "ip": str(sockaddr[0]), "ms": elapsed}
        except OSError:
            continue
    return {"reachable": False}


def check_dns(protocol: str = "auto") -> dict[str, object]:
    """Test DNS resolution latency for multiple domains.

    Tests both A (IPv4) and AAAA (IPv6) resolution when ``protocol`` is
    ``"auto"``.  Respects ``--protocol v4`` / ``--protocol v6`` to restrict
    to a single address family.
    """
    info: dict[str, object] = {
        "hostname": socket.gethostname(),
        "fqdn": "",
        "resolvers": [],
        "v4": {},
        "v6": {},
    }

    try:
        info["fqdn"] = socket.getfqdn()
    except OSError:
        pass

    domains = [host for host, _port in CONNECTIVITY_HOSTS]

    # IPv6 first
    if protocol in ("auto", "v6"):
        for domain in domains:
            info["v6"][domain] = _dns_resolve(domain, socket.AF_INET6)
    if protocol in ("auto", "v4"):
        for domain in domains:
            info["v4"][domain] = _dns_resolve(domain, socket.AF_INET)

    try:
        with open("/etc/resolv.conf") as f:
            for line in f:
                parts = line.split()
                if line.startswith("nameserver") and len(parts) >= 2:
                    info["resolvers"].append(parts[1])
    except OSError:
        pass

    return info


def _dns_resolve(domain: str, family: int) -> list[dict[str, object]]:
    """Resolve *domain* three times and return latency samples."""
    results: list[dict[str, object]] = []
    for _ in range(3):
        try:
            start = time.perf_counter()
            addr = socket.getaddrinfo(domain, None, family=family)
            elapsed = (time.perf_counter() - start) * 1000
            results.append(
                {
                    "ip": addr[0][4][0] if addr else "?",
                    "ms": round(elapsed, 1),
                }
            )
        except OSError:
            results.append({"ip": "N/A", "ms": -1})
    return results


def _family_addresses(
    iface_info: dict[str, object], family: str
) -> list[ipaddress.IPv4Address | ipaddress.IPv6Address]:
    """Usable addresses of *family* from collected interface data.

    Excludes loopback, link-local, unspecified and multicast addresses.
    Globally-scoped addresses (per the IANA registries in ``ipaddress``)
    come first, private/ULA ones after.
    """
    version = 6 if family == "v6" else 4
    global_addrs: list[ipaddress.IPv4Address | ipaddress.IPv6Address] = []
    other_addrs: list[ipaddress.IPv4Address | ipaddress.IPv6Address] = []
    interfaces = iface_info.get("interfaces") if isinstance(iface_info, dict) else None
    for ifc in interfaces or []:
        for raw in ifc.get("ips", []):
            try:
                ip = ipaddress.ip_interface(raw).ip
            except ValueError:
                continue
            if ip.version != version:
                continue
            if ip.is_loopback or ip.is_link_local or ip.is_unspecified or ip.is_multicast:
                continue
            (global_addrs if ip.is_global else other_addrs).append(ip)
    return global_addrs + other_addrs


def _reachability_report(
    family: str, iface_info: dict[str, object], probes: dict[str, object]
) -> dict[str, object]:
    """Build a per-family reachability section from interface data + probes.

    ``addr`` is the best address found on any interface (global scope
    preferred); ``is_global`` distinguishes it from private/ULA space.
    """
    addresses = _family_addresses(iface_info, family)
    best = addresses[0] if addresses else None
    return {
        "available": best is not None,
        "addr": str(best) if best is not None else "",
        "is_global": bool(best is not None and best.is_global),
        "test_sites": probes,
    }


def _resolve_probe_target(target: str, protocol: str) -> tuple[str | None, str | None]:
    """Resolve *target* to (family, numeric address), preferring IPv6.

    Returns (None, None) when no address of the requested family exists.
    Resolving here (instead of letting ping decide) makes the address
    family deterministic so MTU header arithmetic stays correct.
    """
    families = {"v4": socket.AF_INET, "v6": socket.AF_INET6}
    order = ["v6", "v4"] if protocol == "auto" else [protocol]
    for family in order:
        try:
            infos = socket.getaddrinfo(target, None, family=families[family])
        except OSError:
            continue
        if infos:
            return family, str(infos[0][4][0])
    return None, None


def check_mtu(
    target: str = "cloudflare.com",
    protocol: str = "auto",
    os_type: str = "linux",
) -> dict[str, object]:
    """Path MTU discovery using system ping with fragmentation disabled.

    Success is defined POSITIVELY (a parsed reply); failure patterns are
    matched in BOTH stdout and stderr, because iputils prints the local
    "message too long" error to stderr while the PING banner goes to
    stdout.  Header sizes are per family: 28 bytes for IPv4 (20 IP + 8
    ICMP), 48 for IPv6 (40 IPv6 + 8 ICMPv6).  On Linux uses ``ping -M
    do``; on FreeBSD ``ping -D``.
    """
    info: dict[str, object] = {"path_mtu": 0, "target": target, "method": ""}

    family, addr = _resolve_probe_target(target, protocol)
    if family is None or addr is None:
        info["method"] = "target did not resolve"
        return info
    header = 48 if family == "v6" else 28

    base = _ping_base_cmd(os_type, family)
    if base is None:
        info["method"] = "no ping binary with IPv6 support found"
        return info
    df_args = ["-D"] if os_type == "freebsd" else ["-M", "do"]
    wait_args = _ping_wait_args(os_type, base, 2)

    for size in (1500, 1400, 1300, 1280, 1200, 1000):
        payload = size - header
        cmd = [*base, *df_args, "-c", "1", *wait_args, "-s", str(payload), "--", addr]
        res = run(cmd, timeout=12)  # generous: covers the wait plus extra slack
        if res is None:
            info["method"] = f"'{base[0]}' not found"
            return info
        out, err = res
        received = _parse_ping_output(out)[1]
        if received and received > 0:
            info["path_mtu"] = size
            info["method"] = f"ping DF probe ({size}B OK)"
            break
        combined = (out + err).lower()
        if "frag needed" in combined or "message too long" in combined:
            continue  # definitively too big — try a smaller size
        # timeout, black-holed ICMP errors, or unreachable — try smaller anyway
    else:
        info["method"] = "ping DF probe (no probe size succeeded)"

    return info


def _parse_ss_output(stdout: str) -> list[dict[str, str]]:
    """Parse ``ss -tlnp`` output into a list of {addr, process, bucket} dicts."""
    entries: list[dict[str, str]] = []
    for line in stdout.split("\n")[1:]:
        if "LISTEN" not in line:
            continue
        parts = line.split()
        if len(parts) < 5:
            continue
        # ss -tlnp columns: State Recv-Q Send-Q LocalAddress:Port PeerAddress:Port Process
        #                           0      1       2             3             4      5+
        local = parts[3]
        # IPv6 addresses in ss output use bracket notation: [::]:port
        bucket = "tcp6" if local.startswith("[") else "tcp4"
        process = parts[-1] if len(parts) > 5 else ""
        entries.append({"addr": local, "process": process, "bucket": bucket})
    return entries


def check_listening_ports(os_type: str = "linux") -> dict[str, object]:
    """Show locally listening TCP ports, split by address family.

    On Linux, uses ``ss -tlnp`` for the port listing.  IPv6 entries are
    detected by the bracket notation (``[::]:port``) in the local address
    column.  When process info is missing (ports owned by other users),
    re-runs the command via ``sudo -n`` for non-interactive privilege
    escalation.  Entries that remain unresolvable are shown as
    ``(no permission)``.

    On FreeBSD, uses ``sockstat -l4 -l6``; dual-stack sockets (``tcp46``)
    are reported in their own bucket.
    """
    if os_type == "freebsd":
        return _check_listening_ports_freebsd()
    return _check_listening_ports_linux()


def _check_listening_ports_linux() -> dict[str, object]:
    """Listening ports on Linux via ss."""
    info: dict[str, object] = {"tcp4": [], "tcp6": [], "tcp46": [], "error": ""}
    res = run(["ss", "-tlnp"], timeout=5)
    if res is None:
        info["error"] = "'ss' not found"
        return info
    entries = _parse_ss_output(res[0])

    # If any process fields came back empty, try sudo for the full picture.
    if any(e["process"] == "" for e in entries):
        sudo_res = run(["sudo", "-n", "ss", "-tlnp"], timeout=5)
        if sudo_res and sudo_res[0]:
            sudo_entries = _parse_ss_output(sudo_res[0])
            # Merge: overwrite entries where the sudo output gave us a process name.
            sudo_map = {e["addr"]: e for e in sudo_entries}
            for e in entries:
                if e["process"] == "" and e["addr"] in sudo_map:
                    e["process"] = sudo_map[e["addr"]]["process"]

    for entry in entries:
        process = entry["process"] if entry["process"] else "(no permission)"
        info[entry["bucket"]].append({"addr": entry["addr"], "process": process})

    return info


def _check_listening_ports_freebsd() -> dict[str, object]:
    """Listening ports on FreeBSD via sockstat (tcp46 = dual-stack)."""
    info: dict[str, object] = {"tcp4": [], "tcp6": [], "tcp46": [], "error": ""}
    res = run(["sockstat", "-l4", "-l6"], timeout=5)
    if res is None:
        info["error"] = "'sockstat' not found"
        return info
    for line in res[0].split("\n")[1:]:  # skip header
        parts = line.split()
        if len(parts) < 6:
            continue
        proto = parts[4]  # tcp4, tcp6, or tcp46 (dual-stack)
        if proto not in ("tcp4", "tcp6", "tcp46"):
            continue
        info[proto].append({"addr": parts[5], "process": parts[1]})
    return info


def compute_grade(data: dict[str, object]) -> dict[str, object]:
    """Compute an A–F summary grade based on key metrics.

    Scoring factors (each 0–20 points, total 100):
        - Packet loss (lower is better)
        - Latency (lower is better)
        - DNS resolution speed (lower is better)
        - IPv4 reachability (presence of address + test-site reachability)
        - IPv6 / dual-stack support (presence of address + test-site reachability)

    Letter mapping: A ≥ 90, B ≥ 80, C ≥ 70, D ≥ 60, F < 60.
    """
    score = 0
    breakdown: dict[str, int] = {}

    # --- Packet loss (20 pts) ---
    ping = data.get("ping", {})
    loss = float(ping.get("loss_pct", 100))
    if loss <= 0:
        loss_score = 20
    elif loss <= 5:
        loss_score = 15
    elif loss <= 10:
        loss_score = 10
    elif loss <= 30:
        loss_score = 5
    else:
        loss_score = 0
    breakdown["packet_loss"] = loss_score
    score += loss_score

    # --- Latency (20 pts) ---
    avg_ms = float(ping.get("avg_ms", 0))
    if avg_ms <= 0:
        latency_score = 0
    elif avg_ms < 30:
        latency_score = 20
    elif avg_ms < 80:
        latency_score = 15
    elif avg_ms < 150:
        latency_score = 10
    elif avg_ms < 300:
        latency_score = 5
    else:
        latency_score = 0
    breakdown["latency"] = latency_score
    score += latency_score

    # --- DNS speed (20 pts) ---
    dns = data.get("dns", {})
    dns_samples = 0
    dns_total = 0.0
    for family in ("v6", "v4"):
        for domain_results in dns.get(family, {}).values():
            for r in domain_results:
                if r["ms"] > 0:
                    dns_samples += 1
                    dns_total += r["ms"]
    dns_score = 0
    if dns_samples > 0:
        dns_avg = dns_total / dns_samples
        if dns_avg < 20:
            dns_score = 20
        elif dns_avg < 50:
            dns_score = 15
        elif dns_avg < 100:
            dns_score = 10
        elif dns_avg < 200:
            dns_score = 5
    breakdown["dns_speed"] = dns_score
    score += dns_score

    # --- IPv4 reachability (20 pts) ---
    ipv4 = data.get("ipv4", {})
    v4_score = 0
    if ipv4.get("available"):
        v4_score += 10
        # bonus for each reachable test site
        sites = ipv4.get("test_sites", {})
        reachable_count = sum(1 for s in sites.values() if s.get("reachable"))
        v4_score += min(reachable_count * 5, 10)
    breakdown["ipv4_reachability"] = v4_score
    score += v4_score

    # --- IPv6 / dual-stack (20 pts) ---
    ipv6 = data.get("ipv6", {})
    v6_score = 0
    if ipv6.get("available"):
        v6_score += 10
        sites = ipv6.get("test_sites", {})
        reachable_count = sum(1 for s in sites.values() if s.get("reachable"))
        v6_score += min(reachable_count * 5, 10)
    breakdown["ipv6_dualstack"] = v6_score
    score += v6_score

    # Letter mapping
    if score >= 90:
        letter = "A"
    elif score >= 80:
        letter = "B"
    elif score >= 70:
        letter = "C"
    elif score >= 60:
        letter = "D"
    else:
        letter = "F"

    grade_color = GREEN if letter in ("A", "B") else YELLOW if letter in ("C", "D") else RED

    return {
        "grade": letter,
        "score": score,
        "max_score": 100,
        "colour": grade_color,
        "breakdown": breakdown,
    }


def _ms_colour(ms: float) -> str:
    """Return an ANSI colour code for a given latency value in ms."""
    if ms <= 0:
        return DIM
    if ms < 50:
        return GREEN
    if ms < 200:
        return YELLOW
    return RED


def _print_reachability(family: str, info: dict[str, object] | None) -> None:
    """Pretty-print one per-family reachability section."""
    if not info:
        return
    label = "IPv6" if family == "v6" else "IPv4"
    if not info.get("available"):
        print(f"  {YELLOW}○{RESET} No {label} address detected on any interface")
        return
    scope = "Global" if info.get("is_global") else "Private"
    print(f"  {GREEN}✓{RESET} {scope} {label} address: {info['addr']}")
    suffix = " over IPv6" if family == "v6" else ""
    for site, result in info.get("test_sites", {}).items():
        if result.get("reachable"):
            print(f"  {GREEN}✓{RESET} {site:<20s} {result['ms']}ms  ({result['ip']})")
        else:
            print(f"  {RED}✗{RESET} {site:<20s} unreachable{suffix}")


def print_report(data: dict[str, object]) -> None:
    """Render the human-readable diagnostic report to stdout."""
    grade_info = data.get("grade", {})
    grade_letter = str(grade_info.get("grade", "?"))
    grade_colour = str(grade_info.get("colour", RESET))

    # --- Header ---
    os_label = str(data.get("os", "Linux")).capitalize()
    print(f"\n{BOLD}{'═' * 60}{RESET}")
    print(f"{BOLD}  Network Diagnostics  v{VERSION}  — {os_label}{RESET}")
    print(f"{BOLD}{'═' * 60}{RESET}")
    print(f"  Hostname: {data['hostname']}")
    print(f"  Protocol: {data.get('protocol', 'auto')}")
    score = grade_info.get("score", "?")
    max_score = grade_info.get("max_score", 100)
    print(f"  Summary Grade: {grade_colour}{BOLD}{grade_letter}{RESET}  ({score}/{max_score})")

    # --- Interfaces ---
    print(f"\n{BOLD}── Interfaces ──{RESET}")
    iface = data.get("interfaces", {})
    if iface.get("error"):
        print(f"  {DIM}unavailable: {iface['error']}{RESET}")
    for ifc in iface.get("interfaces", []):
        state_color = GREEN if ifc["state"] == "UP" else RED
        ips = ", ".join(ifc["ips"])
        print(f"  {ifc['name']:<8s} {state_color}{ifc['state']:<5s}{RESET}  {ips}")
    if iface.get("default_route_v6"):
        print(f"  Default IPv6 route → {iface['default_route_v6']}")
    if iface.get("default_route_v4"):
        print(f"  Default IPv4 route → {iface['default_route_v4']}")

    # --- Internet connectivity ---
    print(f"\n{BOLD}── Internet Connectivity ──{RESET}")
    conn = data.get("connectivity", {})
    if conn.get("error"):
        print(f"  {DIM}unavailable: {conn['error']}{RESET}")
    elif conn:
        for family_label, fam_key in [("IPv6", "v6"), ("IPv4", "v4")]:
            fam_data = conn.get(fam_key, {})
            status_parts = []
            for host in fam_data:
                result = fam_data[host]
                if result.get("reachable"):
                    ms = result.get("ms", 0)
                    colour = _ms_colour(ms)
                    status_parts.append(f"{GREEN}✓{RESET} {host} {colour}{ms:.0f}ms{RESET}")
                else:
                    status_parts.append(f"{RED}✗{RESET} {host}")
            hosts_line = ", ".join(status_parts) if status_parts else f"{DIM}(none){RESET}"
            print(f"  {family_label}: {hosts_line}")

    # --- Latency (ICMP ping) ---
    print(f"\n{BOLD}── Latency (ICMP ping) ──{RESET}")
    _print_ping_block("IPv6", data.get("ping_v6"))
    _print_ping_block("IPv4", data.get("ping_v4"))

    # --- DNS ---
    print(f"\n{BOLD}── DNS Resolution ──{RESET}")
    dns = data.get("dns", {})
    if dns:
        print(f"  FQDN: {dns.get('fqdn', '?')}")
        if dns.get("resolvers"):
            print(f"  Resolvers: {', '.join(dns['resolvers'])}")
        for family_label, fam_key in [("AAAA (IPv6)", "v6"), ("A (IPv4)", "v4")]:
            domains = dns.get(fam_key, {})
            if not domains:
                continue
            print(f"  {BOLD}{family_label}:{RESET}")
            for domain, results in domains.items():
                valid = [r["ms"] for r in results if r["ms"] > 0]
                avg = statistics.mean(valid) if valid else 0
                colour = _ms_colour(avg)
                ips = [r["ip"] for r in results if r["ip"] != "N/A"]
                ip_str = ips[0] if ips else "N/A"
                print(f"    {domain:<25s} → {colour}{ip_str:<18s}{RESET} {avg:.0f}ms")

    # --- IPv6 reachability ---
    print(f"\n{BOLD}── IPv6 Reachability ──{RESET}")
    _print_reachability("v6", data.get("ipv6"))

    # --- IPv4 reachability ---
    print(f"\n{BOLD}── IPv4 Reachability ──{RESET}")
    _print_reachability("v4", data.get("ipv4"))

    # --- Path MTU ---
    print(f"\n{BOLD}── Path MTU ──{RESET}")
    mtu = data.get("mtu", {})
    if mtu:
        if mtu.get("path_mtu"):
            colour = YELLOW if mtu["path_mtu"] < 1500 else GREEN
            print(f"  {colour}{mtu['path_mtu']}B{RESET}  ({mtu.get('method', '')})")
        elif mtu.get("method"):
            print(f"  {DIM}{mtu['method']}{RESET}")
        elif mtu.get("error"):
            print(f"  {DIM}unavailable: {mtu['error']}{RESET}")

    # --- Listening ports ---
    print(f"\n{BOLD}── Listening Ports ──{RESET}")
    ports = data.get("listening", {})
    if ports:
        if ports.get("error"):
            print(f"  {DIM}unavailable: {ports['error']}{RESET}")
        else:
            total = sum(len(v) for v in ports.values() if isinstance(v, list))
            if total > 0:
                for version in ["tcp6", "tcp4", "tcp46"]:
                    for entry in ports.get(version, []):
                        print(f"  {version}  {entry['addr']:<22s}  {DIM}{entry['process']}{RESET}")
            else:
                print("  (none)")

    # --- Grade breakdown ---
    print(f"\n{BOLD}── Grade Breakdown ──{RESET}")
    bd = grade_info.get("breakdown", {})
    labels = {
        "packet_loss": "Packet loss",
        "latency": "Latency",
        "dns_speed": "DNS speed",
        "ipv6_dualstack": "IPv6 / dual-stack",
        "ipv4_reachability": "IPv4 reachability",
    }
    for key, label in labels.items():
        pts = bd.get(key, 0)
        bar = "█" * (pts // 2) + "░" * (10 - pts // 2)
        print(f"  {label:<20s} {bar} {pts}/20")

    # --- Footer ---
    print(f"\n{BOLD}{'═' * 60}{RESET}")
    print(
        f"  {BOLD}Overall Grade: {grade_colour}{grade_letter}{RESET}  "
        f"({grade_info.get('score', '?')}/{grade_info.get('max_score', 100)})"
    )
    print(f"{BOLD}{'═' * 60}{RESET}\n")


def _print_ping_block(label: str, ping_info: dict[str, object] | None) -> None:
    """Pretty-print a single ping result block for a given protocol."""
    if not isinstance(ping_info, dict):
        return
    if ping_info.get("received", 0) == 0:
        note = ping_info.get("note") or ping_info.get("error")
        suffix = f" ({note})" if note else ""
        print(f"  {label}: {RED}no response{suffix}{RESET}")
        return

    loss = float(ping_info.get("loss_pct", 0))
    loss_color = RED if loss > 10 else YELLOW if loss > 0 else GREEN
    avg = float(ping_info.get("avg_ms", 0))
    latency_color = _ms_colour(avg)
    print(
        f"  {label}: {ping_info.get('sent')}/{ping_info.get('received')} received"
        f"  Loss: {loss_color}{loss}%{RESET}"
        f"  Avg: {latency_color}{avg:.1f}ms{RESET}"
        f"  Min/Max: {ping_info.get('min_ms', 0):.1f}/{ping_info.get('max_ms', 0):.1f}ms"
        f"  σ: {ping_info.get('stddev_ms', 0):.1f}ms"
    )


def main() -> None:
    """Parse arguments, collect diagnostics, and print the report."""
    parser = argparse.ArgumentParser(
        description="Network diagnostics — latency, DNS, MTU, dual-stack, ports",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument(
        "--target",
        default="cloudflare.com",
        help="Target host for tests (default: cloudflare.com)",
    )
    parser.add_argument(
        "--protocol",
        choices=["auto", "v4", "v6"],
        default="auto",
        help="Address family: auto (dual-stack), v4 (IPv4 only), v6 (IPv6 only)",
    )
    parser.add_argument(
        "--version",
        action="version",
        version=f"%(prog)s {VERSION}",
    )
    args = parser.parse_args()

    target: str = args.target
    protocol: str = args.protocol

    os_type = detect_os()

    data: dict[str, object] = {
        "hostname": socket.gethostname(),
        "version": VERSION,
        "os": os_type,
        "protocol": protocol,
        "interfaces": {},
        "connectivity": {},
        "ping": {},
        "ping_v4": None,
        "ping_v6": None,
        "dns": {},
        "ipv4": {},
        "ipv6": {},
        "mtu": {},
        "listening": {},
        "grade": {},
    }

    # Phase 1: interface info (instant, sequential)
    _status("Checking interfaces")
    data["interfaces"] = check_interface_info(os_type)
    iface_error = data["interfaces"].get("error")
    _status_done(f"warning: {iface_error}" if iface_error else "done")

    # Phase 2: all other checks run in parallel
    with ThreadPoolExecutor(max_workers=6) as executor:
        future_map: dict[Future[object], tuple[str, str]] = {}

        # Connectivity (TCP handshakes; results shared with reachability)
        fut: Future[object] = executor.submit(check_connectivity)
        future_map[fut] = ("connectivity", "internet connectivity")

        # Ping(s) — honour protocol preference, IPv6 first
        if protocol in ("auto", "v6"):
            fut = executor.submit(ping_host, target, os_type, protocol="v6")
            future_map[fut] = ("ping_v6", "ping (IPv6)")
        if protocol in ("auto", "v4"):
            fut = executor.submit(ping_host, target, os_type, protocol="v4")
            future_map[fut] = ("ping_v4", "ping (IPv4)")

        fut = executor.submit(check_dns, protocol)
        future_map[fut] = ("dns", "DNS resolution")
        fut = executor.submit(check_mtu, target, protocol, os_type)
        future_map[fut] = ("mtu", "path MTU")
        fut = executor.submit(check_listening_ports, os_type)
        future_map[fut] = ("listening", "listening ports")

        # Collect results as they complete, showing progress on stderr
        print(f"  Running {len(future_map)} checks in parallel...", file=sys.stderr)
        for future in as_completed(future_map):
            key, label = future_map[future]
            try:
                data[key] = future.result()
            except Exception as exc:
                data[key] = {"error": str(exc)}
                print(f"    {label}... failed ({exc})", file=sys.stderr)
                continue
            note = data[key].get("note") if isinstance(data[key], dict) else None
            if note:
                print(f"    {label}... warning ({note})", file=sys.stderr)
            else:
                print(f"    {label}... done", file=sys.stderr)

    # Derive per-family reachability from interface data + shared probes
    conn = data.get("connectivity")
    probes_v4 = conn.get("v4", {}) if isinstance(conn, dict) else {}
    probes_v6 = conn.get("v6", {}) if isinstance(conn, dict) else {}
    data["ipv6"] = _reachability_report("v6", data["interfaces"], probes_v6)
    data["ipv4"] = _reachability_report("v4", data["interfaces"], probes_v4)

    # Post-process ping results: pick the best for the dual-stack summary
    if protocol == "v4":
        data["ping"] = data.get("ping_v4", {})
    elif protocol == "v6":
        data["ping"] = data.get("ping_v6", {})
    else:
        # Dual-stack: prefer IPv6 when it answered (IPv6-first)
        ping_v4 = data.get("ping_v4")
        ping_v6 = data.get("ping_v6")
        v6_recv = int(ping_v6.get("received", 0)) if isinstance(ping_v6, dict) else 0
        if v6_recv > 0 and isinstance(ping_v6, dict):
            data["ping"] = ping_v6
        elif isinstance(ping_v4, dict):
            data["ping"] = ping_v4
        else:
            data["ping"] = {}

    # Compute summary grade
    data["grade"] = compute_grade(data)

    print_report(data)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\nInterrupted.", file=sys.stderr)
        sys.exit(130)