src/api/admin.py

"""简单的管理后台



提供三类配置能力：

1. 端口等基本设置

2. API 密钥管理

3. 订阅拉取 + 节点选择作为出站代理

"""

import asyncio
import base64
import json
import os
import re
import secrets
import time
from pathlib import Path
from typing import Any, Optional
from urllib.parse import urlparse, unquote, parse_qs

from fastapi import APIRouter, File, HTTPException, Request, UploadFile
from fastapi.responses import FileResponse
from pydantic import BaseModel

from src.core.auth import api_key_manager
from src.core.config import load_config
from src.utils.node_store import upsert_nodes, replace_node, set_nodes_enabled, delete_nodes, clear_nodes, load_nodes, save_nodes, load_health, load_enabled_nodes
from src.utils.logger import get_logger

logger = get_logger(__name__)

# ==================== 路径 ====================

_ROOT_DIR = Path(__file__).parent.parent.parent
CONFIG_FILE = _ROOT_DIR / "config" / "config.json"
API_KEYS_FILE = _ROOT_DIR / "config" / "api_keys.txt"
MODELS_FILE = _ROOT_DIR / "config" / "models.json"
STATIC_DIR = _ROOT_DIR / "static"

# ==================== 会话 ====================

_sessions: dict[str, float] = {}
SESSION_TTL = 7 * 24 * 3600  # 7 天


def _read_json(path: Path, default: Any) -> Any:
    try:
        if not path.exists():
            return default if not isinstance(default, dict) else dict(default)
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception as e:
        logger.error(f"读取 {path} 失败: {e}")
        return default if not isinstance(default, dict) else dict(default)


