joy-cli/scripts/manhuang/probe-proxy-path.py

#!/usr/bin/env python3
"""蛮荒 · 网络供应链探针（N0–N5，加深版）

- N0 订阅/配置元数据（hash，无明文 URL）
- N1 Clash/Meta controller 只读
- N2 节点延迟抽样（controller /delay API）
- N3 双出口身份：直连 vs mixed-port
- N4 profile 期望声明
- N5 串线 v1：global 模式风险 + 生产域经代理可达性（只做 HEAD/GET 健康，不爬）

输出：~/.cache/manhuang/net-path/status.json

用法:
  python3 scripts/manhuang/probe-proxy-path.py
  python3 scripts/manhuang/probe-proxy-path.py --deep   # 多节点 delay + 串线探测
  python3 scripts/manhuang/probe-proxy-path.py --json
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import socket
import ssl
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

HOME = Path.home()
CACHE_DIR = HOME / ".cache" / "manhuang" / "net-path"
STATUS_PATH = CACHE_DIR / "status.json"
HISTORY_DIR = CACHE_DIR / "history"
DEFAULT_CFG = HOME / ".config" / "manhuang" / "proxy-path.yml"
EXAMPLE_CFG = (
    Path(__file__).resolve().parents[2] / "configs" / "manhuang" / "proxy-path.example.yml"
)

DEFAULT_CONTROLLERS = [
    {"name": "clash-meta-9090", "kind": "clash-meta", "base_url": "http://127.0.0.1:9090"},
    {
        "name": "clash-verge-9097",
        "kind": "clash-meta",
        "base_url": "http://127.0.0.1:9097",
        "secret_env": "CLASH_CONTROLLER_SECRET",
    },
]

# 串线探测用：只打路径根/健康，不爬业务
PROD_PROBE_PATHS = {
    "default": "/",
}


def _now() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _load_yaml(path: Path) -> dict[str, Any]:
    if not path.is_file():
        return {}
    try:
        import yaml  # type: ignore

        data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
        return data if isinstance(data, dict) else {}
    except Exception:
        return {}


def _http_json(
    url: str,
    *,
    timeout: float = 2.5,
    headers: dict[str, str] | None = None,
    proxy: str | None = None,
    method: str = "GET",
) -> tuple[int, Any, int]:
    """返回 (http_code, body_or_error, elapsed_ms)。"""
    t0 = time.time()
    req = urllib.request.Request(
        url,
        headers=headers or {"User-Agent": "joy-manhuang-net-path/1.1"},
        method=method,
    )
    handlers = []
    if proxy:
        handlers.append(urllib.request.ProxyHandler({"http": proxy, "https": proxy}))
    else:
        handlers.append(urllib.request.ProxyHandler({}))
    # 不校验也有用；但我们用默认 SSL
    opener = urllib.request.build_opener(*handlers)
    try:
        with opener.open(req, timeout=timeout) as resp:
            body = resp.read().decode("utf-8", "replace")
            code = int(getattr(resp, "status", 200) or 200)
            ms = int((time.time() - t0) * 1000)
            try:
                return code, json.loads(body), ms
            except json.JSONDecodeError:
                return code, body[:300], ms
    except urllib.error.HTTPError as e:
        ms = int((time.time() - t0) * 1000)
        try:
            raw = e.read().decode("utf-8", "replace")
            data: Any = json.loads(raw) if raw else None
        except Exception:
            data = None
        return int(e.code), data, ms
    except Exception as e:
        ms = int((time.time() - t0) * 1000)
        return 0, {"error": str(e)[:160]}, ms


def _tcp_open(host: str, port: int, timeout: float = 0.8) -> bool:
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except OSError:
        return False


def _parse_base(url: str) -> tuple[str, int]:
    u = url.replace("http://", "").replace("https://", "").split("/")[0]
    if ":" in u:
        h, p = u.rsplit(":", 1)
        try:
            return h, int(p)
        except ValueError:
            return h, 0
    return u, 80


def _hash_short(s: str, n: int = 12) -> str:
    return hashlib.sha256(s.encode("utf-8", "replace")).hexdigest()[:n]


def _auth_headers(secret_env: str | None) -> dict[str, str]:
    h = {"User-Agent": "joy-manhuang-net-path/1.1"}
    if secret_env:
        sec = os.environ.get(secret_env, "").strip()
        if sec:
            h["Authorization"] = f"Bearer {sec}"
    return h


def probe_controller(ctrl: dict[str, Any]) -> dict[str, Any]:
    name = str(ctrl.get("name") or "controller")
    base = str(ctrl.get("base_url") or "").rstrip("/")
    kind = str(ctrl.get("kind") or "clash-meta")
    secret_env = ctrl.get("secret_env")
    if not base:
        return {"name": name, "ok": False, "error": "no_base_url"}

    host, port = _parse_base(base)
    port_open = _tcp_open(host, port) if port else False
    out: dict[str, Any] = {
        "name": name,
        "kind": kind,
        "base_url": base,
        "port": port,
        "port_open": port_open,
        "ok": False,
    }
    if not port_open:
        out["error"] = "port_closed"
        return out

    headers = _auth_headers(str(secret_env) if secret_env else None)
    code, ver, _ = _http_json(f"{base}/version", headers=headers)
    out["version_http"] = code
    if code in (401, 403):
        out["error"] = "auth_required"
        out["hint"] = f"set env {secret_env or 'CLASH_CONTROLLER_SECRET'}"
        return out
    if code != 200:
        out["error"] = f"version_http_{code}"
        if isinstance(ver, dict) and ver.get("error"):
            out["detail"] = ver.get("error")
        return out

    if isinstance(ver, dict):
        out["meta"] = bool(ver.get("meta"))
        out["version"] = str(ver.get("version") or "")[:40]

    code_c, cfg, _ = _http_json(f"{base}/configs", headers=headers)
    mode = None
    mixed_port = None
    if code_c == 200 and isinstance(cfg, dict):
        mode = cfg.get("mode")
        mixed_port = cfg.get("mixed-port") or cfg.get("port") or cfg.get("socks-port")
        out["mode"] = mode
        out["mixed_port"] = mixed_port
        out["allow_lan"] = cfg.get("allow-lan")
        out["log_level"] = cfg.get("log-level")

    code_p, proxies, _ = _http_json(f"{base}/proxies", headers=headers)
    proxy_summary: dict[str, Any] = {}
    raw_px: dict[str, Any] = {}
    if code_p == 200 and isinstance(proxies, dict):
        px = proxies.get("proxies") or {}
        raw_px = px if isinstance(px, dict) else {}
        by_type: dict[str, int] = {}
        for v in raw_px.values():
            if not isinstance(v, dict):
                continue
            t = str(v.get("type") or "other")
            by_type[t] = by_type.get(t, 0) + 1
        g = raw_px.get("GLOBAL") if isinstance(raw_px.get("GLOBAL"), dict) else {}
        now = g.get("now") if g else None
        proxy_summary = {
            "total": len(raw_px),
            "by_type": dict(sorted(by_type.items(), key=lambda x: -x[1])[:12]),
            "global_now": now,
            "global_type": (
                (raw_px.get(now) or {}).get("type")
                if now and isinstance(raw_px.get(now), dict)
                else None
            ),
            "selectors": sum(
                1 for v in raw_px.values() if isinstance(v, dict) and v.get("type") == "Selector"
            ),
            "leaf_nodes": sum(
                1
                for v in raw_px.values()
                if isinstance(v, dict)
                and v.get("type") in ("Trojan", "Vmess", "Vless", "Shadowsocks", "Hysteria", "Hysteria2", "WireGuard", "Tuic")
            ),
        }
        out["proxies"] = proxy_summary
        out["_px_names"] = list(raw_px.keys())  # 内部用，写盘前剔除
        out["_px_global_all"] = list(g.get("all") or []) if g else []

    out["ok"] = True
    out["headers_env"] = str(secret_env) if secret_env else None
    out["n1"] = {
        "controller_up": True,
        "mode": mode,
        "mixed_port": mixed_port,
        "nodes": proxy_summary.get("total"),
        "selected": proxy_summary.get("global_now"),
        "leaf_nodes": proxy_summary.get("leaf_nodes"),
    }
    return out


def probe_node_delays(
    base: str,
    headers: dict[str, str],
    names: list[str],
    *,
    timeout_ms: int = 5000,
    url: str = "http://www.gstatic.com/generate_204",
) -> list[dict[str, Any]]:
    """Clash /proxies/{name}/delay"""
    out: list[dict[str, Any]] = []
    for name in names:
        if not name or name in ("DIRECT", "REJECT", "REJECT-DROP", "Pass", "Compatible"):
            continue
        enc = urllib.parse.quote(name, safe="")
        q = urllib.parse.urlencode({"timeout": timeout_ms, "url": url})
        code, data, ms = _http_json(
            f"{base}/proxies/{enc}/delay?{q}",
            headers=headers,
            timeout=timeout_ms / 1000.0 + 1.5,
        )
        delay = None
        if isinstance(data, dict) and data.get("delay") is not None:
            try:
                delay = int(data["delay"])
            except (TypeError, ValueError):
                delay = None
        out.append({
            "name": name,
            "http": code,
            "delay_ms": delay,
            "probe_ms": ms,
            "ok": delay is not None and delay > 0 and delay < 10000,
        })
    return out


def _lookup_ip_meta(ip: str) -> dict[str, Any]:
    """出口 IP 的 geo/ASN：优先本地 mmdb，回退 ip-api.com（仅 IP 元数据）。"""
    if not ip or ip in ("—", "unknown"):
        return {}
    meta: dict[str, Any] = {"ip": ip}
    # 1) maxminddb + Clash Country.mmdb（可能只有 country）
    mmdb_paths = [
        HOME / "Library/Application Support/io.github.clash-verge-rev.clash-verge-rev/Country.mmdb",
        HOME / "Library/Application Support/com.metacubex.ClashX.meta/Country.mmdb",
        Path("/usr/local/share/GeoIP/GeoLite2-Country.mmdb"),
    ]
    try:
        import maxminddb  # type: ignore

        for mp in mmdb_paths:
            if not mp.is_file():
                continue
            try:
                with maxminddb.open_database(str(mp)) as reader:
                    rec = reader.get(ip)
                if isinstance(rec, dict):
                    country = (
                        (rec.get("country") or {}).get("iso_code")
                        or rec.get("country_code")
                        or (rec.get("registered_country") or {}).get("iso_code")
                    )
                    if country:
                        meta["country"] = country
                        meta["geo_source"] = f"mmdb:{mp.name}"
                break
            except Exception:
                continue
    except ImportError:
        pass

    # 2) ip-api 补 ASN/org/city（免费、无需 key；失败忽略）
    try:
        code, data, _ = _http_json(
            f"http://ip-api.com/json/{urllib.parse.quote(ip)}"
            f"?fields=status,country,countryCode,city,as,org,query",
            timeout=5.0,
            proxy=None,
        )
        if code == 200 and isinstance(data, dict) and data.get("status") == "success":
            meta.setdefault("country", data.get("countryCode") or data.get("country"))
            meta["city"] = data.get("city")
            meta["asn"] = data.get("as")
            meta["org"] = data.get("org")
            meta["geo_source"] = (meta.get("geo_source") or "") + "+ip-api"
    except Exception:
        pass
    return meta


def probe_system_proxy() -> dict[str, Any]:
    """macOS scutil --proxy：系统是否把流量指到 789x。"""
    out: dict[str, Any] = {"ok": False, "enabled": False}
    try:
        r = subprocess_run_scutil()
        out["raw_keys"] = sorted(r.keys())
        http_on = bool(r.get("HTTPEnable"))
        https_on = bool(r.get("HTTPSEnable"))
        socks_on = bool(r.get("SOCKSEnable"))
        out["http"] = {
            "enable": http_on,
            "host": r.get("HTTPProxy"),
            "port": r.get("HTTPPort"),
        }
        out["https"] = {
            "enable": https_on,
            "host": r.get("HTTPSProxy"),
            "port": r.get("HTTPSPort"),
        }
        out["socks"] = {
            "enable": socks_on,
            "host": r.get("SOCKSProxy"),
            "port": r.get("SOCKSPort"),
        }
        out["exceptions"] = list(r.get("ExceptionsList") or [])[:20]
        out["enabled"] = http_on or https_on or socks_on
        ports = []
        for k in ("HTTPPort", "HTTPSPort", "SOCKSPort"):
            if r.get(k) is not None:
                try:
                    ports.append(int(r[k]))
                except (TypeError, ValueError):
                    pass
        out["ports"] = sorted(set(ports))
        out["points_local_clash"] = any(
            str(r.get(h) or "") in ("127.0.0.1", "localhost")
            for h in ("HTTPProxy", "HTTPSProxy", "SOCKSProxy")
        )
        out["ok"] = True
    except Exception as e:
        out["error"] = str(e)[:120]
    return out


def subprocess_run_scutil() -> dict[str, Any]:
    import subprocess

    # scutil --proxy 输出类字典文本
    r = subprocess.run(
        ["scutil", "--proxy"],
        capture_output=True,
        text=True,
        timeout=5,
    )
    text = r.stdout or ""
    # 也试 plutil 风格：defaults 无。手写解析
    result: dict[str, Any] = {}
    # HTTPEnable : 1
    for line in text.splitlines():
        line = line.strip()
        if " : " not in line:
            continue
        k, v = line.split(" : ", 1)
        k, v = k.strip(), v.strip()
        if v.isdigit():
            result[k] = int(v)
        elif v in ("0", "1"):
            result[k] = int(v)
        else:
            result[k] = v
    # ExceptionsList 数组
    if "ExceptionsList" in text:
        exc = []
        in_arr = False
        for line in text.splitlines():
            s = line.strip()
            if s.startswith("ExceptionsList"):
                in_arr = True
                continue
            if in_arr:
                if s.startswith("}"):
                    break
                if " : " in s:
                    exc.append(s.split(" : ", 1)[1].strip())
        if exc:
            result["ExceptionsList"] = exc
    return result


def probe_node_count_delta(current_nodes: int | None, profile_files: int | None) -> dict[str, Any]:
    """对比上次 status / history 的节点数与 profile 文件数。"""
    prev_nodes = None
    prev_profiles = None
    prev_ts = None
    if STATUS_PATH.is_file():
        try:
            old = json.loads(STATUS_PATH.read_text(encoding="utf-8"))
            prev_nodes = (old.get("primary") or {}).get("nodes")
            prev_profiles = (old.get("n0_subscription_meta") or {}).get("profile_files")
            prev_ts = old.get("ts")
        except Exception:
            pass
    delta_nodes = None
    if current_nodes is not None and prev_nodes is not None:
        try:
            delta_nodes = int(current_nodes) - int(prev_nodes)
        except (TypeError, ValueError):
            delta_nodes = None
    delta_profiles = None
    if profile_files is not None and prev_profiles is not None:
        try:
            delta_profiles = int(profile_files) - int(prev_profiles)
        except (TypeError, ValueError):
            delta_profiles = None
    return {
        "nodes_now": current_nodes,
        "nodes_prev": prev_nodes,
        "nodes_delta": delta_nodes,
        "profiles_now": profile_files,
        "profiles_prev": prev_profiles,
        "profiles_delta": delta_profiles,
        "prev_ts": prev_ts,
        "changed": bool(
            (delta_nodes not in (None, 0)) or (delta_profiles not in (None, 0))
        ),
    }


def probe_egress(mixed_port: int | None) -> dict[str, Any]:
    result: dict[str, Any] = {
        "direct": None,
        "via_proxy": None,
        "same_as_direct": None,
        "compare": "unknown",
    }
    code, data, ms = _http_json("https://api.ipify.org?format=json", timeout=8.0, proxy=None)
    if code == 200 and isinstance(data, dict) and data.get("ip"):
        result["direct"] = {"ip": str(data["ip"]), "source": "ipify", "ms": ms}
    elif code == 200 and isinstance(data, str) and data.strip():
        result["direct"] = {"ip": data.strip()[:64], "source": "ipify-raw", "ms": ms}

    if mixed_port:
        proxy = f"http://127.0.0.1:{int(mixed_port)}"
        last_err = None
        for _ in range(2):
            code_p, data_p, ms_p = _http_json(
                "https://api.ipify.org?format=json", timeout=10.0, proxy=proxy
            )
            if code_p == 200 and isinstance(data_p, dict) and data_p.get("ip"):
                result["via_proxy"] = {
                    "ip": str(data_p["ip"]),
                    "source": "ipify",
                    "ms": ms_p,
                    "proxy": f"127.0.0.1:{mixed_port}",
                }
                break
            if code_p == 200 and isinstance(data_p, str) and data_p.strip():
                result["via_proxy"] = {
                    "ip": data_p.strip()[:64],
                    "source": "ipify-raw",
                    "ms": ms_p,
                    "proxy": f"127.0.0.1:{mixed_port}",
                }
                break
            last_err = (data_p or {}).get("error") if isinstance(data_p, dict) else f"http_{code_p}"
            time.sleep(0.3)
        if not result.get("via_proxy"):
            result["via_proxy"] = {
                "ok": False,
                "error": last_err or "proxy_egress_fail",
                "proxy": f"127.0.0.1:{mixed_port}",
            }

    dip = (result.get("direct") or {}).get("ip")
    pip = (result.get("via_proxy") or {}).get("ip")
    if dip and pip:
        result["same_as_direct"] = dip == pip
        result["compare"] = "same" if dip == pip else "diverged"
    elif mixed_port and not pip:
        result["compare"] = "proxy_fail"

    # geo/ASN  enrichment
    for key in ("direct", "via_proxy"):
        block = result.get(key)
        if isinstance(block, dict) and block.get("ip"):
            geo = _lookup_ip_meta(str(block["ip"]))
            if geo:
                block["geo"] = {
                    k: geo.get(k)
                    for k in ("country", "city", "asn", "org", "geo_source")
                    if geo.get(k)
                }
    return result


def _fetch_live_direct_suffixes(controller_base: str | None = None) -> set[str]:
    """从 live Clash /rules 读取已 DIRECT 的 Domain/DomainSuffix（小写）。"""
    base = (controller_base or "http://127.0.0.1:9090").rstrip("/")
    out: set[str] = set()
    code, data, _ = _http_json(f"{base}/rules", timeout=3.0, proxy=None)
    rules = data.get("rules") if isinstance(data, dict) else data
    if not isinstance(rules, list):
        return out
    for r in rules:
        if not isinstance(r, dict):
            continue
        if str(r.get("proxy") or "") != "DIRECT":
            continue
        typ = str(r.get("type") or "")
        payload = str(r.get("payload") or "").lower().strip()
        if payload and typ in ("DomainSuffix", "Domain", "DomainKeyword"):
            out.add(payload)
    return out


def _host_covered_by_direct(host: str, direct_set: set[str]) -> bool:
    h = host.lower().strip(".")
    if h in direct_set:
        return True
    parts = h.split(".")
    for i in range(len(parts) - 1):
        if ".".join(parts[i:]) in direct_set:
            return True
    return False


def probe_serial_v1(
    *,
    mode: str | None,
    selected: str | None,
    mixed_port: int | None,
    never_via_proxy: list[str],
    deep: bool,
    system_proxy: dict[str, Any] | None = None,
    controller_base: str | None = None,
) -> dict[str, Any]:
    """串线检测：策略 + scutil + live DIRECT 规则覆盖。

    mixed-port HTTP 成功 ≠ 走机场（rule+DIRECT 时仍 200）。
    主信号：mode/global + 系统代理 + 生产域是否已在 DIRECT 规则中。
    """
    findings: list[dict[str, Any]] = []
    risk = "ok"
    sp = system_proxy or {}
    direct_set = _fetch_live_direct_suffixes(controller_base)
    mode_l = str(mode or "").lower()

    # 系统代理 ON 是本机常态（关了上不了外网）；只有 global 时才与串线叠加成 crit。
    if not sp.get("enabled"):
        findings.append({
            "id": "system_proxy_off",
            "level": "warn",
            "title": "系统代理未开启",
            "detail": "本机外网依赖 Clash 系统代理；OFF 时浏览器/多数 App 可能上不了外网",
            "next": "打开 Clash「系统代理」；隔离靠 rule+生产 DIRECT，不要用关代理当隔离",
        })
        if risk != "crit":
            risk = "warn"
    elif sp.get("points_local_clash"):
        ports = sp.get("ports") or []
        if mode_l == "global":
            findings.append({
                "id": "system_proxy_to_clash",
                "level": "crit",
                "title": "系统代理 ON + Clash global（生产域会进节点）",
                "detail": f"scutil → 127.0.0.1 ports={ports} · mode=global selected={selected}",
                "next": "改 mode=rule 并加生产 DOMAIN-SUFFIX DIRECT（系统代理保持 ON）",
            })
            risk = "crit"
        else:
            findings.append({
                "id": "system_proxy_on_expected",
                "level": "info",
                "title": "系统代理 ON（常态）",
                "detail": (
                    f"scutil → 127.0.0.1 ports={ports} · mode={mode} · "
                    f"DIRECT规则 {len(direct_set)} 条 · 关代理会上不了外网"
                ),
                "next": "保持系统代理 ON；只靠 rule+生产 DIRECT 隔离神州",
            })
            # 不抬高 risk：ON 是约定，不是缺陷

    if mode_l == "global" and selected and selected not in ("DIRECT", "Reject", "REJECT"):
        findings.append({
            "id": "global_mode_via_node",
            "level": "crit" if sp.get("enabled") else "warn",
            "title": "Clash 为 global 且选中节点",
            "detail": f"mode=global now={selected}",
            "next": "改为 rule + 生产 DIRECT",
        })
        if sp.get("enabled"):
            risk = "crit"
        elif risk != "crit":
            risk = "warn"

    missing_direct: list[str] = []
    covered: list[str] = []
    for host in (never_via_proxy or [])[:12]:
        host = str(host).strip().lstrip(".")
        if not host or "/" in host:
            continue
        if _host_covered_by_direct(host, direct_set):
            covered.append(host)
        else:
            missing_direct.append(host)

    if missing_direct:
        findings.append({
            "id": "prod_missing_direct_rule",
            "level": "crit" if (sp.get("enabled") or mode_l == "global") else "warn",
            "title": f"生产域未在 DIRECT 规则中：{', '.join(missing_direct[:4])}",
            "detail": f"missing={missing_direct} covered={covered}",
            "next": "合并 clash-rules-shenzhou-direct.snippet.yaml 并 reload",
        })
        if sp.get("enabled") or mode_l == "global":
            risk = "crit"
        elif risk != "crit":
            risk = "warn"
    elif covered:
        findings.append({
            "id": "prod_direct_ok",
            "level": "info",
            "title": f"生产域已 DIRECT ×{len(covered)}",
            "detail": ", ".join(covered[:8]),
            "next": "保持 rule；订阅更新后复查",
        })

    serial_hits: list[dict[str, Any]] = []
    if deep and mixed_port and never_via_proxy:
        proxy = f"http://127.0.0.1:{int(mixed_port)}"
        for host in never_via_proxy[:8]:
            host = str(host).strip().lstrip(".")
            if not host or "/" in host:
                continue
            url = f"https://{host}/"
            code, _, ms = _http_json(url, timeout=6.0, proxy=proxy, method="GET")
            serial_hits.append({
                "host": host,
                "via_mixed_http": code,
                "ms": ms,
                "reached": code in (200, 204, 301, 302, 303, 307, 308, 401, 403, 404),
                "rule_direct": _host_covered_by_direct(host, direct_set),
            })

    return {
        "risk": risk,
        "findings": findings[:20],
        "prod_via_proxy": serial_hits,
        "prod_direct_covered": covered,
        "prod_direct_missing": missing_direct,
        "direct_rule_n": len(direct_set),
        "never_via_proxy": list(never_via_proxy or [])[:30],
        "system_proxy": {
            "enabled": sp.get("enabled"),
            "ports": sp.get("ports"),
            "points_local_clash": sp.get("points_local_clash"),
        },
        "policy": {
            "mode": mode,
            "selected": selected,
            "global_risk": mode_l == "global",
            "system_proxy_on": bool(sp.get("enabled")),
            "prod_direct_ok": bool(covered) and not missing_direct,
        },
    }


def load_config(path: Path | None) -> dict[str, Any]:
    cfg: dict[str, Any] = {}
    for p in (path, DEFAULT_CFG, EXAMPLE_CFG):
        if p and Path(p).is_file():
            cfg = _load_yaml(Path(p))
            cfg["_config_path"] = str(p)
            break
    if not cfg.get("controllers"):
        cfg["controllers"] = DEFAULT_CONTROLLERS
    return cfg


def _append_history(status: dict[str, Any]) -> None:
    try:
        HISTORY_DIR.mkdir(parents=True, exist_ok=True)
        day = datetime.now(timezone.utc).strftime("%Y%m%d")
        row = {
            "ts": status.get("ts"),
            "level": status.get("level"),
            "selected": (status.get("primary") or {}).get("selected"),
            "mode": (status.get("primary") or {}).get("mode"),
            "egress_proxy": ((status.get("n2_n3_egress") or {}).get("via_proxy") or {}).get("ip"),
            "egress_direct": ((status.get("n2_n3_egress") or {}).get("direct") or {}).get("ip"),
            "same": (status.get("n2_n3_egress") or {}).get("same_as_direct"),
            "delay_selected": (status.get("n2_delays") or {}).get("selected_ms"),
            "serial_risk": (status.get("n4_n5_isolation") or {}).get("risk"),
        }
        with (HISTORY_DIR / f"{day}.jsonl").open("a", encoding="utf-8") as f:
            f.write(json.dumps(row, ensure_ascii=False) + "\n")
    except OSError:
        pass


def run(config_path: Path | None = None, *, deep: bool = False) -> dict[str, Any]:
    t0 = time.time()
    cfg = load_config(config_path)
    controllers_cfg = [
        c for c in (cfg.get("controllers") or [])
        if isinstance(c, dict) and c.get("enabled", True)
    ] or DEFAULT_CONTROLLERS

    probed = [probe_controller(c) for c in controllers_cfg]
    # 去掉内部字段
    for c in probed:
        c.pop("_px_names", None)
        c.pop("_px_global_all", None)

    # 重新取 primary 的完整版（含内部名）— 再 probe 一次太贵；用第二次轻量逻辑
    primary_full = None
    for c in controllers_cfg:
        full = probe_controller(c)
        if full.get("ok"):
            primary_full = full
            break
    if primary_full is None and controllers_cfg:
        primary_full = probe_controller(controllers_cfg[0])

    primary = next((c for c in probed if c.get("ok")), None) or (probed[0] if probed else {})

    mixed = (primary_full or primary or {}).get("mixed_port")
    if not mixed:
        for cand in (7893, 7897, 7890, 7892, 7891):
            if _tcp_open("127.0.0.1", cand):
                mixed = cand
                break

    # N2 delays
    delays: dict[str, Any] = {"selected_ms": None, "samples": [], "median_ms": None}
    if primary_full and primary_full.get("ok"):
        base = str(primary_full.get("base_url") or "").rstrip("/")
        headers = _auth_headers(primary_full.get("headers_env"))
        # re-fetch names from full
        names_all = primary_full.get("_px_global_all") or []
        selected = (primary_full.get("proxies") or {}).get("global_now")
        sample_names: list[str] = []
        if selected:
            sample_names.append(str(selected))
        # deep: 多采几个叶节点
        leaf_types = {"Trojan", "Vmess", "Vless", "Shadowsocks", "Hysteria2", "Hysteria"}
        # need proxies map - re-get
        code_p, proxies, _ = _http_json(f"{base}/proxies", headers=headers)
        px = (proxies or {}).get("proxies") if isinstance(proxies, dict) else {}
        if deep and isinstance(px, dict):
            for n in names_all:
                if n == selected:
                    continue
                node = px.get(n) or {}
                if isinstance(node, dict) and node.get("type") in leaf_types:
                    sample_names.append(n)
                if len(sample_names) >= 6:
                    break
        elif selected:
            sample_names = [str(selected)]
        dlist = probe_node_delays(base, headers, sample_names[:6])
        delays["samples"] = dlist
        ok_d = [x["delay_ms"] for x in dlist if x.get("ok") and x.get("delay_ms")]
        if selected:
            for x in dlist:
                if x.get("name") == selected:
                    delays["selected_ms"] = x.get("delay_ms")
        if ok_d:
            ok_d_sorted = sorted(ok_d)
            delays["median_ms"] = ok_d_sorted[len(ok_d_sorted) // 2]
            delays["min_ms"] = ok_d_sorted[0]
            delays["max_ms"] = ok_d_sorted[-1]

    # strip internal
    if primary_full:
        primary_full.pop("_px_names", None)
        primary_full.pop("_px_global_all", None)
        primary_full.pop("headers_env", None)

    egress = probe_egress(int(mixed) if mixed else None)
    never = list(cfg.get("never_via_proxy") or cfg.get("never_probe") or [])
    selected = (primary or {}).get("proxies", {}).get("global_now") if primary else None
    if not selected and primary_full:
        selected = (primary_full.get("proxies") or {}).get("global_now")
    mode = (primary or {}).get("mode") or (primary_full or {}).get("mode")

    # N0 配置指纹（先于 delta）
    n0: dict[str, Any] = {"ok": False, "sources": []}
    verge = HOME / "Library/Application Support/io.github.clash-verge-rev.clash-verge-rev"
    for label, path in [
        ("clash-verge-config", verge / "config.yaml"),
        ("clash-verge-runtime", verge / "clash-verge.yaml"),
        ("profiles-yaml", verge / "profiles.yaml"),
    ]:
        if path.is_file():
            try:
                raw = path.read_bytes()
                n0["sources"].append({
                    "id": label,
                    "path_basename": path.name,
                    "bytes": len(raw),
                    "sha12": _hash_short(raw.decode("utf-8", "replace")),
                    "mtime": int(path.stat().st_mtime),
                })
            except OSError:
                pass
    n0["ok"] = bool(n0["sources"])
    if (verge / "profiles").is_dir():
        try:
            n0["profile_files"] = sum(1 for _ in (verge / "profiles").iterdir())
            n0["profiles_dir"] = True
        except OSError:
            n0["profiles_dir"] = True

    nodes_now = (primary.get("proxies") or {}).get("total") if primary else None
    n0_delta = probe_node_count_delta(nodes_now, n0.get("profile_files"))
    n0["delta"] = n0_delta

    sys_proxy = probe_system_proxy()
    ctrl_base = None
    if primary and primary.get("base_url"):
        ctrl_base = str(primary.get("base_url"))
    elif primary_full and primary_full.get("base_url"):
        ctrl_base = str(primary_full.get("base_url"))
    isolation = probe_serial_v1(
        mode=mode,
        selected=selected,
        mixed_port=int(mixed) if mixed else None,
        never_via_proxy=never,
        deep=deep,
        system_proxy=sys_proxy,
        controller_base=ctrl_base,
    )

    # overall level
    serial_risk = isolation.get("risk") or "ok"
    if not (primary and primary.get("ok")):
        overall = "err"
        headline = "未检测到可用 Clash controller"
    elif serial_risk == "crit":
        overall = "err"
        if sys_proxy.get("enabled"):
            headline = f"串线 · 系统代理→Clash({mode}) · {selected}"
        else:
            headline = f"串线风险 · {selected} 可摸到生产域"
    elif egress.get("compare") == "same":
        overall = "warn"
        headline = f"Clash up · {selected} · 出口≈直连"
    elif serial_risk == "warn" or str(mode).lower() == "global":
        overall = "warn"
        headline = f"Clash up · mode={mode} · 选中={selected}"
    else:
        overall = "ok"
        headline = f"Clash up · mode={mode} · 选中={selected}"

    if delays.get("selected_ms"):
        headline += f" · 延迟 {delays['selected_ms']}ms"
    if n0_delta.get("changed"):
        headline += f" · 节点Δ{n0_delta.get('nodes_delta')}"

    eg_block = egress.get("via_proxy") or egress.get("direct") or {}
    eg_ip = eg_block.get("ip") or "—"
    eg_geo = eg_block.get("geo") or {}
    eg_detail = {
        "same": "≈直连",
        "diverged": "经代理",
        "proxy_fail": "代理出口失败",
        "unknown": "未比较",
    }.get(str(egress.get("compare")), "未比较")
    if eg_geo.get("country") or eg_geo.get("asn"):
        eg_detail += f" · {eg_geo.get('country') or ''} {eg_geo.get('asn') or ''}".strip()

    cards = [
        {
            "id": "controller",
            "label": "控制面",
            "level": "ok" if primary and primary.get("ok") else "err",
            "value": (primary or {}).get("version") or "down",
            "detail": f"mode={mode} port={mixed}",
        },
        {
            "id": "selected",
            "label": "当前节点",
            "level": "ok" if primary and primary.get("ok") else "dim",
            "value": selected or "—",
            "detail": f"delay={delays.get('selected_ms') or '—'}ms · nodes={nodes_now}",
        },
        {
            "id": "egress",
            "label": "出口 IP",
            "level": (
                "warn" if egress.get("same_as_direct") is True
                else ("ok" if (egress.get("via_proxy") or {}).get("ip") else "dim")
            ),
            "value": eg_ip,
            "detail": eg_detail,
        },
        {
            "id": "sys_proxy",
            "label": "系统代理",
            # ON 是常态（关了上不了外网）；仅 ON+global 视为 err，OFF 为 warn
            "level": (
                "err" if (sys_proxy.get("enabled") and str(mode).lower() == "global")
                else ("ok" if sys_proxy.get("enabled") else "warn")
            ),
            "value": "ON" if sys_proxy.get("enabled") else "OFF",
            "detail": (
                f"ports={sys_proxy.get('ports') or []} · "
                + (
                    "global 时会串线"
                    if (sys_proxy.get("enabled") and str(mode).lower() == "global")
                    else ("必须 ON" if sys_proxy.get("enabled") else "关了上不了外网")
                )
            ),
        },
        {
            "id": "serial",
            "label": "串线",
            "level": "err" if serial_risk == "crit" else ("warn" if serial_risk == "warn" else "ok"),
            "value": serial_risk,
            "detail": f"findings={len(isolation.get('findings') or [])}",
        },
    ]

    # 可行动 actions（给 board / UI）
    actions: list[dict[str, Any]] = []
    for f in isolation.get("findings") or []:
        if f.get("level") in ("crit", "warn"):
            actions.append({
                "priority": 1 if f.get("level") == "crit" else 2,
                "kind": "net_path",
                "level": f.get("level"),
                "title": f.get("title"),
                "detail": f.get("detail"),
                "next": f.get("next"),
                "tab": "mh-channels",
                "actionable": True,
                "at": _now(),
                "evidence": [{"id": f.get("id")}],
            })
    if egress.get("compare") == "proxy_fail" and mixed:
        actions.append({
            "priority": 2,
            "kind": "net_path",
            "level": "warn",
            "title": f"mixed-port :{mixed} 出口探测失败",
            "detail": (egress.get("via_proxy") or {}).get("error"),
            "next": "检查系统代理/防火墙是否拦 127.0.0.1 mixed-port",
            "tab": "mh-channels",
            "actionable": True,
            "at": _now(),
        })
    if delays.get("selected_ms") and delays["selected_ms"] > 2000:
        actions.append({
            "priority": 2,
            "kind": "net_path",
            "level": "warn",
            "title": f"当前节点延迟偏高 {delays['selected_ms']}ms",
            "detail": f"selected={selected}",
            "next": "换低延迟节点或检查中继",
            "tab": "mh-channels",
            "actionable": True,
            "at": _now(),
        })
    if n0_delta.get("changed") and (abs(n0_delta.get("nodes_delta") or 0) >= 3):
        actions.append({
            "priority": 2,
            "kind": "net_path",
            "level": "warn",
            "title": f"订阅/节点数变化 Δnodes={n0_delta.get('nodes_delta')}",
            "detail": f"nodes {n0_delta.get('nodes_prev')}→{n0_delta.get('nodes_now')} · profiles Δ={n0_delta.get('profiles_delta')}",
            "next": "核对机场订阅是否更新/抽节点；见 net-path history",
            "tab": "mh-channels",
            "actionable": True,
            "at": _now(),
        })

    # path_id：controller+mode+selected+egress（无订阅明文）
    eg_ip = ""
    if isinstance(egress, dict):
        eg_ip = str(((egress.get("via_proxy") or egress.get("direct") or {}) or {}).get("ip") or "")
    path_raw = "|".join([
        str((primary or {}).get("name") or ""),
        str(mode or ""),
        str(selected or ""),
        str(mixed or ""),
        eg_ip,
    ])
    path_id = "path_" + hashlib.sha1(path_raw.encode()).hexdigest()[:12] if path_raw.strip("|") else None

    status = {
        "ts": _now(),
        "ok": overall != "err",
        "level": overall,
        "headline": headline,
        "elapsed_ms": int((time.time() - t0) * 1000),
        "deep": deep,
        "config_path": cfg.get("_config_path"),
        "path_id": path_id,
        "n0_subscription_meta": n0,
        "n1_controllers": probed,
        "primary": {
            "name": (primary or {}).get("name"),
            "ok": bool(primary and primary.get("ok")),
            "mode": mode,
            "mixed_port": mixed,
            "version": (primary or {}).get("version"),
            "selected": selected,
            "nodes": (primary.get("proxies") or {}).get("total") if primary else None,
            "leaf_nodes": (primary.get("proxies") or {}).get("leaf_nodes") if primary else None,
            "by_type": (primary.get("proxies") or {}).get("by_type") if primary else None,
            "selected_type": (primary.get("proxies") or {}).get("global_type") if primary else None,
        },
        "n2_delays": delays,
        "n2_n3_egress": egress,
        "n4_n5_isolation": isolation,
        "system_proxy": sys_proxy,
        "subscription_delta": n0_delta,
        "layers": {
            "N0": "ok" if n0.get("ok") else "missing",
            "N1": "ok" if primary and primary.get("ok") else "err",
            "N2": "ok" if delays.get("selected_ms") else "warn",
            "N3": "ok" if egress.get("compare") in ("same", "diverged") else "warn",
            # 系统代理 ON 是常态（关了上不了外网）
            "N4": "ok" if sys_proxy.get("enabled") else "warn",
            "N5": isolation.get("risk") or "unset",
        },
        "cards": cards,
        "actions": actions[:12],
    }

    CACHE_DIR.mkdir(parents=True, exist_ok=True)
    tmp = STATUS_PATH.with_suffix(".tmp")
    tmp.write_text(json.dumps(status, ensure_ascii=False, indent=2), encoding="utf-8")
    tmp.rename(STATUS_PATH)
    _append_history(status)

    # 串线 crit 时自动 ensure-direct（不关系统代理）
    if str(isolation.get("risk") or "") == "crit" and os.environ.get("MANHUANG_AUTO_ENSURE", "1") not in ("0", "false", "no"):
        try:
            root = Path(__file__).resolve().parents[2]
            ctl = root / "scripts" / "manhuang" / "clashctl"
            if ctl.is_file():
                subprocess.run(
                    ["bash", str(ctl), "ensure-direct"],
                    timeout=45,
                    capture_output=True,
                    text=True,
                    env={**os.environ, "CLASH_PREFER_HTTP": "1"},
                )
                status["auto_ensure"] = True
        except Exception as e:
            status["auto_ensure"] = False
            status["auto_ensure_error"] = str(e)[:120]

    return status


def main() -> int:
    ap = argparse.ArgumentParser(description="蛮荒网络供应链探针 N0–N5（加深）")
    ap.add_argument("--config", type=Path, default=None)
    ap.add_argument("--deep", action="store_true", help="多节点 delay + 生产域经代理可达")
    ap.add_argument("--json", action="store_true")
    args = ap.parse_args()
    st = run(args.config, deep=args.deep)
    if args.json:
        print(json.dumps(st, ensure_ascii=False, indent=2))
    else:
        print(f"[{st.get('level')}] {st.get('headline')}")
        print(f"  wrote {STATUS_PATH} ({st.get('elapsed_ms')}ms) deep={st.get('deep')}")
        for c in st.get("cards") or []:
            print(f"  - {c.get('label')}: {c.get('value')} ({c.get('detail')})")
        acts = st.get("actions") or []
        if acts:
            print(f"  actions: {len(acts)}")
            for a in acts[:5]:
                print(f"    [{a.get('level')}] {a.get('title')}")
    return 0 if st.get("ok") else 1


if __name__ == "__main__":
    sys.exit(main())