#!/usr/bin/env python3 """ Fetch proxy subscriptions every run and test each node through a temporary mihomo instance. Overseas nodes open Google, domestic-return nodes open Baidu. """ from __future__ import annotations import argparse import base64 import contextlib import hashlib import html import ipaddress import json import os import shutil import socket import subprocess import sys import tempfile import time import traceback from datetime import datetime, timezone from pathlib import Path from typing import Any from urllib import error, parse, request SUPPORTED_TYPES = {"vmess", "vless", "trojan", "hysteria2", "anytls"} PROXY_ENV_KEYS = { "HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "NO_PROXY", "http_proxy", "https_proxy", "all_proxy", "no_proxy", } def load_yaml_module(): try: import yaml # type: ignore return yaml except Exception as exc: raise SystemExit( "Missing dependency: PyYAML\n" "Install it with:\n" " python3 -m pip install -r requirements.txt\n" f"Original error: {exc}" ) def no_proxy_env() -> dict[str, str]: env = os.environ.copy() for key in PROXY_ENV_KEYS: env.pop(key, None) env["NO_PROXY"] = "127.0.0.1,localhost" env["no_proxy"] = "127.0.0.1,localhost" return env def is_fake_ip(remote_ip: str) -> bool: try: ip = ipaddress.ip_address(remote_ip) except ValueError: return False return ip in ipaddress.ip_network("198.18.0.0/15") def is_ip_address(value: str) -> bool: try: ipaddress.ip_address(value) return True except ValueError: return False def resolve_server_ip(server: str) -> str: if not server or is_ip_address(server): return server dig = shutil.which("dig") if not dig: return server for resolver in ("1.1.1.1", "8.8.8.8", "223.5.5.5"): try: completed = subprocess.run( [dig, f"@{resolver}", "+short", server, "A", "+time=3", "+tries=1"], text=True, capture_output=True, timeout=5, env=no_proxy_env(), ) except Exception: continue for line in completed.stdout.splitlines(): candidate = line.strip() if candidate and is_ip_address(candidate) and not is_fake_ip(candidate): return candidate return server def preflight_proxy_check(url: str, timeout: int) -> dict[str, Any]: curl = shutil.which("curl") if not curl: return {"ok": False, "warning": "curl not found; skip proxy preflight"} try: completed = subprocess.run( [ curl, "--noproxy", "*", "--silent", "--show-error", "--output", "/dev/null", "--max-time", str(timeout), "--write-out", "%{http_code} %{remote_ip}", url, ], text=True, capture_output=True, env=no_proxy_env(), timeout=timeout + 3, ) except Exception as exc: return {"ok": False, "warning": f"direct preflight failed: {exc}"} parts = completed.stdout.strip().split() http_code = parts[0] if parts else "000" remote_ip = parts[1] if len(parts) > 1 else "" warning = None if completed.returncode == 0 and http_code.isdigit() and int(http_code) > 0: warning = ( "direct request succeeded even with --noproxy '*'; if this network should not reach " "the test URL directly, a global proxy/TUN may be intercepting traffic" ) if remote_ip and is_fake_ip(remote_ip): warning = ( f"direct request resolved to Fake-IP {remote_ip}; TUN/enhanced mode is intercepting traffic" ) return { "ok": completed.returncode == 0, "http_code": http_code, "remote_ip": remote_ip, "warning": warning, "stderr": completed.stderr.strip(), } def load_config(path: Path) -> dict[str, Any]: if not path.exists(): raise SystemExit(f"Config file not found: {path}") yaml = load_yaml_module() data = yaml.safe_load(path.read_text(encoding="utf-8")) or {} if not isinstance(data, dict): raise SystemExit("Config root must be a YAML object.") return data def find_core(configured_path: str | None) -> str: candidates = [configured_path] if configured_path else [] candidates.extend(["mihomo", "clash-meta", "clash"]) for item in candidates: if not item: continue found = shutil.which(item) if found: return found path = Path(item).expanduser() if path.exists() and os.access(path, os.X_OK): return str(path) raise SystemExit( "Cannot find mihomo/clash core.\n" "Install mihomo, or set mihomo_path in config.yaml.\n" "macOS example:\n" " brew install mihomo" ) def fetch_url(url: str, timeout: int, use_system_proxy: bool) -> str: handlers: list[Any] = [] if not use_system_proxy: handlers.append(request.ProxyHandler({})) opener = request.build_opener(*handlers) req = request.Request( url, headers={ "User-Agent": "node-monitor/0.1", "Cache-Control": "no-cache", "Pragma": "no-cache", }, ) with opener.open(req, timeout=timeout) as resp: raw = resp.read() return raw.decode("utf-8", errors="replace").strip() def b64_decode_text(value: str) -> str | None: compact = "".join(value.strip().split()) if not compact: return None for candidate in (compact, compact.replace("-", "+").replace("_", "/")): padded = candidate + "=" * (-len(candidate) % 4) try: decoded = base64.b64decode(padded, validate=False) text = decoded.decode("utf-8", errors="replace") if "://" in text or text.lstrip().startswith(("proxies:", "{", "[")): return text except Exception: continue return None def percent_decode(value: str | None) -> str: return parse.unquote(value or "") def split_host_port(netloc: str, default_port: int | None = None) -> tuple[str, int]: if "@" in netloc: netloc = netloc.rsplit("@", 1)[1] if netloc.startswith("["): host, _, rest = netloc[1:].partition("]") port_text = rest[1:] if rest.startswith(":") else "" else: host, sep, port_text = netloc.rpartition(":") if not sep: host, port_text = netloc, "" if not host: raise ValueError("missing server host") if port_text: return percent_decode(host), int(port_text) if default_port is None: raise ValueError("missing server port") return percent_decode(host), default_port def parse_bool(value: Any) -> bool: if isinstance(value, bool): return value return str(value).lower() in {"1", "true", "yes", "y"} def first_query(qs: dict[str, list[str]], *names: str) -> str | None: for name in names: values = qs.get(name) if values: return values[0] return None def add_transport(node: dict[str, Any], qs: dict[str, list[str]]) -> None: network = first_query(qs, "type", "network", "net") if not network: return node["network"] = network if network == "ws": host = first_query(qs, "host") path = first_query(qs, "path") ws_opts: dict[str, Any] = {} if path: ws_opts["path"] = path if host: ws_opts["headers"] = {"Host": host} if ws_opts: node["ws-opts"] = ws_opts elif network == "grpc": service_name = first_query(qs, "serviceName", "service-name", "grpc-service-name") if service_name: node["grpc-opts"] = {"grpc-service-name": service_name} def add_tls_fields(node: dict[str, Any], qs: dict[str, list[str]]) -> None: security = first_query(qs, "security") sni = first_query(qs, "sni", "servername", "peer") fp = first_query(qs, "fp", "client-fingerprint") alpn = first_query(qs, "alpn") insecure = first_query(qs, "allowInsecure", "insecure", "skip-cert-verify") if security in {"tls", "reality"}: node["tls"] = True if sni: node["servername"] = sni node["sni"] = sni if fp: node["client-fingerprint"] = fp if alpn: node["alpn"] = [item for item in alpn.split(",") if item] if insecure is not None: node["skip-cert-verify"] = parse_bool(insecure) if security == "reality": reality: dict[str, Any] = {} pbk = first_query(qs, "pbk", "public-key") sid = first_query(qs, "sid", "short-id") spider_x = first_query(qs, "spx", "spider-x") if pbk: reality["public-key"] = pbk if sid: reality["short-id"] = sid if spider_x: reality["spider-x"] = spider_x if reality: node["reality-opts"] = reality def parse_vmess_link(link: str) -> dict[str, Any]: payload = link[len("vmess://") :] decoded = b64_decode_text(payload) if not decoded: raise ValueError("cannot decode vmess payload") data = json.loads(decoded) node: dict[str, Any] = { "name": str(data.get("ps") or data.get("name") or data.get("add") or "vmess"), "type": "vmess", "server": str(data["add"]), "port": int(data.get("port") or 443), "uuid": str(data["id"]), "alterId": int(data.get("aid") or 0), "cipher": str(data.get("scy") or data.get("cipher") or "auto"), } if data.get("tls") == "tls": node["tls"] = True if data.get("sni"): node["servername"] = data["sni"] node["sni"] = data["sni"] if data.get("fp"): node["client-fingerprint"] = data["fp"] net = data.get("net") if net and net != "tcp": node["network"] = net if net == "ws": ws_opts: dict[str, Any] = {} if data.get("path"): ws_opts["path"] = data["path"] if data.get("host"): ws_opts["headers"] = {"Host": data["host"]} if ws_opts: node["ws-opts"] = ws_opts elif net == "grpc" and data.get("path"): node["grpc-opts"] = {"grpc-service-name": data["path"]} return node def parse_url_link(link: str) -> dict[str, Any]: parsed = parse.urlsplit(link) scheme = parsed.scheme.lower() qs = parse.parse_qs(parsed.query, keep_blank_values=True) name = percent_decode(parsed.fragment) or scheme username = percent_decode(parsed.username or "") password = percent_decode(parsed.password or "") host, port = split_host_port(parsed.netloc, 443) if scheme == "vless": node: dict[str, Any] = { "name": name, "type": "vless", "server": host, "port": port, "uuid": username, "encryption": first_query(qs, "encryption") or "none", } flow = first_query(qs, "flow") if flow: node["flow"] = flow add_tls_fields(node, qs) add_transport(node, qs) return node if scheme == "trojan": node = { "name": name, "type": "trojan", "server": host, "port": port, "password": username or password, } add_tls_fields(node, qs) add_transport(node, qs) return node if scheme in {"hysteria2", "hy2"}: node = { "name": name, "type": "hysteria2", "server": host, "port": port, "password": username or password, } add_tls_fields(node, qs) obfs = first_query(qs, "obfs") obfs_password = first_query(qs, "obfs-password", "obfs_password") if obfs: if ":" in obfs and not obfs_password: obfs, obfs_password = obfs.split(":", 1) node["obfs"] = obfs if obfs_password: node["obfs-password"] = obfs_password return node if scheme == "anytls": security = first_query(qs, "security") node = { "name": name, "type": "anytls", "server": host, "port": port, "password": username or password, } if security == "reality": node["_unsupported_reason"] = "mihomo does not support AnyTLS + Reality" add_tls_fields(node, qs) return node raise ValueError(f"unsupported link scheme: {scheme}") def parse_link(line: str) -> dict[str, Any] | None: stripped = line.strip() if not stripped or stripped.startswith("#") or "://" not in stripped: return None scheme = stripped.split("://", 1)[0].lower() if scheme == "vmess": return parse_vmess_link(stripped) if scheme in {"vless", "trojan", "hysteria2", "hy2", "anytls"}: return parse_url_link(stripped) return None def parse_subscription_text(text: str, source: str) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: yaml = load_yaml_module() nodes: list[dict[str, Any]] = [] errors: list[dict[str, Any]] = [] candidates = [text] decoded = b64_decode_text(text) if decoded and decoded != text: candidates.insert(0, decoded) for candidate in candidates: try: data = yaml.safe_load(candidate) if isinstance(data, dict) and isinstance(data.get("proxies"), list): for proxy in data["proxies"]: if not isinstance(proxy, dict): continue proxy_type = str(proxy.get("type", "")).lower() if proxy_type in SUPPORTED_TYPES: item = dict(proxy) item["type"] = proxy_type item.setdefault("name", f"{proxy_type}-{len(nodes) + 1}") item["_source"] = source nodes.append(item) if nodes: return nodes, errors except Exception: pass for line in candidates[0].splitlines(): try: node = parse_link(line) if node: node["_source"] = source nodes.append(node) except Exception as exc: errors.append({"source": source, "line": line[:160], "error": str(exc)}) return nodes, errors def matches_keywords(node: dict[str, Any], keywords: list[str]) -> bool: name = str(node.get("name", "")) return any(keyword and keyword in name for keyword in keywords) def classify_node(node: dict[str, Any], domestic_keywords: list[str]) -> str: if matches_keywords(node, domestic_keywords): return "domestic" return "overseas" def fingerprint_node(node: dict[str, Any]) -> str: comparable = {k: v for k, v in node.items() if not k.startswith("_") and k != "name"} raw = json.dumps(comparable, ensure_ascii=False, sort_keys=True, default=str) return hashlib.sha256(raw.encode("utf-8")).hexdigest() def node_id(node: dict[str, Any]) -> str: return fingerprint_node(node)[:10] def unique_nodes(nodes: list[dict[str, Any]]) -> list[dict[str, Any]]: seen: set[str] = set() name_counts: dict[str, int] = {} result: list[dict[str, Any]] = [] for node in nodes: fp = fingerprint_node(node) if fp in seen: continue seen.add(fp) base_name = str(node.get("name") or f"{node.get('type', 'node')}-{len(result) + 1}") name_counts[base_name] = name_counts.get(base_name, 0) + 1 node["name"] = base_name if name_counts[base_name] == 1 else f"{base_name} #{name_counts[base_name]}" node["_node_id"] = node_id(node) result.append(node) return result def clean_node_for_mihomo(node: dict[str, Any]) -> dict[str, Any]: return {k: v for k, v in node.items() if not k.startswith("_")} def ensure_tls_server_name(proxy_node: dict[str, Any], original_server: str) -> None: if not original_server or is_ip_address(original_server): return if proxy_node.get("servername") or proxy_node.get("sni"): return if proxy_node.get("tls") is not True: return proxy_node["servername"] = original_server proxy_node["sni"] = original_server def free_port() -> int: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: sock.bind(("127.0.0.1", 0)) return int(sock.getsockname()[1]) def wait_for_port(port: int, process: subprocess.Popen[Any], timeout: int) -> None: deadline = time.time() + timeout while time.time() < deadline: if process.poll() is not None: raise RuntimeError("mihomo exited before opening local port") with contextlib.closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock: sock.settimeout(0.2) if sock.connect_ex(("127.0.0.1", port)) == 0: return time.sleep(0.1) raise TimeoutError("timed out waiting for mihomo local port") def make_mihomo_config(node: dict[str, Any], mixed_port: int) -> dict[str, Any]: name = str(node["name"]) proxy_node = dict(node) original_server = str(proxy_node.get("server") or "") ensure_tls_server_name(proxy_node, original_server) server_ip = str(proxy_node.get("_server_ip") or "") if server_ip and is_ip_address(server_ip) and not is_fake_ip(server_ip): proxy_node["server"] = server_ip return { "mixed-port": mixed_port, "allow-lan": False, "bind-address": "127.0.0.1", "mode": "rule", "log-level": "warning", "ipv6": False, "unified-delay": True, "proxies": [clean_node_for_mihomo(proxy_node)], "proxy-groups": [{"name": "PROXY", "type": "select", "proxies": [name]}], "rules": ["MATCH,PROXY"], } def dump_yaml(path: Path, data: Any) -> None: yaml = load_yaml_module() path.write_text( yaml.safe_dump(data, allow_unicode=True, sort_keys=False), encoding="utf-8", ) def test_node( node: dict[str, Any], core_path: str, test_url: str, expected_status: int, timeout: int, startup_timeout: int, ) -> dict[str, Any]: started_at = time.perf_counter() result = { "name": node.get("name"), "node_id": node.get("_node_id") or node_id(node), "type": node.get("type"), "server": node.get("server"), "server_ip": node.get("_server_ip") or node.get("server"), "port": node.get("port"), "alive": False, "http_code": None, "latency_ms": None, "test_profile": node.get("_test_profile"), "test_url": test_url, "expected_status": expected_status, "error": None, } if node.get("_unsupported_reason"): result["error"] = node["_unsupported_reason"] return result with tempfile.TemporaryDirectory(prefix="node-monitor-") as tmpdir: mixed_port = free_port() config_path = Path(tmpdir) / "mihomo.yaml" dump_yaml(config_path, make_mihomo_config(node, mixed_port)) proc = subprocess.Popen( [core_path, "-f", str(config_path)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=no_proxy_env(), ) try: wait_for_port(mixed_port, proc, startup_timeout) time.sleep(1.2) curl = shutil.which("curl") if not curl: raise RuntimeError("curl is required for proxy verification") completed = subprocess.run( [ curl, "--proxy", f"http://127.0.0.1:{mixed_port}", "--noproxy", "", "--silent", "--show-error", "--output", "/dev/null", "--max-time", str(timeout), "--write-out", "%{http_code} %{time_total}", test_url, ], text=True, capture_output=True, env=no_proxy_env(), timeout=timeout + 3, ) parts = completed.stdout.strip().split() if parts: with contextlib.suppress(ValueError): result["http_code"] = int(parts[0]) if len(parts) > 1: with contextlib.suppress(ValueError): result["curl_time_total_ms"] = round(float(parts[1]) * 1000) result["curl_exit_code"] = completed.returncode result["alive"] = completed.returncode == 0 and result["http_code"] == expected_status if not result["alive"]: stderr = completed.stderr.strip() if completed.returncode != 0: result["error"] = stderr or f"curl exit {completed.returncode}" else: result["error"] = f"HTTP {result['http_code']}, expected {expected_status}" except Exception as exc: stderr = "" with contextlib.suppress(Exception): if proc.stderr: stderr = proc.stderr.read()[-800:] result["error"] = f"{exc}; {stderr}".strip("; ") finally: result["latency_ms"] = round((time.perf_counter() - started_at) * 1000) proc.terminate() try: proc.wait(timeout=2) except subprocess.TimeoutExpired: proc.kill() proc.wait(timeout=2) return result def write_results(results_dir: Path, nodes: list[dict[str, Any]], results: list[dict[str, Any]], meta: dict[str, Any]) -> None: results_dir.mkdir(parents=True, exist_ok=True) alive_names = {item["name"] for item in results if item.get("alive")} alive_nodes = [clean_node_for_mihomo(node) for node in nodes if node.get("name") in alive_names] dump_yaml(results_dir / "alive.yaml", {"proxies": alive_nodes}) failed_lines = [ node_display_name(item) for item in results if not item.get("alive") ] failed_simple = [ { "name": item.get("name"), "type": item.get("type"), "profile": profile_label(str(item.get("test_profile") or "overseas")), "display_name": node_display_name(item), } for item in results if not item.get("alive") ] (results_dir / "dead.json").write_text( "\n".join(failed_lines) + ("\n" if failed_lines else ""), encoding="utf-8", ) (results_dir / "dead_detail.json").write_text( json.dumps([item for item in results if not item.get("alive")], ensure_ascii=False, indent=2), encoding="utf-8", ) (results_dir / "failed_nodes.txt").write_text("\n".join(failed_lines) + ("\n" if failed_lines else ""), encoding="utf-8") (results_dir / "failed_nodes.json").write_text( json.dumps(failed_simple, ensure_ascii=False, indent=2), encoding="utf-8", ) (results_dir / "last_run.json").write_text( json.dumps({"meta": meta, "results": results}, ensure_ascii=False, indent=2), encoding="utf-8", ) summary = [ f"运行时间: {meta['run_at']}", f"订阅数量: {meta['subscription_count']}", f"订阅节点: {meta['fetched_nodes']}", f"手动节点: {meta['manual_nodes']}", f"回国线路: {meta['domestic_nodes']}", f"出海线路: {meta['overseas_nodes']}", f"不支持节点: {meta['unsupported']}", f"已检测: {len(results)}", f"可用: {sum(1 for item in results if item.get('alive'))}", f"失败: {sum(1 for item in results if not item.get('alive'))}", ] (results_dir / "summary.txt").write_text("\n".join(summary) + "\n", encoding="utf-8") def telegram_config(config: dict[str, Any]) -> dict[str, Any]: data = config.get("telegram") or {} return data if isinstance(data, dict) else {} def telegram_enabled(config: dict[str, Any]) -> bool: return parse_bool(telegram_config(config).get("enabled", False)) def telegram_target_context(config: dict[str, Any], config_path: Path) -> dict[str, str]: tg = telegram_config(config) deploy_dir = str(tg.get("deploy_dir") or config_path.parent) config_file = str(tg.get("config_file") or config_path) return { "server_name": str(tg.get("server_name") or socket.gethostname()), "deploy_dir": deploy_dir, "config_file": config_file, } def truncate_message(text: str, limit: int = 3900) -> str: if len(text) <= limit: return text return text[: limit - 40].rstrip() + "\n\n...内容过长，已截断" def read_telegram_token(config: dict[str, Any]) -> str: tg = telegram_config(config) env_name = str(tg.get("bot_token_env") or "TG_BOT_TOKEN") env_token = os.environ.get(env_name, "") if env_token: return env_token token_file = str(tg.get("token_file") or "") if not token_file: return "" path = Path(token_file).expanduser() if not path.exists(): return "" raw = path.read_text(encoding="utf-8").strip() if not raw: return "" try: data = json.loads(raw) except Exception: return raw if isinstance(data, dict): key = str(tg.get("token_json_key") or "token") value = data.get(key) or data.get("bot_token") or data.get("TG_BOT_TOKEN") return str(value or "") return str(data or "") def read_telegram_chat_id(config: dict[str, Any]) -> str: tg = telegram_config(config) env_name = str(tg.get("chat_id_env") or "TG_CHAT_ID") return os.environ.get(env_name, "") or str(tg.get("chat_id") or "") def send_telegram_message(config: dict[str, Any], text: str) -> bool: tg = telegram_config(config) if not parse_bool(tg.get("enabled", False)): return False token = read_telegram_token(config) chat_id = read_telegram_chat_id(config) proxy_url = os.environ.get(str(tg.get("proxy_url_env") or "TG_PROXY_URL"), "") if not token or not chat_id: print("[tg] skip: telegram token or chat_id is not set") return False try: import requests # type: ignore except Exception as exc: print(f"[tg] skip: requests is not installed: {exc}") return False api_base = str(tg.get("api_base") or "https://api.telegram.org").rstrip("/") timeout = int(tg.get("timeout_seconds", 20)) proxies = {"http": proxy_url, "https": proxy_url} if proxy_url else None try: resp = requests.post( f"{api_base}/bot{token}/sendMessage", json={ "chat_id": chat_id, "text": html.escape(truncate_message(text)), "parse_mode": "HTML", }, proxies=proxies, timeout=timeout, ) resp.raise_for_status() except Exception as exc: print(f"[tg] failed: {exc}") return False print("[tg] sent") return True def telegram_notify_after_seen_count(config: dict[str, Any]) -> int: tg = telegram_config(config) with contextlib.suppress(Exception): return int(tg.get("notify_after_seen_count", 1)) return 1 def telegram_notify_each_result(config: dict[str, Any]) -> bool: tg = telegram_config(config) return parse_bool(tg.get("notify_each_result", False)) def ignored_failed_node_rules(config: dict[str, Any]) -> list[dict[str, Any] | str]: raw = config.get("ignored_failed_nodes") if raw is None: raw = config.get("ignored_nodes") if isinstance(raw, list): return raw return [] def node_matches_ignore_rule(node: dict[str, Any], rule: dict[str, Any] | str) -> bool: node_name = str(node.get("name") or "") node_id_value = str(node.get("node_id") or node.get("_node_id") or "") server_value = str(node.get("server_ip") or node.get("server") or "") port_value = str(node.get("port") or "") if isinstance(rule, str): rule_value = rule.strip() return bool(rule_value) and rule_value in {node_name, node_id_value, server_value} if not isinstance(rule, dict): return False checks: list[bool] = [] if rule.get("node_id"): checks.append(str(rule.get("node_id")) == node_id_value) if rule.get("name"): checks.append(str(rule.get("name")) == node_name) if rule.get("name_contains"): checks.append(str(rule.get("name_contains")) in node_name) if rule.get("server_ip"): checks.append(str(rule.get("server_ip")) == server_value) if rule.get("server"): checks.append(str(rule.get("server")) == str(node.get("server") or "")) if rule.get("port") is not None: checks.append(str(rule.get("port")) == port_value) if not checks: return False return all(checks) def filter_ignored_failed_nodes( config: dict[str, Any], results: list[dict[str, Any]], ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: rules = ignored_failed_node_rules(config) if not rules: return results, [] kept: list[dict[str, Any]] = [] ignored: list[dict[str, Any]] = [] for item in results: if any(node_matches_ignore_rule(item, rule) for rule in rules): ignored.append(item) else: kept.append(item) return kept, ignored def build_failure_notification( config: dict[str, Any], config_path: Path, meta: dict[str, Any], results: list[dict[str, Any]], ) -> str | None: fetch_errors = meta.get("fetch_errors") or [] failed_nodes = [item for item in results if not item.get("alive")] if not fetch_errors and not failed_nodes: return None ctx = telegram_target_context(config, config_path) lines = [ "⚠️ 节点监测告警", f"服务器: {ctx['server_name']}", f"目录: {ctx['deploy_dir']}", f"配置文件: {ctx['config_file']}", f"时间: {meta.get('run_at')}", ] if meta.get("subscription_label"): lines.append(f"订阅: {meta.get('subscription_label')}") if fetch_errors: lines.extend( [ "", "订阅地址拉取失败，请到上面的配置文件里修改 subscriptions.url。", ] ) for item in fetch_errors[:5]: lines.append(f"- {item.get('url')}: {item.get('error')}") if len(fetch_errors) > 5: lines.append(f"- 还有 {len(fetch_errors) - 5} 条订阅拉取错误未显示") if failed_nodes: lines.extend( [ "", f"复检后仍握手失败: {len(failed_nodes)} 个", ] ) for item in failed_nodes[:80]: lines.append(f"- {node_display_name(item)}") if len(failed_nodes) > 80: lines.append(f"- 还有 {len(failed_nodes) - 80} 个失败节点未显示") return "\n".join(lines) def build_result_notification( config: dict[str, Any], config_path: Path, results_dir: Path, meta: dict[str, Any], results: list[dict[str, Any]], ) -> str | None: alive_count = sum(1 for item in results if item.get("alive")) failed_nodes = [item for item in results if not item.get("alive")] if alive_count <= 0: return None ctx = telegram_target_context(config, config_path) total = len(results) failed_count = len(failed_nodes) title = "✅ 节点检测完成" if failed_count == 0 else "⚠️ 节点检测完成（有失败）" lines = [ title, f"服务器: {ctx['server_name']}", f"目录: {ctx['deploy_dir']}", f"配置文件: {ctx['config_file']}", f"结果目录: {results_dir}", f"时间: {meta.get('run_at')}", ] if meta.get("subscription_label"): lines.append(f"订阅: {meta.get('subscription_label')}") lines.extend( [ f"已检测: {total}", f"可用: {alive_count}", f"失败: {failed_count}", f"回国线路: {meta.get('domestic_nodes')}", f"出海线路: {meta.get('overseas_nodes')}", ] ) if failed_nodes: lines.append("") lines.append("失败节点:") for item in failed_nodes[:30]: lines.append(f"- {node_display_name(item)}") if len(failed_nodes) > 30: lines.append(f"- 还有 {len(failed_nodes) - 30} 个失败节点未显示") return "\n".join(lines) def failure_signature(meta: dict[str, Any], results: list[dict[str, Any]]) -> str: fetch_errors = [ { "url": item.get("url"), "error": item.get("error"), } for item in (meta.get("fetch_errors") or []) ] failed_nodes = [ { "name": item.get("name"), "type": item.get("type"), "server": item.get("server"), "server_ip": item.get("server_ip"), "port": item.get("port"), "error": item.get("error"), } for item in results if not item.get("alive") ] payload = json.dumps( {"fetch_errors": fetch_errors, "failed_nodes": failed_nodes}, ensure_ascii=False, sort_keys=True, ) return hashlib.sha256(payload.encode("utf-8")).hexdigest() def should_send_failure_notification( results_dir: Path, meta: dict[str, Any], results: list[dict[str, Any]], notify_after_seen_count: int = 1, ) -> bool: state_path = results_dir / "alert_state.json" fetch_errors = meta.get("fetch_errors") or [] failed_nodes = [item for item in results if not item.get("alive")] if not fetch_errors and not failed_nodes: with contextlib.suppress(FileNotFoundError): state_path.unlink() return False signature = failure_signature(meta, results) previous: dict[str, Any] = {} if state_path.exists(): with contextlib.suppress(Exception): previous = json.loads(state_path.read_text(encoding="utf-8")) current = { "signature": signature, "first_seen_at": previous.get("first_seen_at") if previous.get("signature") == signature else meta.get("run_at"), "last_seen_at": meta.get("run_at"), "seen_count": int(previous.get("seen_count") or 0) + 1 if previous.get("signature") == signature else 1, "notified": bool(previous.get("notified")) if previous.get("signature") == signature else False, "failed_count": len(failed_nodes), "fetch_error_count": len(fetch_errors), } if current["seen_count"] < max(1, notify_after_seen_count): state_path.write_text(json.dumps(current, ensure_ascii=False, indent=2), encoding="utf-8") print( f"[tg] pending: failure seen {current['seen_count']}/" f"{max(1, notify_after_seen_count)}, wait for next run" ) return False if current["notified"]: state_path.write_text(json.dumps(current, ensure_ascii=False, indent=2), encoding="utf-8") print("[tg] skip: same failure already notified") return False current["notified"] = True state_path.write_text(json.dumps(current, ensure_ascii=False, indent=2), encoding="utf-8") return True def build_nodes(config: dict[str, Any]) -> tuple[list[dict[str, Any]], dict[str, Any], list[dict[str, Any]]]: timeout = int(config.get("subscription_timeout_seconds", 20)) subscriptions = config.get("subscriptions") or [] manual_links = config.get("manual_nodes") or [] all_nodes: list[dict[str, Any]] = [] parse_errors: list[dict[str, Any]] = [] fetch_errors: list[dict[str, Any]] = [] fetched_count = 0 for index, sub in enumerate(subscriptions, start=1): if isinstance(sub, str): sub = {"url": sub} if not isinstance(sub, dict) or not sub.get("url") or sub.get("enabled") is False: continue url = str(sub["url"]) try: text = fetch_url(url, timeout=timeout, use_system_proxy=bool(sub.get("use_system_proxy", True))) nodes, errors = parse_subscription_text(text, url) fetched_count += len(nodes) all_nodes.extend(nodes) parse_errors.extend(errors) print(f"[fetch] #{index} ok: {len(nodes)} nodes from {url}") except Exception as exc: fetch_errors.append({"url": url, "error": str(exc)}) print(f"[fetch] #{index} failed: {url}: {exc}") manual_count = 0 for line in manual_links: try: node = parse_link(str(line)) if node: node["_source"] = "manual" all_nodes.append(node) manual_count += 1 except Exception as exc: parse_errors.append({"source": "manual", "line": str(line)[:160], "error": str(exc)}) domestic_keywords = [str(item) for item in config.get("domestic_keywords", ["回国"])] unique = unique_nodes(all_nodes) domestic_count = 0 overseas_count = 0 server_ip_cache: dict[str, str] = {} for node in unique: profile = classify_node(node, domestic_keywords) node["_test_profile"] = profile server = str(node.get("server") or "") if server not in server_ip_cache: server_ip_cache[server] = resolve_server_ip(server) node["_server_ip"] = server_ip_cache[server] if profile == "domestic": domestic_count += 1 else: overseas_count += 1 unsupported = sum(1 for node in unique if node.get("_unsupported_reason")) meta = { "run_at": datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds"), "subscription_count": len(subscriptions), "fetched_nodes": fetched_count, "manual_nodes": manual_count, "domestic_nodes": domestic_count, "overseas_nodes": overseas_count, "deduped": len(all_nodes) - len(unique), "unsupported": unsupported, "fetch_errors": fetch_errors, "parse_errors": parse_errors, } return unique, meta, fetch_errors + parse_errors def profile_label(profile: str) -> str: return "回国线路" if profile == "domestic" else "出海线路" def node_display_name(item: dict[str, Any]) -> str: profile = str(item.get("test_profile") or item.get("_test_profile") or "overseas") server = item.get("server_ip") or item.get("_server_ip") or item.get("server") or "-" return f"{item.get('name')} ({item.get('type')}, {server}, {profile_label(profile)})" def test_node_with_retries( node: dict[str, Any], core_path: str, test_url: str, expected_status: int, timeout: int, startup_timeout: int, retry_attempts: int, ) -> dict[str, Any]: attempts: list[dict[str, Any]] = [] last_result: dict[str, Any] | None = None total_attempts = max(1, retry_attempts) for attempt in range(1, total_attempts + 1): try: result = test_node(node, core_path, test_url, expected_status, timeout, startup_timeout) except Exception as exc: result = { "name": node.get("name"), "node_id": node.get("_node_id") or node_id(node), "type": node.get("type"), "server": node.get("server"), "server_ip": node.get("_server_ip") or node.get("server"), "port": node.get("port"), "alive": False, "http_code": None, "latency_ms": None, "test_profile": node.get("_test_profile"), "test_url": test_url, "expected_status": expected_status, "error": str(exc), "traceback": traceback.format_exc(), } result["attempt"] = attempt last_result = result attempts.append( { "attempt": attempt, "alive": result.get("alive"), "http_code": result.get("http_code"), "latency_ms": result.get("latency_ms"), "curl_exit_code": result.get("curl_exit_code"), "error": result.get("error"), } ) if result.get("alive"): break if attempt < total_attempts: time.sleep(0.5) assert last_result is not None last_result["attempts"] = len(attempts) last_result["attempt_results"] = attempts return last_result def run_loaded_nodes( config: dict[str, Any], config_path: Path, results_dir: Path, nodes: list[dict[str, Any]], meta: dict[str, Any], errors: list[dict[str, Any]], args: argparse.Namespace, run_label: str | None = None, ) -> tuple[list[dict[str, Any]], list[dict[str, Any]], dict[str, Any]]: if args.limit > 0: nodes = nodes[: args.limit] if run_label: print(f"[订阅] {run_label}") print( f"[节点] 待检测={len(nodes)} 回国线路={meta['domestic_nodes']} 出海线路={meta['overseas_nodes']} " f"去重={meta['deduped']} 不支持={meta['unsupported']}" ) if args.dry_run: results_dir.mkdir(parents=True, exist_ok=True) dump_yaml(results_dir / "parsed_nodes.yaml", {"proxies": [clean_node_for_mihomo(node) for node in nodes]}) (results_dir / "parse_errors.json").write_text(json.dumps(errors, ensure_ascii=False, indent=2), encoding="utf-8") print(f"[dry-run] wrote {results_dir / 'parsed_nodes.yaml'}") return nodes, [], meta if not nodes: results: list[dict[str, Any]] = [] write_results(results_dir, nodes, results, meta) notification = build_failure_notification(config, config_path, meta, results) if notification and should_send_failure_notification( results_dir, meta, results, telegram_notify_after_seen_count(config), ): send_telegram_message(config, notification) print("[完成] 没有可检测节点") print(f"[完成] 结果目录: {results_dir}") return nodes, results, meta core_path = find_core(config.get("mihomo_path")) overseas_test_url = str(config.get("overseas_test_url", config.get("test_url", "https://www.google.com/robots.txt"))) overseas_expected_status = int(config.get("overseas_expected_status", config.get("expected_status", 200))) domestic_test_url = str(config.get("domestic_test_url", "https://www.baidu.com/")) domestic_expected_status = int(config.get("domestic_expected_status", 200)) timeout = int(config.get("timeout_seconds", 10)) startup_timeout = int(config.get("startup_timeout_seconds", 8)) retry_attempts = int(config.get("retry_attempts", 5)) recheck_failed_nodes = parse_bool(config.get("recheck_failed_nodes", True)) recheck_retry_attempts = int(config.get("recheck_retry_attempts", 10)) results: list[dict[str, Any]] = [] node_entries = list(enumerate(nodes, start=1)) for index, node in node_entries: profile = str(node.get("_test_profile") or "overseas") if profile == "domestic": test_url = domestic_test_url expected_status = domestic_expected_status else: test_url = overseas_test_url expected_status = overseas_expected_status display = f"{index}/{len(nodes)} {node_display_name(node)}" sys.stdout.write(f"\r\033[K[检测中] {display}") sys.stdout.flush() result = test_node_with_retries( node, core_path, test_url, expected_status, timeout, startup_timeout, retry_attempts, ) status = "✅" if result.get("alive") else "❌" sys.stdout.write(f"\r\033[K{status} {display}--{result.get('latency_ms')}ms--尝试{result.get('attempts')}次\n") sys.stdout.flush() result["check_round"] = "初检" results.append(result) if recheck_failed_nodes: failed_entries = [ (index, node, results[index - 1]) for index, node in node_entries if not results[index - 1].get("alive") ] if failed_entries: print(f"[复检] 初检失败={len(failed_entries)} 每节点最多尝试{recheck_retry_attempts}次") for recheck_index, (original_index, node, first_result) in enumerate(failed_entries, start=1): profile = str(node.get("_test_profile") or "overseas") if profile == "domestic": test_url = domestic_test_url expected_status = domestic_expected_status else: test_url = overseas_test_url expected_status = overseas_expected_status display = f"{original_index}/{len(nodes)} {node_display_name(node)}" sys.stdout.write(f"\r\033[K[复检中] {recheck_index}/{len(failed_entries)} {display}") sys.stdout.flush() result = test_node_with_retries( node, core_path, test_url, expected_status, timeout, startup_timeout, recheck_retry_attempts, ) result["check_round"] = "复检" result["first_check"] = { "alive": first_result.get("alive"), "http_code": first_result.get("http_code"), "latency_ms": first_result.get("latency_ms"), "curl_exit_code": first_result.get("curl_exit_code"), "error": first_result.get("error"), "attempts": first_result.get("attempts"), } status = "✅" if result.get("alive") else "❌" sys.stdout.write( f"\r\033[K{status} [复检] {display}--{result.get('latency_ms')}ms--尝试{result.get('attempts')}次\n" ) sys.stdout.flush() results[original_index - 1] = result results, ignored_results = filter_ignored_failed_nodes(config, results) if ignored_results: print(f"[排除] 已忽略 {len(ignored_results)} 个配置中的误报节点") for item in ignored_results[:10]: print(node_display_name(item)) if len(ignored_results) > 10: print(f"[排除] 还有 {len(ignored_results) - 10} 个忽略节点未显示") write_results(results_dir, nodes, results, meta) result_notification = build_result_notification(config, config_path, results_dir, meta, results) result_notification_sent = False if telegram_notify_each_result(config) and result_notification: result_notification_sent = send_telegram_message(config, result_notification) if not result_notification_sent: notification = build_failure_notification(config, config_path, meta, results) if notification and should_send_failure_notification( results_dir, meta, results, telegram_notify_after_seen_count(config), ): send_telegram_message(config, notification) print(f"[完成] ✅={sum(1 for item in results if item.get('alive'))} ❌={sum(1 for item in results if not item.get('alive'))}") failed = [item for item in results if not item.get("alive")] if failed: print("[握手失败]") for item in failed: print(node_display_name(item)) print(f"[完成] 结果目录: {results_dir}") return nodes, results, meta def redact_subscription_url(url: str) -> str: if "token=" in url: return url.split("token=", 1)[0] + "token=REDACTED" return url def enabled_subscriptions(config: dict[str, Any]) -> list[dict[str, Any]]: result: list[dict[str, Any]] = [] for sub in config.get("subscriptions") or []: if isinstance(sub, str): sub = {"url": sub} if isinstance(sub, dict) and sub.get("url") and sub.get("enabled") is not False: result.append(dict(sub)) return result def write_separate_summary(results_dir: Path, summaries: list[dict[str, Any]]) -> None: results_dir.mkdir(parents=True, exist_ok=True) total_nodes = sum(int(item.get("nodes") or 0) for item in summaries) total_alive = sum(int(item.get("alive") or 0) for item in summaries) total_failed = sum(int(item.get("failed") or 0) for item in summaries) lines = [ f"运行时间: {datetime.now(timezone.utc).astimezone().isoformat(timespec='seconds')}", f"订阅分组: {len(summaries)}", f"已检测: {total_nodes}", f"可用: {total_alive}", f"失败: {total_failed}", ] for item in summaries: lines.append( f"订阅{item['index']}: 节点={item['nodes']} 可用={item['alive']} 失败={item['failed']} " f"目录={item['results_dir']}" ) (results_dir / "summary.txt").write_text("\n".join(lines) + "\n", encoding="utf-8") def main() -> int: parser = argparse.ArgumentParser(description="Monitor proxy node liveness with real HTTP 200 checks.") parser.add_argument("-c", "--config", default="config.yaml", help="Path to YAML config.") parser.add_argument("--limit", type=int, default=0, help="Test only the first N nodes.") parser.add_argument("--dry-run", action="store_true", help="Fetch and parse subscriptions without testing nodes.") parser.add_argument("--skip-preflight", action="store_true", help="Skip global proxy/TUN preflight warning.") args = parser.parse_args() config_path = Path(args.config).expanduser().resolve() config = load_config(config_path) results_dir = (config_path.parent / str(config.get("results_dir", "results"))).resolve() preflight: dict[str, Any] | None = None if not args.skip_preflight: preflight_url = str(config.get("preflight_url", config.get("overseas_test_url", "https://www.google.com/robots.txt"))) preflight = preflight_proxy_check(preflight_url, int(config.get("preflight_timeout_seconds", 8))) if preflight.get("warning"): print(f"[preflight] warning: {preflight['warning']}") else: print( f"[preflight] direct check: code={preflight.get('http_code')} " f"remote_ip={preflight.get('remote_ip') or '-'}" ) if parse_bool(config.get("process_subscriptions_separately", False)): subscriptions = enabled_subscriptions(config) summaries: list[dict[str, Any]] = [] for sub_index, sub in enumerate(subscriptions, start=1): single_config = dict(config) single_config["subscriptions"] = [sub] single_config["manual_nodes"] = [] label = f"{sub_index}/{len(subscriptions)} {redact_subscription_url(str(sub.get('url') or ''))}" sub_results_dir = results_dir / f"subscription-{sub_index}" nodes, meta, errors = build_nodes(single_config) meta["subscription_index"] = sub_index meta["subscription_label"] = label if preflight is not None and sub_index == 1: meta["preflight"] = preflight nodes, run_results, meta = run_loaded_nodes( single_config, config_path, sub_results_dir, nodes, meta, errors, args, label, ) summaries.append( { "index": sub_index, "nodes": len(run_results) if run_results else len(nodes), "alive": sum(1 for item in run_results if item.get("alive")), "failed": sum(1 for item in run_results if not item.get("alive")), "results_dir": str(sub_results_dir), } ) print( f"[订阅完成] {sub_index}/{len(subscriptions)} " f"✅={summaries[-1]['alive']} ❌={summaries[-1]['failed']}" ) write_separate_summary(results_dir, summaries) print(f"[完成] 分订阅结果目录: {results_dir}") return 0 nodes, meta, errors = build_nodes(config) if preflight is not None: meta["preflight"] = preflight run_loaded_nodes(config, config_path, results_dir, nodes, meta, errors, args) return 0 if __name__ == "__main__": raise SystemExit(main())