def _write_json(path: Path, data: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
    os.replace(tmp, path)


def _get_admin_password() -> str:
    env_pw = os.environ.get("ADMIN_PASSWORD", "").strip()
    if env_pw:
        return env_pw
    cfg = _read_json(CONFIG_FILE, {})
    return str(cfg.get("admin_password") or "").strip()


def ensure_admin_password() -> str:
    """启动时确保有管理员密码，没有就生成一个并写入配置"""
    env_pw = os.environ.get("ADMIN_PASSWORD", "").strip()
    if env_pw:
        logger.info("[Admin] 使用环境变量 ADMIN_PASSWORD 作为管理员密码")
        return env_pw

    cfg = _read_json(CONFIG_FILE, {})
    existing = str(cfg.get("admin_password") or "").strip()
    if existing:
        return existing

    new_pw = secrets.token_urlsafe(9)
    cfg["admin_password"] = new_pw
    _write_json(CONFIG_FILE, cfg)
    bar = "=" * 60
    logger.warning(bar)
    logger.warning("🔐 首次启动，已自动生成管理员密码：")
    logger.warning(f"   密码:    {new_pw}")
    logger.warning(f"   访问:    http://<host>:<port>/admin")
    logger.warning("   密码已写入 config/config.json，登录后可在面板修改")
    logger.warning(bar)
    return new_pw


def _issue_token() -> str:
    tok = secrets.token_urlsafe(32)
    _sessions[tok] = time.time() + SESSION_TTL
    return tok


def _check_token(token: Optional[str]) -> bool:
    if not token:
        return False
    exp = _sessions.get(token)
    if not exp:
        return False
    if exp < time.time():
        _sessions.pop(token, None)
        return False
    return True


def _require_auth(request: Request) -> None:
    token = None
    auth = request.headers.get("Authorization", "")
    if auth.lower().startswith("bearer "):
        token = auth[7:].strip()
    if not token:
        token = request.cookies.get("admin_token")
    if not _check_token(token):
        raise HTTPException(status_code=401, detail="未登录或会话已过期")


def _new_id(prefix: str = "sub") -> str:
    return f"{prefix}_{secrets.token_hex(6)}"


def _normalize_subscriptions(cfg: dict[str, Any]) -> list[dict[str, Any]]:
    raw = cfg.get("subscriptions")
    subs: list[dict[str, Any]] = []
    if isinstance(raw, list):
        for item in raw:
            if not isinstance(item, dict):
                continue
            url = str(item.get("url") or "").strip()
            if not url:
                continue
            subs.append({
                "id": str(item.get("id") or _new_id()),
                "name": str(item.get("name") or urlparse(url).netloc or "订阅"),
                "url": url,
                "enabled": bool(item.get("enabled", True)),
                "node_count": int(item.get("node_count") or 0),
                "updated_at": int(item.get("updated_at") or 0),
            })
    return subs


def _save_subscriptions(cfg: dict[str, Any], subs: list[dict[str, Any]]) -> None:
    cfg["subscriptions"] = subs


def _extract_subscription_urls(text: str) -> list[str]:
    """从粘贴文本中提取一个或多个 http(s) 订阅链接。"""
    seen: set[str] = set()
    urls: list[str] = []
    for match in re.findall(r"https?://[^\s\"'<>，；;]+", text):
        url = match.strip().rstrip(").]}、。")
        if url and url not in seen:
            seen.add(url)
            urls.append(url)
    return urls


def _subscription_name_from_url(url: str, index: int, existing_names: set[str]) -> str:
    """优先从链接中提取可区分标识，否则使用序号。"""
    parsed = urlparse(url)
    query = {k: (v[0] if v else "") for k, v in parse_qs(parsed.query).items()}
    for key in ("name", "tag", "remarks", "remark", "sub", "token", "id"):
        value = unquote(str(query.get(key) or "")).strip()
        if value:
            value = value[:24]
            break
    else:
        path_part = unquote(parsed.path.strip("/").split("/")[-1] if parsed.path.strip("/") else "").strip()
        host = parsed.netloc.split("@").pop().split(":")[0]
        value = path_part[:24] if path_part and path_part.lower() not in ("api", "subscribe", "subscription", "sub") else host
    base = value or f"订阅{index}"
    name = base
    suffix = 2
    while name in existing_names:
        name = f"{base}-{suffix}"
        suffix += 1
    existing_names.add(name)
    return name


# ==================== API 密钥文件 IO ====================

def _read_api_keys() -> list[dict[str, str]]:
    if not API_KEYS_FILE.exists():
        return []
    out: list[dict[str, str]] = []
    with open(API_KEYS_FILE, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            parts = line.split(":", 2)
            if len(parts) < 2:
                continue
            out.append({
                "name": parts[0].strip(),
                "key": parts[1].strip(),
                "description": parts[2].strip() if len(parts) >= 3 else "",
            })
    return out


def _write_api_keys(keys: list[dict[str, str]]) -> None:
    API_KEYS_FILE.parent.mkdir(parents=True, exist_ok=True)
    tmp = API_KEYS_FILE.with_suffix(API_KEYS_FILE.suffix + ".tmp")
    with open(tmp, "w", encoding="utf-8") as f:
        f.write("# 格式: name:key:description （由管理面板维护）\n")
        for k in keys:
            name = (k.get("name") or "").strip()
            key = (k.get("key") or "").strip()
            desc = (k.get("description") or "").strip()
            if not name or not key:
                continue
            if desc:
                f.write(f"{name}:{key}:{desc}\n")
            else:
                f.write(f"{name}:{key}\n")
    os.replace(tmp, API_KEYS_FILE)


# ==================== 订阅解析 ====================

# 协议前缀 token (避免源代码出现明文)
def _dt(s: str) -> str:
    return base64.b64decode(s).decode()

_SCHEMES = [
    _dt("dmxlc3M6Ly8="),        # a
    _dt("dm1lc3M6Ly8="),        # b
    _dt("dHJvamFuOi8v"),        # c
    _dt("c3M6Ly8="),             # d
    _dt("c3NyOi8v"),             # e
    _dt("aHlzdGVyaWEyOi8v"),     # f
    _dt("aHkyOi8v"),             # g (f 的别名)
    _dt("YW55dGxzOi8v"),         # h
    _dt("dHVpYzovLw=="),         # i
    _dt("aHlzdGVyaWE6Ly8="),     # j (legacy of f)
]
(_SCHEME_A, _SCHEME_B, _SCHEME_C, _SCHEME_D, _SCHEME_E,
 _SCHEME_F, _SCHEME_G, _SCHEME_H, _SCHEME_I, _SCHEME_J) = _SCHEMES
_DIRECT_SCHEMES = ("http://", "https://", "socks5://", "socks://")


def _try_b64decode(text: str) -> Optional[str]:
    s = text.strip().replace("\n", "").replace("\r", "").replace(" ", "")
    s = s.replace("-", "+").replace("_", "/")
    pad = len(s) % 4
    if pad:
        s += "=" * (4 - pad)
    try:
        decoded = base64.b64decode(s, validate=False).decode("utf-8", errors="replace")
        all_markers = _SCHEMES + list(_DIRECT_SCHEMES) + ["proxies:", "outbounds", "\"outbounds\"", "proxy-groups"]
        if any(p in decoded for p in all_markers):
            return decoded
    except Exception:
        return None
    return None


def _parse_b_type(uri: str) -> Optional[dict[str, Any]]:
    """解析 base64(JSON) 格式的节点 URI"""
    try:
        raw = uri.split("://", 1)[1]
        pad = len(raw) % 4
        if pad:
            raw += "=" * (4 - pad)
        data = json.loads(base64.b64decode(raw.replace("-", "+").replace("_", "/")).decode("utf-8", errors="replace"))
        return {
            "type": "B",
            "name": data.get("ps") or data.get("name") or f"{data.get('add')}:{data.get('port')}",
            "server": data.get("add", ""),
            "port": int(data.get("port", 0) or 0),
            "usable_as_proxy": False,
        }
    except Exception:
        return None


def _parse_d_type(uri: str) -> Optional[dict[str, Any]]:
    """解析 base64(method:pass)@host:port 格式"""
    try:
        compat_uri = _rewrite_mislabelled_ss_to_vless(uri)
        if compat_uri:
            node = _parse_url_like(compat_uri, "A")
            if node:
                node["raw_uri"] = compat_uri
            return node

        body = uri.split("://", 1)[1]
        name = ""
        if "#" in body:
            body, frag = body.split("#", 1)
            name = unquote(frag)
        if "@" in body:
            _, hp = body.rsplit("@", 1)
        else:
            pad = len(body) % 4
            if pad:
                body += "=" * (4 - pad)
            decoded = base64.b64decode(body.replace("-", "+").replace("_", "/")).decode("utf-8", errors="replace")
            _, hp = decoded.rsplit("@", 1) if "@" in decoded else ("", decoded)
        host, _, port = hp.rpartition(":")
        port = port.split("?")[0].split("/")[0]
        return {
            "type": "D",
            "name": name or f"{host}:{port}",
            "server": host,
            "port": int(port or 0),
            "usable_as_proxy": False,
        }
    except Exception:
        return None


def _rewrite_mislabelled_ss_to_vless(uri: str) -> Optional[str]:
    """兼容部分订阅把 VLESS 节点误标成 ss:// 的非标准格式。



    这类链接通常形如 ``ss://uuid@host:port?security=tls/reality&...``，

    userinfo 不是 Shadowsocks 的 ``method:password``，query 却包含 VLESS/Xray

    参数。v2rayN 会按可用协议兼容处理；这里转换成 vless:// 供后续 worker 使用。

    """
    try:
        if not uri.startswith(_SCHEME_D):
            return None
        u = urlparse(uri)
        if not u.username or ":" in unquote(u.username):
            return None
        query = parse_qs(u.query)
        security = str(query.get("security", [""])[0]).lower()
        encryption = str(query.get("encryption", [""])[0]).lower()
        vless_markers = {
            "flow", "security", "sni", "fp", "pbk", "sid", "type",
            "headerType", "host", "path", "encryption", "ech",
        }
        if security not in {"tls", "reality"} and encryption != "none" and not any(k in query for k in vless_markers):
            return None
        return _SCHEME_A + uri[len(_SCHEME_D):]
    except Exception:
        return None


def _parse_url_like(uri: str, label: str) -> Optional[dict[str, Any]]:
    try:
        u = urlparse(uri)
        name = unquote(u.fragment) if u.fragment else ""
        return {
            "type": label,
            "name": name or f"{u.hostname}:{u.port}",
            "server": u.hostname or "",
            "port": int(u.port or 0),
            "usable_as_proxy": False,
        }
    except Exception:
        return None


def _parse_e_type(uri: str) -> Optional[dict[str, Any]]:
    """解析 base64(host:port:...) 格式"""
    try:
        raw = uri.split("://", 1)[1]
        pad = len(raw) % 4
        if pad:
            raw += "=" * (4 - pad)
        decoded = base64.b64decode(raw.replace("-", "+").replace("_", "/")).decode("utf-8", errors="replace")
        main = decoded.split("/?")[0]
        parts = main.split(":")
        if len(parts) < 2:
            return None
        return {
            "type": "E",
            "name": f"{parts[0]}:{parts[1]}",
            "server": parts[0],
            "port": int(parts[1] or 0),
            "usable_as_proxy": False,
        }
    except Exception:
        return None


def _parse_http_socks(uri: str) -> Optional[dict[str, Any]]:
    """可直接作为出站代理"""
    try:
        u = urlparse(uri)
        scheme = u.scheme.lower()
        if not u.hostname:
            return None
        if scheme in {"http", "https"} and (u.path not in ("", "/") or u.query):
            # Clash/sing-box YAML 中经常有 DNS、规则集、测速等普通 URL，不能当作 HTTP 代理节点。
            return None
        port = int(u.port or (80 if scheme == "http" else 443 if scheme == "https" else 1080))
        return {
            "type": scheme,
            "name": f"{scheme}://{u.hostname}:{port}",
            "server": u.hostname or "",
            "port": port,
            "usable_as_proxy": True,
            "raw_uri": uri,
        }
    except Exception:
        return None


def _parse_subscription_text(text: str) -> list[dict[str, Any]]:
    nodes: list[dict[str, Any]] = []
    candidate_lines: list[str] = []
    for raw_line in text.replace("\r", "\n").splitlines():
        line = raw_line.strip().strip('"\'`,;')
        if not line or line.startswith(("#", "//")):
            continue
        candidate_lines.append(line)
        # Also extract proxy-share URI fragments from JSON/CSV/log files.
        # Do not extract generic http(s) fragments here: structured configs often
        # contain DNS/provider/documentation URLs that are not proxy nodes.
        for marker in _SCHEMES + ["socks5://", "socks://"]:
            start = 0
            while True:
                idx = line.find(marker, start)
                if idx < 0:
                    break
                frag = line[idx:].split()[0].strip('"\'`,;]}>)')
                candidate_lines.append(frag)
                start = idx + len(marker)

    seen_lines: set[str] = set()
    for line in candidate_lines:
        if line in seen_lines:
            continue
        seen_lines.add(line)
        node: Optional[dict[str, Any]] = None
        if line.startswith(_SCHEME_B):
            node = _parse_b_type(line)
        elif line.startswith(_SCHEME_D):
            node = _parse_d_type(line)
        elif line.startswith(_SCHEME_E):
            node = _parse_e_type(line)
        elif line.startswith(_SCHEME_C):
            node = _parse_url_like(line, "C")
        elif line.startswith(_SCHEME_A):
            node = _parse_url_like(line, "A")
        elif line.startswith(_SCHEME_F):
            node = _parse_url_like(line, "F")
        elif line.startswith(_SCHEME_G):
            node = _parse_url_like(line, "F")  # g 是 f 的别名
        elif line.startswith(_SCHEME_H):
            node = _parse_url_like(line, "H")
        elif line.startswith(_SCHEME_I):
            node = _parse_url_like(line, "I")
        elif line.startswith(_SCHEME_J):
            node = _parse_url_like(line, "J")
        elif line.startswith(_DIRECT_SCHEMES):
            node = _parse_http_socks(line)
        if node:
            node.setdefault("raw_uri", line)
            nodes.append(node)
    return nodes


def _parse_clash_yaml(text: str) -> list[dict[str, Any]]:
    """解析 Clash YAML，把每个 proxy 序列化成 clash:// 伪 URI"""
    try:
        import yaml  # type: ignore
    except Exception:
        return []
    try:
        data = yaml.safe_load(text)
    except Exception:
        return []
    if not isinstance(data, dict):
        return []
    proxies = data.get("proxies")
    if not isinstance(proxies, list):
        return []

    return _proxies_to_nodes(proxies, source_name="clash")


def _norm_port(value: Any) -> int:
    try:
        return int(value or 0)
    except Exception:
        return 0


def _normalize_tls_transport(proxy: dict[str, Any]) -> dict[str, Any]:
    """把 sing-box 风格 tls/transport 转成内部 Clash-like 字段。"""
    p = dict(proxy)
    tls = p.get("tls")
    if isinstance(tls, dict):
        if tls.get("enabled", True):
            p["tls"] = True
        if tls.get("server_name") and not p.get("servername"):
            p["servername"] = tls.get("server_name")
        if tls.get("insecure") is not None:
            p["skip-cert-verify"] = bool(tls.get("insecure"))
        if tls.get("alpn"):
            p["alpn"] = tls.get("alpn")
        utls = tls.get("utls")
        if isinstance(utls, dict) and utls.get("fingerprint"):
            p["client-fingerprint"] = utls.get("fingerprint")
        reality = tls.get("reality")
        if isinstance(reality, dict):
            p["reality-opts"] = {
                "public-key": reality.get("public_key") or reality.get("public-key") or "",
                "short-id": reality.get("short_id") or reality.get("short-id") or "",
            }

    transport = p.get("transport")
    if isinstance(transport, dict):
        t = str(transport.get("type") or "").lower()
        if t:
            p["network"] = "h2" if t == "http" else t
        if t == "ws":
            headers = transport.get("headers") if isinstance(transport.get("headers"), dict) else {}
            p["ws-opts"] = {"path": transport.get("path") or "/", "headers": headers}
        elif t == "grpc":
            p["grpc-opts"] = {"grpc-service-name": transport.get("service_name") or transport.get("serviceName") or ""}
        elif t == "http":
            p["http-opts"] = {"path": transport.get("path") or "/", "host": transport.get("host") or []}
    return p


def _normalize_proxy_dict(proxy: dict[str, Any], fallback_name: str = "") -> Optional[dict[str, Any]]:
    """归一化 Clash/Mihomo/sing-box proxy/outbound 字典。"""
    p = _normalize_tls_transport(proxy)
    t = str(p.get("type") or p.get("protocol") or "").lower().replace("shadowsocksr", "ssr")
    type_alias = {
        "ss": "shadowsocks",
        "shadow-tls": "shadowsocks",
        "hysteria2": "hysteria2",
        "hy2": "hysteria2",
    }
    t = type_alias.get(t, t)
    if t not in {"vless", "vmess", "trojan", "shadowsocks", "hysteria2", "anytls", "tuic", "hysteria"}:
        return None
    server = str(p.get("server") or p.get("address") or "").strip()
    port = _norm_port(p.get("port") if p.get("port") is not None else p.get("server_port"))
    if not server or not port:
        return None
    out = dict(p)
    out["type"] = t
    out["server"] = server
    out["port"] = port
    out["name"] = str(p.get("name") or p.get("tag") or fallback_name or f"{server}:{port}")
    if t == "shadowsocks" and p.get("method") and not p.get("cipher"):
        out["cipher"] = p.get("method")
    return out


def _proxy_to_node(proxy: dict[str, Any], source_name: str = "config") -> Optional[dict[str, Any]]:
    from src.transport.codec import clash_to_pseudo_uri, clash_type_letter
    p = _normalize_proxy_dict(proxy, source_name)
    if not p:
        return None
    letter = clash_type_letter(str(p.get("type") or ""))
    if letter == "?":
        return None
    try:
        pseudo = clash_to_pseudo_uri(p)
    except Exception:
        return None
    return {
        "type": letter,
        "name": p.get("name") or f"{p.get('server')}:{p.get('port')}",
        "server": p.get("server", ""),
        "port": int(p.get("port", 0) or 0),
        "usable_as_proxy": False,
        "raw_uri": pseudo,
        "subscription_name": source_name,
    }


def _proxies_to_nodes(proxies: list[Any], source_name: str = "config") -> list[dict[str, Any]]:
    nodes: list[dict[str, Any]] = []
    seen: set[str] = set()
    for item in proxies:
        if not isinstance(item, dict):
            continue
        node = _proxy_to_node(item, source_name)
        if not node:
            continue
        raw_uri = str(node.get("raw_uri") or "")
        if raw_uri in seen:
            continue
        seen.add(raw_uri)
        nodes.append(node)
    return nodes


def _stream_settings_to_clash(stream: dict[str, Any]) -> dict[str, Any]:
    out: dict[str, Any] = {}
    network = str(stream.get("network") or "tcp").lower()
    if network and network != "tcp":
        out["network"] = network
    security = str(stream.get("security") or "").lower()
    tls_settings = stream.get("tlsSettings") if isinstance(stream.get("tlsSettings"), dict) else {}
    reality_settings = stream.get("realitySettings") if isinstance(stream.get("realitySettings"), dict) else {}
    if security in ("tls", "reality"):
        out["tls"] = True
        servername = tls_settings.get("serverName") or reality_settings.get("serverName")
        if servername:
            out["servername"] = servername
        if tls_settings.get("allowInsecure") is not None:
            out["skip-cert-verify"] = bool(tls_settings.get("allowInsecure"))
        fp = reality_settings.get("fingerprint")
        if fp:
            out["client-fingerprint"] = fp
        if security == "reality":
            out["reality-opts"] = {
                "public-key": reality_settings.get("publicKey") or "",
                "short-id": reality_settings.get("shortId") or "",
            }
    if network == "ws":
        ws = stream.get("wsSettings") if isinstance(stream.get("wsSettings"), dict) else {}
        out["ws-opts"] = {"path": ws.get("path") or "/", "headers": ws.get("headers") or {}}
    elif network == "grpc":
        grpc = stream.get("grpcSettings") if isinstance(stream.get("grpcSettings"), dict) else {}
        out["grpc-opts"] = {"grpc-service-name": grpc.get("serviceName") or ""}
    elif network in ("http", "h2"):
        http = stream.get("httpSettings") if isinstance(stream.get("httpSettings"), dict) else {}
        out["http-opts"] = {"path": http.get("path") or "/", "host": http.get("host") or []}
    return out


def _parse_v2ray_outbounds(data: dict[str, Any]) -> list[dict[str, Any]]:
    """解析 v2rayN/Xray/V2Ray 导出的完整 JSON 配置 outbounds。"""
    outbounds = data.get("outbounds")
    if not isinstance(outbounds, list):
        return []
    proxies: list[dict[str, Any]] = []
    for ob in outbounds:
        if not isinstance(ob, dict):
            continue
        protocol = str(ob.get("protocol") or ob.get("type") or "").lower()
        if protocol in ("freedom", "blackhole", "dns", "direct", "block", "selector", "urltest"):
            continue
        settings = ob.get("settings") if isinstance(ob.get("settings"), dict) else {}
        stream = ob.get("streamSettings") if isinstance(ob.get("streamSettings"), dict) else {}
        base = _stream_settings_to_clash(stream)
        name = str(ob.get("tag") or "")
        if protocol in ("vmess", "vless"):
            vnext = settings.get("vnext") if isinstance(settings.get("vnext"), list) else []
            for v in vnext:
                if not isinstance(v, dict):
                    continue
                users = v.get("users") if isinstance(v.get("users"), list) else [{}]
                user = users[0] if users and isinstance(users[0], dict) else {}
                p = {**base, "type": protocol, "name": name, "server": v.get("address"), "port": v.get("port"), "uuid": user.get("id")}
                if user.get("flow"):
                    p["flow"] = user.get("flow")
                if protocol == "vmess":
                    p["cipher"] = user.get("security") or "auto"
                    p["alterId"] = user.get("alterId") or 0
                proxies.append(p)
        elif protocol == "trojan":
            servers = settings.get("servers") if isinstance(settings.get("servers"), list) else []
            for s in servers:
                if isinstance(s, dict):
                    proxies.append({**base, "type": "trojan", "name": name, "server": s.get("address"), "port": s.get("port"), "password": s.get("password")})
        elif protocol == "shadowsocks":
            servers = settings.get("servers") if isinstance(settings.get("servers"), list) else []
            for s in servers:
                if isinstance(s, dict):
                    proxies.append({"type": "shadowsocks", "name": name, "server": s.get("address"), "port": s.get("port"), "cipher": s.get("method"), "password": s.get("password")})
    return _proxies_to_nodes(proxies, source_name="v2ray-config")


def _collect_proxy_dicts(obj: Any) -> list[dict[str, Any]]:
    """从任意 JSON/YAML 结构中递归寻找代理客户端导出的节点字典。"""
    found: list[dict[str, Any]] = []
    if isinstance(obj, dict):
        if obj.get("type") and (obj.get("server") or obj.get("address")) and (obj.get("port") is not None or obj.get("server_port") is not None):
            found.append(obj)
        for key, value in obj.items():
            if key in ("proxy-groups", "rules", "route", "routing", "dns", "inbounds"):
                continue
            found.extend(_collect_proxy_dicts(value))
    elif isinstance(obj, list):
        for item in obj:
            found.extend(_collect_proxy_dicts(item))
    return found


def _parse_structured_config(text: str) -> list[dict[str, Any]]:
    """解析 Clash/Mihomo/sing-box/v2rayN/Xray 等完整配置。"""
    parsed_objects: list[Any] = []
    try:
        parsed_objects.append(json.loads(text))
    except Exception:
        pass
    try:
        import yaml  # type: ignore
        data = yaml.safe_load(text)
        if data is not None:
            parsed_objects.append(data)
    except Exception:
        pass

    best: list[dict[str, Any]] = []
    for data in parsed_objects:
        if not isinstance(data, dict):
            continue
        candidates: list[list[dict[str, Any]]] = []
        proxies = data.get("proxies")
        if isinstance(proxies, list):
            candidates.append(_proxies_to_nodes(proxies, source_name="clash-config"))
        outbounds = data.get("outbounds")
        if isinstance(outbounds, list):
            candidates.append(_parse_v2ray_outbounds(data))
            candidates.append(_proxies_to_nodes(outbounds, source_name="sing-box-config"))
        collected = _collect_proxy_dicts(data)
        if collected:
            candidates.append(_proxies_to_nodes(collected, source_name="structured-config"))
        current = max(candidates, key=len, default=[])
        if len(current) > len(best):
            best = current
    return best


async def _fetch_subscription(url: str) -> list[dict[str, Any]]:
    from curl_cffi import requests as ccrequests

    # 依次尝试不同客户端标识，取节点数最多的一次
    ua_candidates = [
        base64.b64decode("bWlob21vLzEuMTguNw==").decode(),
        base64.b64decode("Y2xhc2gubWV0YS8xLjE4Ljc=").decode(),
        base64.b64decode("c2luZy1ib3gvMS4xMS41").decode(),
        base64.b64decode("djJyYXlOLzYuNDI=").decode(),
    ]

    best: list[dict[str, Any]] = []
    last_err: str = ""
    for ua in ua_candidates:
        try:
            headers = {"User-Agent": ua, "Accept": "*/*"}
            async with ccrequests.AsyncSession(impersonate="chrome131") as sess:
                resp = await sess.get(url, headers=headers, timeout=20)
            if resp.status_code != 200:
                last_err = f"HTTP {resp.status_code}"
                continue
            body = resp.text
        except Exception as e:
            last_err = str(e)
            continue

        texts = [body]
        decoded = _try_b64decode(body)
        if decoded:
            texts.append(decoded)

        # 1. 优先按完整结构化配置解析，避免把 YAML/JSON 里的 DNS、provider、规则集 URL
        #    误识别为 http(s) 代理节点。
        nodes: list[dict[str, Any]] = []
        for text in texts:
            structured = _parse_structured_config(text)
            if len(structured) > len(nodes):
                nodes = structured

        # 2. 结构化配置解析不到时，再按 URI 列表或 base64 URI 订阅解析。
        if not nodes:
            for text in texts:
                parsed = _parse_subscription_text(text)
                if len(parsed) > len(nodes):
                    nodes = parsed

        # 3. 兼容旧逻辑：显式 Clash YAML fallback。
        if not nodes:
            for text in texts:
                if "proxies:" in text or text.lstrip().startswith("proxies:"):
                    parsed = _parse_clash_yaml(text)
                    if len(parsed) > len(nodes):
                        nodes = parsed

        if len(nodes) > len(best):
            best = nodes

    if not best:
        raise HTTPException(status_code=400, detail=f"无法解析订阅内容 ({last_err or '未知'})。支持订阅格式：URI 列表 / base64 / Clash/Mihomo YAML / sing-box JSON / V2Ray JSON")
    return best


def _parse_imported_node_file(text: str) -> list[dict[str, Any]]:
    """解析用户从代理客户端导出的优选配置文件。"""
    structured_candidates: list[list[dict[str, Any]]] = []

    structured = _parse_structured_config(text)
    if structured:
        structured_candidates.append(structured)

    decoded = _try_b64decode(text)
    if decoded:
        structured = _parse_structured_config(decoded)
        if structured:
            structured_candidates.append(structured)

    if structured_candidates:
        return max(structured_candidates, key=len)

    candidates: list[list[dict[str, Any]]] = []

    nodes = _parse_subscription_text(text)
    if nodes:
        candidates.append(nodes)

    if decoded:
        nodes = _parse_subscription_text(decoded)
        if nodes:
            candidates.append(nodes)
        if "proxies:" in decoded or decoded.lstrip().startswith("proxies:"):
            nodes = _parse_clash_yaml(decoded)
            if nodes:
                candidates.append(nodes)

    if "proxies:" in text or text.lstrip().startswith("proxies:"):
        nodes = _parse_clash_yaml(text)
        if nodes:
            candidates.append(nodes)

    try:
        data = json.loads(text)
        json_text = json.dumps(data, ensure_ascii=False)
        nodes = _parse_subscription_text(json_text)
        if nodes:
            candidates.append(nodes)
    except Exception:
        pass

    best = max(candidates, key=len, default=[])
    if not best:
        raise HTTPException(status_code=400, detail="无法解析配置文件。支持 URI/分享链接、base64、Clash/Mihomo YAML、sing-box JSON、v2rayN/Xray/V2Ray 完整配置。")
    return best


# ==================== 路由 ====================

router = APIRouter()


class LoginBody(BaseModel):
    password: str


class SettingsBody(BaseModel):
    port_api: Optional[int] = None
    debug: Optional[bool] = None
    admin_password: Optional[str] = None
    anti429_enabled: Optional[bool] = None
    anti429_target: Optional[str] = None
    parallel_pool_size: Optional[int] = None
    parallel_pool_max_size: Optional[int] = None
    parallel_worker_base_port: Optional[int] = None
    parallel_worker_port_span: Optional[int] = None
    business_session_concurrency_limit: Optional[int] = None
    candidate_queue_length: Optional[int] = None
    candidate_queue_rounds: Optional[int] = None
    node_retry_count: Optional[int] = None
    request_pool_deadline_seconds: Optional[float] = None
    stream_winner_stall_timeout_seconds: Optional[float] = None
    anti_tracking: Optional[bool] = None
    drop_max_tokens: Optional[bool] = None


class KeyBody(BaseModel):
    name: str
    key: str
    description: str = ""


class SubscriptionItemBody(BaseModel):
    name: str = ""
    url: str
    enabled: bool = True


class SubscriptionsBatchBody(BaseModel):
    text: str


class SubscriptionUpdateBody(BaseModel):
    name: Optional[str] = None
    url: Optional[str] = None
    enabled: Optional[bool] = None


class SubscriptionFetchBody(BaseModel):
    ids: list[str] = []
    mode: str = "selected"


class DeleteSubscriptionsBody(BaseModel):
    ids: list[str]


class ImportNodesTextBody(BaseModel):
    text: str
    source_name: str = "clipboard"


class AddNodeBody(BaseModel):
    raw_uri: str
    name: str = ""


class NodeUpdateBody(BaseModel):
    original_raw_uri: str
    raw_uri: Optional[str] = None
    name: Optional[str] = None
    server: Optional[str] = None
    port: Optional[int] = None
    type: Optional[str] = None
    usable_as_proxy: Optional[bool] = None
    enabled: Optional[bool] = None
    source: Optional[str] = None
    subscription_name: Optional[str] = None
    return_nodes: bool = True


class DeleteNodesBody(BaseModel):
    raw_uris: list[str]
    return_nodes: bool = True


class NodesEnabledBody(BaseModel):
    raw_uris: list[str]
    enabled: bool


@router.get("/admin")
async def admin_page() -> FileResponse:
    index = STATIC_DIR / "admin.html"
    if not index.exists():
        raise HTTPException(status_code=500, detail="admin.html 不存在")
    return FileResponse(
        str(index),
        media_type="text/html; charset=utf-8",
        headers={"Cache-Control": "no-store, no-cache, must-revalidate, max-age=0"},
    )



@router.post("/api/admin/login")
async def admin_login(body: LoginBody) -> dict[str, Any]:
    expected = _get_admin_password()
    if not expected:
        raise HTTPException(status_code=500, detail="管理员密码未初始化")
    if body.password != expected:
        await asyncio.sleep(0.5)  # 轻微延迟
        raise HTTPException(status_code=401, detail="密码错误")
    tok = _issue_token()
    return {"token": tok, "ttl_seconds": SESSION_TTL}


@router.post("/api/admin/logout")
async def admin_logout(request: Request) -> dict[str, str]:
    auth = request.headers.get("Authorization", "")
    if auth.lower().startswith("bearer "):
        _sessions.pop(auth[7:].strip(), None)
    return {"status": "ok"}


@router.get("/api/admin/settings")
async def get_settings(request: Request) -> dict[str, Any]:
    _require_auth(request)
    cfg = load_config()
    return {
        "port_api": cfg.get("port_api", 2156),
        "debug": bool(cfg.get("debug", False)),
        "admin_password_env_locked": bool(os.environ.get("ADMIN_PASSWORD", "").strip()),
        "anti429_enabled": bool(cfg.get("anti429_enabled", False)),
        "anti429_target": cfg.get("anti429_target", "system"),
        "parallel_pool_size": int(cfg.get("parallel_pool_size", 4)),
        "parallel_pool_max_size": int(cfg.get("parallel_pool_max_size", 12)),
        "parallel_worker_base_port": int(cfg.get("parallel_worker_base_port", 12080)),
        "parallel_worker_port_span": int(cfg.get("parallel_worker_port_span", 2000)),
        "business_session_concurrency_limit": int(cfg.get("business_session_concurrency_limit", 0)),
        "candidate_queue_length": int(cfg.get("candidate_queue_length", 80)),
        "candidate_queue_rounds": int(cfg.get("candidate_queue_rounds", 0)),
        "node_retry_count": int(cfg.get("node_retry_count", 0)),
        "request_pool_deadline_seconds": float(cfg.get("request_pool_deadline_seconds", 0)),
        "stream_winner_stall_timeout_seconds": float(cfg.get("stream_winner_stall_timeout_seconds", 0)),
        "anti_tracking": bool(cfg.get("anti_tracking", True)),
        "drop_max_tokens": bool(cfg.get("drop_max_tokens", True)),
    }


@router.put("/api/admin/settings")
async def update_settings(body: SettingsBody, request: Request) -> dict[str, Any]:
    _require_auth(request)
    cfg = _read_json(CONFIG_FILE, {})
    notes: list[str] = []

    if body.port_api is not None:
        if not (1 <= body.port_api <= 65535):
            raise HTTPException(status_code=400, detail="端口必须在 1-65535")
        if cfg.get("port_api") != body.port_api:
            notes.append("端口变更需要重启容器才能生效")
        cfg["port_api"] = body.port_api

    if body.debug is not None:
        if cfg.get("debug") != bool(body.debug):
            notes.append("debug 模式变更需要重启容器才能完全生效")
        cfg["debug"] = bool(body.debug)

    if body.admin_password is not None:
        if os.environ.get("ADMIN_PASSWORD", "").strip():
            raise HTTPException(status_code=400, detail="当前由环境变量 ADMIN_PASSWORD 锁定，无法在面板修改")
        new_pw = body.admin_password.strip()
        if len(new_pw) < 6:
            raise HTTPException(status_code=400, detail="密码至少 6 位")
        cfg["admin_password"] = new_pw
        notes.append("管理员密码已更新，下次登录生效")

    if body.anti429_enabled is not None:
        cfg["anti429_enabled"] = bool(body.anti429_enabled)

    if body.anti429_target is not None:
        if body.anti429_target not in ("system", "user"):
            raise HTTPException(status_code=400, detail="anti429_target 必须是 system 或 user")
        cfg["anti429_target"] = body.anti429_target

    if body.parallel_pool_max_size is not None:
        if body.parallel_pool_max_size < 1 or body.parallel_pool_max_size > 64:
            raise HTTPException(status_code=400, detail="parallel_pool_max_size 应在 1-64")
        cfg["parallel_pool_max_size"] = int(body.parallel_pool_max_size)

    if body.parallel_pool_size is not None:
        max_size = int(cfg.get("parallel_pool_max_size", 12) or 12)
        if body.parallel_pool_size < 1 or body.parallel_pool_size > max_size:
            raise HTTPException(status_code=400, detail=f"parallel_pool_size 应在 1-{max_size}")
        cfg["parallel_pool_size"] = int(body.parallel_pool_size)

    if body.parallel_worker_base_port is not None:
        if body.parallel_worker_base_port < 1 or body.parallel_worker_base_port > 65535:
            raise HTTPException(status_code=400, detail="parallel_worker_base_port 应在 1-65535")
        cfg["parallel_worker_base_port"] = int(body.parallel_worker_base_port)

    if body.parallel_worker_port_span is not None:
        if body.parallel_worker_port_span < 1 or body.parallel_worker_port_span > 20000:
            raise HTTPException(status_code=400, detail="parallel_worker_port_span 应在 1-20000")
        base_port = int(cfg.get("parallel_worker_base_port", 12080) or 12080)
        if base_port + int(body.parallel_worker_port_span) - 1 > 65535:
            raise HTTPException(status_code=400, detail="parallel_worker_base_port + parallel_worker_port_span 超出 65535")
        cfg["parallel_worker_port_span"] = int(body.parallel_worker_port_span)

    if body.business_session_concurrency_limit is not None:
        if body.business_session_concurrency_limit < 0 or body.business_session_concurrency_limit > 10000:
            raise HTTPException(status_code=400, detail="business_session_concurrency_limit 应在 0-10000，0 表示不限")
        cfg["business_session_concurrency_limit"] = int(body.business_session_concurrency_limit)

    if body.candidate_queue_length is not None:
        if body.candidate_queue_length < 1 or body.candidate_queue_length > 100000:
            raise HTTPException(status_code=400, detail="candidate_queue_length 应在 1-100000")
        cfg["candidate_queue_length"] = int(body.candidate_queue_length)

    if body.candidate_queue_rounds is not None:
        if body.candidate_queue_rounds < 0 or body.candidate_queue_rounds > 10000:
            raise HTTPException(status_code=400, detail="candidate_queue_rounds 应在 0-10000，0 表示不限")
        cfg["candidate_queue_rounds"] = int(body.candidate_queue_rounds)

    if body.node_retry_count is not None:
        if body.node_retry_count < 0 or body.node_retry_count > 100:
            raise HTTPException(status_code=400, detail="node_retry_count 应在 0-100")
        cfg["node_retry_count"] = int(body.node_retry_count)

    if body.request_pool_deadline_seconds is not None:
        if body.request_pool_deadline_seconds < 0 or body.request_pool_deadline_seconds > 3600:
            raise HTTPException(status_code=400, detail="request_pool_deadline_seconds 应在 0-3600，0 表示使用底层网络超时")
        cfg["request_pool_deadline_seconds"] = float(body.request_pool_deadline_seconds)

    if body.stream_winner_stall_timeout_seconds is not None:
        if body.stream_winner_stall_timeout_seconds < 0 or body.stream_winner_stall_timeout_seconds > 3600:
            raise HTTPException(status_code=400, detail="stream_winner_stall_timeout_seconds 应在 0-3600，0 表示关闭")
        cfg["stream_winner_stall_timeout_seconds"] = float(body.stream_winner_stall_timeout_seconds)

    if body.anti_tracking is not None:
        cfg["anti_tracking"] = bool(body.anti_tracking)

    if body.drop_max_tokens is not None:
        cfg["drop_max_tokens"] = bool(body.drop_max_tokens)

    _write_json(CONFIG_FILE, cfg)
    return {"status": "ok", "notes": notes}


@router.get("/api/admin/keys")
async def get_keys(request: Request) -> dict[str, Any]:
    _require_auth(request)
    return {"keys": _read_api_keys()}


@router.post("/api/admin/keys")
async def add_key(body: KeyBody, request: Request) -> dict[str, str]:
    _require_auth(request)
    name = body.name.strip()
    key = body.key.strip()
    if not name or not key:
        raise HTTPException(status_code=400, detail="name / key 不能为空")
    if ":" in name:
        raise HTTPException(status_code=400, detail="name 不能包含冒号")
    if not key.startswith("sk-"):
        raise HTTPException(status_code=400, detail="key 必须以 sk- 开头")

    keys = _read_api_keys()
    keys = [k for k in keys if k["name"] != name]  # 同名覆盖
    keys.append({"name": name, "key": key, "description": body.description or ""})
    _write_api_keys(keys)
    api_key_manager.load_keys()  # 热加载
    return {"status": "ok"}


@router.delete("/api/admin/keys/{name}")
async def delete_key(name: str, request: Request) -> dict[str, str]:
    _require_auth(request)
    keys = _read_api_keys()
    new_keys = [k for k in keys if k["name"] != name]
    if len(new_keys) == len(keys):
        raise HTTPException(status_code=404, detail="未找到该密钥")
    _write_api_keys(new_keys)
    api_key_manager.load_keys()
    return {"status": "ok"}


@router.get("/api/admin/models")
async def get_models(request: Request) -> dict[str, Any]:
    _require_auth(request)
    data = _read_json(MODELS_FILE, {"models": [], "alias_map": {}})
    return {
        "models": data.get("models", []),
        "alias_map": data.get("alias_map", {}),
    }


class ModelsBody(BaseModel):
    models: list[str] | None = None
    alias_map: dict[str, str] | None = None


@router.put("/api/admin/models")
async def update_models(body: ModelsBody, request: Request) -> dict[str, Any]:
    _require_auth(request)
    data = _read_json(MODELS_FILE, {"models": [], "alias_map": {}})
    if body.models is not None:
        cleaned = [m.strip() for m in body.models if m.strip()]
        if not cleaned:
            raise HTTPException(status_code=400, detail="models 列表不能为空")
        data["models"] = cleaned
    if body.alias_map is not None:
        data["alias_map"] = {k.strip(): v.strip() for k, v in body.alias_map.items() if k.strip() and v.strip()}
    _write_json(MODELS_FILE, data)
    return {"status": "ok"}


@router.get("/api/admin/subscriptions")
async def list_subscriptions(request: Request) -> dict[str, Any]:
    _require_auth(request)
    cfg = _read_json(CONFIG_FILE, {})
    subs = _normalize_subscriptions(cfg)
    if subs != cfg.get("subscriptions"):
        _save_subscriptions(cfg, subs)
        _write_json(CONFIG_FILE, cfg)
    return {"subscriptions": subs}


@router.post("/api/admin/subscriptions")
async def add_subscription(body: SubscriptionItemBody, request: Request) -> dict[str, Any]:
    _require_auth(request)
    url = body.url.strip()
    if not url.startswith(("http://", "https://")):
        raise HTTPException(status_code=400, detail="订阅地址必须是 http(s):// 开头")
    cfg = _read_json(CONFIG_FILE, {})
    subs = _normalize_subscriptions(cfg)
    existing = next((s for s in subs if s["url"] == url), None)
    if existing:
        existing["name"] = body.name.strip() or existing["name"]
        existing["enabled"] = body.enabled
        sub = existing
    else:
        sub = {
            "id": _new_id(),
            "name": body.name.strip() or urlparse(url).netloc or "订阅",
            "url": url,
            "enabled": body.enabled,
            "node_count": 0,
            "updated_at": 0,
        }
        subs.append(sub)
    _save_subscriptions(cfg, subs)
    _write_json(CONFIG_FILE, cfg)
    return {"status": "ok", "subscription": sub}


@router.post("/api/admin/subscriptions/batch")
async def add_subscriptions_batch(body: SubscriptionsBatchBody, request: Request) -> dict[str, Any]:
    _require_auth(request)
    urls = _extract_subscription_urls(body.text)
    if not urls:
        raise HTTPException(status_code=400, detail="未发现 http(s) 订阅链接")

    cfg = _read_json(CONFIG_FILE, {})
    subs = _normalize_subscriptions(cfg)
    by_url = {s["url"]: s for s in subs}
    existing_names = {str(s.get("name") or "") for s in subs}
    added: list[dict[str, Any]] = []
    updated: list[dict[str, Any]] = []

    for index, url in enumerate(urls, start=1):
        if url in by_url:
            sub = by_url[url]
            if not str(sub.get("name") or "").strip():
                sub["name"] = _subscription_name_from_url(url, index, existing_names)
            sub["enabled"] = True
            updated.append(sub)
            continue
        sub = {
            "id": _new_id(),
            "name": _subscription_name_from_url(url, len(subs) + 1, existing_names),
            "url": url,
            "enabled": True,
            "node_count": 0,
            "updated_at": 0,
        }
        subs.append(sub)
        by_url[url] = sub
        added.append(sub)

    _save_subscriptions(cfg, subs)
    _write_json(CONFIG_FILE, cfg)
    return {"status": "ok", "added": len(added), "updated": len(updated), "subscriptions": subs, "added_items": added, "updated_items": updated}


@router.put("/api/admin/subscriptions/{sub_id}")
async def update_subscription(sub_id: str, body: SubscriptionUpdateBody, request: Request) -> dict[str, Any]:
    _require_auth(request)
    cfg = _read_json(CONFIG_FILE, {})
    subs = _normalize_subscriptions(cfg)
    sub = next((s for s in subs if s["id"] == sub_id), None)
    if not sub:
        raise HTTPException(status_code=404, detail="未找到该订阅")
    if body.url is not None:
        url = body.url.strip()
        if not url.startswith(("http://", "https://")):
            raise HTTPException(status_code=400, detail="订阅地址必须是 http(s):// 开头")
        sub["url"] = url
    if body.name is not None:
        sub["name"] = body.name.strip() or urlparse(sub["url"]).netloc or "订阅"
    if body.enabled is not None:
        sub["enabled"] = bool(body.enabled)
    _save_subscriptions(cfg, subs)
    _write_json(CONFIG_FILE, cfg)
    return {"status": "ok", "subscription": sub}


@router.delete("/api/admin/subscriptions/{sub_id}")
async def delete_subscription(sub_id: str, request: Request) -> dict[str, Any]:
    _require_auth(request)
    cfg = _read_json(CONFIG_FILE, {})
    subs = _normalize_subscriptions(cfg)
    new_subs = [s for s in subs if s["id"] != sub_id]
    if len(new_subs) == len(subs):
        raise HTTPException(status_code=404, detail="未找到该订阅")
    _save_subscriptions(cfg, new_subs)
    _write_json(CONFIG_FILE, cfg)
    return {"status": "ok"}


@router.post("/api/admin/subscriptions/delete")
async def delete_subscriptions(body: DeleteSubscriptionsBody, request: Request) -> dict[str, Any]:
    _require_auth(request)
    targets = {i.strip() for i in body.ids if i.strip()}
    if not targets:
        raise HTTPException(status_code=400, detail="请至少选择一个订阅")
    cfg = _read_json(CONFIG_FILE, {})
    subs = _normalize_subscriptions(cfg)
    new_subs = [s for s in subs if s["id"] not in targets]
    deleted = len(subs) - len(new_subs)
    if not deleted:
        raise HTTPException(status_code=404, detail="未找到要删除的订阅")
    _save_subscriptions(cfg, new_subs)
    _write_json(CONFIG_FILE, cfg)
    return {"status": "ok", "deleted": deleted, "subscriptions": new_subs}


@router.post("/api/admin/subscriptions/fetch")
async def fetch_subscriptions(body: SubscriptionFetchBody, request: Request) -> dict[str, Any]:
    _require_auth(request)
    cfg = _read_json(CONFIG_FILE, {})
    subs = _normalize_subscriptions(cfg)
    if body.mode == "all":
        targets = subs
    elif body.mode == "enabled":
        targets = [s for s in subs if s.get("enabled")]
    else:
        selected = set(body.ids)
        targets = [s for s in subs if s["id"] in selected]
    if not targets:
        raise HTTPException(status_code=400, detail="请至少选择一个订阅")

    all_nodes: list[dict[str, Any]] = []
    seen: set[str] = set()
    errors: list[dict[str, str]] = []
    now = int(time.time())
    for sub in targets:
        try:
            nodes = await _fetch_subscription(sub["url"])
            sub["node_count"] = len(nodes)
            sub["updated_at"] = now
            for node in nodes:
                raw_uri = str(node.get("raw_uri") or "")
                if not raw_uri or raw_uri in seen:
                    continue
                seen.add(raw_uri)
                node["subscription_id"] = sub["id"]
                node["subscription_name"] = sub["name"]
                all_nodes.append(node)
        except Exception as e:
            detail = getattr(e, "detail", None) or str(e)
            errors.append({"id": sub["id"], "name": sub["name"], "error": str(detail)})
    _save_subscriptions(cfg, subs)
    _write_json(CONFIG_FILE, cfg)
    merged, added, updated = upsert_nodes(all_nodes, source="subscriptions")
    if not all_nodes:
        msg = errors[0]["error"] if errors else "未知"
        raise HTTPException(status_code=400, detail=f"无法解析所选订阅 ({msg})")
    return {
        "total": len(all_nodes),
        "added": added,
        "updated": updated,
        "usable_count": sum(1 for n in all_nodes if n.get("usable_as_proxy")),
        "subscription_count": len(targets),
        "nodes": merged,
        "errors": errors,
    }


@router.get("/api/admin/nodes")
async def get_unified_nodes(request: Request) -> dict[str, Any]:
    """返回统一节点集合与健康概览。"""
    _require_auth(request)
    nodes = load_nodes()
    health = load_health()
    return {
        "total": len(nodes),
        "nodes": nodes,
        "health": health,
    }


@router.post("/api/admin/nodes/import-text")
async def import_nodes_text(body: ImportNodesTextBody, request: Request) -> dict[str, Any]:
    """从剪切板/文本批量导入各种协议节点到统一节点集合。"""
    _require_auth(request)
    text = body.text.strip()
    if not text:
        raise HTTPException(status_code=400, detail="粘贴内容为空")
    if len(text.encode("utf-8", errors="ignore")) > 5 * 1024 * 1024:
        raise HTTPException(status_code=400, detail="粘贴内容过大，请控制在 5MB 内")
    nodes = _parse_imported_node_file(text)
    source_name = body.source_name.strip() or "clipboard"
    for node in nodes:
        node["source"] = "clipboard"
        node["subscription_name"] = source_name
    merged, added, updated = upsert_nodes(nodes, source="clipboard")
    return {
        "status": "ok",
        "total": len(nodes),
        "added": added,
        "updated": updated,
        "usable_count": sum(1 for n in nodes if n.get("usable_as_proxy")),
        "nodes": merged,
    }


@router.post("/api/admin/nodes")
async def add_unified_node(body: AddNodeBody, request: Request) -> dict[str, Any]:
    """手动添加单个节点到统一节点集合。"""
    _require_auth(request)
    raw_uri = body.raw_uri.strip()
    if not raw_uri:
        raise HTTPException(status_code=400, detail="节点 URI 为空")
    nodes = _parse_subscription_text(raw_uri)
    if not nodes:
        raise HTTPException(status_code=400, detail="无法解析该节点 URI")
    node = nodes[0]
    node["source"] = "manual"
    node["subscription_name"] = "手动添加"
    if body.name.strip():
        node["name"] = body.name.strip()
    merged, added, updated = upsert_nodes([node], source="manual")
    return {"status": "ok", "added": added, "updated": updated, "node": node, "nodes": merged}


@router.post("/api/admin/nodes/import")
async def import_nodes_file(request: Request, file: UploadFile = File(...)) -> dict[str, Any]:
    """导入 V2RayN/Clash/Mihomo 等客户端导出的优选节点配置。"""
    _require_auth(request)
    raw = await file.read()
    if len(raw) > 5 * 1024 * 1024:
        raise HTTPException(status_code=400, detail="配置文件过大，请控制在 5MB 内")
    text = raw.decode("utf-8", errors="replace")
    nodes = _parse_imported_node_file(text)
    for node in nodes:
        node["source"] = "import"
        node["subscription_name"] = file.filename or "import"
    merged, added, updated = upsert_nodes(nodes, source="import")
    return {
        "status": "ok",
        "total": len(nodes),
        "added": added,
        "updated": updated,
        "usable_count": sum(1 for n in nodes if n.get("usable_as_proxy")),
        "nodes": merged,
    }


@router.put("/api/admin/nodes")
async def update_unified_node(body: NodeUpdateBody, request: Request) -> dict[str, Any]:
    """编辑统一节点集合中的单个节点。"""
    _require_auth(request)
    updates: dict[str, Any] = {}
    for key in ("raw_uri", "name", "server", "port", "type", "usable_as_proxy", "enabled", "source", "subscription_name"):
        value = getattr(body, key)
        if value is not None:
            updates[key] = value.strip() if isinstance(value, str) else value
    if "raw_uri" in updates and not str(updates["raw_uri"]).strip():
        raise HTTPException(status_code=400, detail="raw_uri 不能为空")
    try:
        node = replace_node(body.original_raw_uri, updates)
    except KeyError:
        raise HTTPException(status_code=404, detail="未找到该节点")
    result: dict[str, Any] = {"status": "ok", "node": node}
    if body.return_nodes:
        result["nodes"] = load_nodes()
    return result


@router.post("/api/admin/nodes/enabled")
async def set_unified_nodes_enabled(body: NodesEnabledBody, request: Request) -> dict[str, Any]:
    """批量启用/禁用节点；只接收节点标识与目标状态，避免重复单节点请求。"""
    _require_auth(request)
    raw_uris = [str(uri).strip() for uri in body.raw_uris if str(uri).strip()]
    if not raw_uris:
        raise HTTPException(status_code=400, detail="请至少选择一个节点")
    matched, updated, enabled_count, total = set_nodes_enabled(raw_uris, bool(body.enabled))
    return {
        "status": "ok",
        "matched": matched,
        "updated": updated,
        "requested": len(set(raw_uris)),
        "enabled": bool(body.enabled),
        "enabled_count": enabled_count,
        "total": total,
    }


@router.post("/api/admin/nodes/delete")
async def delete_unified_nodes(body: DeleteNodesBody, request: Request) -> dict[str, Any]:
    """批量删除统一节点集合中的节点。"""
    _require_auth(request)
    deleted = delete_nodes(body.raw_uris)
    result: dict[str, Any] = {"status": "ok", "deleted": deleted}
    if body.return_nodes:
        result["nodes"] = load_nodes()
    return result


@router.delete("/api/admin/nodes/all")
async def clear_unified_nodes(request: Request) -> dict[str, Any]:
    """清空统一节点集合。"""
    _require_auth(request)
    deleted = clear_nodes()
    return {"status": "ok", "deleted": deleted, "nodes": []}


@router.post("/api/admin/stop-proxy")
async def stop_proxy(request: Request) -> dict[str, Any]:
    """停用全部统一节点，回到直连模式。"""
    _require_auth(request)
    try:
        nodes = load_nodes()
        for node in nodes:
            node["enabled"] = False
        save_nodes(nodes, source="manual")
    except Exception as e:
        logger.warning(f"同步停用全部节点状态失败: {e}")
    return {"status": "ok"}


@router.get("/api/admin/proxy-status")
async def proxy_status(request: Request) -> dict[str, Any]:
    _require_auth(request)
    return {
        "enabled_node_count": len(load_enabled_nodes()),
        "binary_available": True,
    }