自动化Clash代理管理 | 原创，AI翻译
这篇帖子详细介绍了一个Python脚本clash.py，旨在自动化管理您的Clash代理配置。它能处理从定期下载更新的代理配置和重启Clash服务，到智能选择并切换至指定组内最快的可用代理等一系列操作。配合clash.py使用的speed.py模块，可并发测试各Clash代理的延迟，确保您的连接始终通过最优服务器路由。
clash.py
import os
import subprocess
import time
import shutil
import argparse
import logging
import requests
import json
import urllib.parse
# 假设speed.py位于同一目录或PYTHONPATH可访问路径
from speed import get_top_proxies
# --- 配置项 ---
CLASH_CONTROLLER_HOST = "127.0.0.1"
CLASH_CONTROLLER_PORT = 9090
CLASH_API_BASE_URL = f"http://{CLASH_CONTROLLER_HOST}:{CLASH_CONTROLLER_PORT}"
# 目标代理组名称，最佳代理将被分配至此
# 请确保该组存在于您的Clash配置中
TARGET_PROXY_GROUP = "🚧代理"
def setup_logging():
"""配置脚本基础日志"""
logging.basicConfig(
filename='clash.log',
level=logging.INFO,
format='%(asctime)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
def start_system_proxy(global_proxy_address):
"""设置系统全局代理环境变量"""
os.environ["GLOBAL_PROXY"] = global_proxy_address # 设置为全局变量以备他用
os.environ["HTTP_PROXY"] = f"http://{global_proxy_address}"
os.environ["HTTPS_PROXY"] = f"http://{global_proxy_address}"
os.environ["http_proxy"] = f"http://{global_proxy_address}"
os.environ["https_proxy"] = f"http://{global_proxy_address}"
# 现代工具通常无需显式设置为"false"，此处保留以兼容原脚本逻辑
os.environ["HTTP_PROXY_REQUEST_FULLURI"] = "false"
os.environ["HTTPS_PROXY_REQUEST_FULLURI"] = "false"
os.environ["ALL_PROXY"] = os.environ["http_proxy"]
logging.info(f"系统全局代理已设置为: {global_proxy_address}")
def stop_system_proxy():
"""清除系统全局代理环境变量"""
os.environ["http_proxy"] = ""
os.environ["HTTP_PROXY"] = ""
os.environ["https_proxy"] = ""
os.environ["HTTPS_PROXY"] = ""
os.environ["HTTP_PROXY_REQUEST_FULLURI"] = "true" # 恢复默认值
os.environ["HTTPS_PROXY_REQUEST_FULLURI"] = "true"
os.environ["ALL_PROXY"] = ""
logging.info("系统全局代理已停用（环境变量已清除）")
def switch_clash_proxy_group(group_name, proxy_name):
"""
将指定Clash代理组中的活动代理切换为新代理
"""
encoded_group_name = urllib.parse.quote(group_name)
url = f"{CLASH_API_BASE_URL}/proxies/{encoded_group_name}"
headers = {"Content-Type": "application/json"}
payload = {"name": proxy_name}
try:
response = requests.put(url, headers=headers, data=json.dumps(payload), timeout=5)
response.raise_for_status()
logging.info(f"成功将'{group_name}'切换至'{proxy_name}'")
return True
except requests.exceptions.ConnectionError:
logging.error(f"错误：无法连接至Clash API接口 {CLASH_API_BASE_URL} 以切换代理")
logging.error("请确保Clash正在运行且external-controller配置正确")
return False
except requests.exceptions.Timeout:
logging.error(f"错误：切换'{group_name}'代理时连接Clash API超时")
return False
except requests.exceptions.RequestException as e:
logging.error(f"切换'{group_name}'代理时发生意外错误: {e}")
return False
def main():
"""主函数：管理Clash配置、重启及选择最佳代理"""
setup_logging()
parser = argparse.ArgumentParser(description="Clash配置管理脚本")
parser.add_argument("--minutes", type=int, default=10, help="更新间隔分钟数（默认：10）")
parser.add_argument("--iterations", type=int, default=1000, help="循环次数（默认：1000）")
parser.add_argument(
"--config-url",
type=str,
default=os.getenv("CLASH_DOWNLOAD_URL"),
help="Clash配置下载URL。默认使用CLASH_DOWNLOAD_URL环境变量，若无则需手动指定"
)
args = parser.parse_args()
ITERATIONS = args.iterations
SLEEP_SECONDS = args.minutes * 60
config_download_url = args.config_url
if not config_download_url:
logging.critical("错误：未提供配置下载URL。请设置CLASH_DOWNLOAD_URL环境变量或使用--config-url参数")
return # 无可用URL时退出
clash_executable_path = "/home/lzw/clash-linux-386-v1.17.0/clash-linux-386"
clash_config_dir = os.path.expanduser("~/.config/clash")
clash_config_path = os.path.join(clash_config_dir, "config.yaml")
for i in range(1, ITERATIONS + 1):
logging.info(f"--- 开始第 {i} 次循环（共 {ITERATIONS} 次） ---")
# 步骤1：停止现有系统代理设置
stop_system_proxy()
# 步骤2：下载并更新Clash配置
try:
logging.info(f"从以下地址下载新配置: {config_download_url}")
subprocess.run(["wget", config_download_url, "-O", "zhs4.yaml"], check=True, capture_output=True)
os.makedirs(clash_config_dir, exist_ok=True)
shutil.move("zhs4.yaml", clash_config_path)
logging.info("Clash配置更新成功！")
except subprocess.CalledProcessError as e:
logging.error(f"下载或移动配置文件失败: {e.stderr.decode().strip()}")
logging.error("跳过本次循环")
time.sleep(10) # 重试前短暂等待
continue
except Exception as e:
logging.error(f"配置更新时发生意外错误: {e}")
logging.error("跳过本次循环")
time.sleep(10)
continue
# 步骤3：后台启动Clash
clash_process = None
try:
# 确保Clash启动时启用了external-controller
# 这通常已在config.yaml中配置
clash_process = subprocess.Popen([clash_executable_path],
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
logging.info(f"Clash已启动，PID为 {clash_process.pid}")
# 等待Clash完全初始化并开放API端口
time.sleep(5)
except FileNotFoundError:
logging.critical(f"未找到Clash可执行文件: {clash_executable_path}")
logging.critical("请确认路径正确且Clash已安装")
return # 关键错误，退出脚本
except Exception as e:
logging.error(f"启动Clash失败: {e}")
logging.error("跳过本次循环")
if clash_process: clash_process.terminate()
time.sleep(10)
continue
# 步骤4：测试代理速度并选择最佳节点
best_proxy_name = None
try:
logging.info("正在测试代理速度以寻找最佳节点...")
top_proxies = get_top_proxies(num_results=1) # 仅获取最佳单个代理
if top_proxies:
best_proxy_name = top_proxies[0]['name']
logging.info(f"已识别最佳代理: '{best_proxy_name}'，延迟 {top_proxies[0]['latency']}ms")
else:
logging.warning("无成功代理测试结果。本次循环无法选择最佳代理")
except Exception as e:
logging.error(f"代理速度测试时出错: {e}")
# 步骤5：将Clash代理组切换至最佳代理（如找到）
if best_proxy_name:
# 设置系统代理前确保Clash配置正确
# 将系统代理指向Clash本地HTTP代理端口
# Clash通常运行在7890端口（请根据实际配置调整）
clash_local_proxy_address = f"{CLASH_CONTROLLER_HOST}:7890"
start_system_proxy(clash_local_proxy_address)
if not switch_clash_proxy_group(TARGET_PROXY_GROUP, best_proxy_name):
logging.error(f"无法将Clash组'{TARGET_PROXY_GROUP}'切换至'{best_proxy_name}'")
else:
logging.warning("未找到最佳代理，本次循环跳过代理组切换和系统代理设置")
# 步骤6：等待指定时长
logging.info(f"等待 {SLEEP_SECONDS / 60} 分钟后进入下一循环...")
time.sleep(SLEEP_SECONDS)
# 步骤7：停止Clash进程
if clash_process:
logging.info("正在终止Clash进程...")
clash_process.terminate()
try:
clash_process.wait(timeout=10) # 给予Clash优雅退出的时间
logging.info("Clash已成功停止")
except subprocess.TimeoutExpired:
logging.warning("Clash未正常终止，强制结束进程")
clash_process.kill()
clash_process.wait() # 确保进程完全终止
except Exception as e:
logging.error(f"等待Clash停止时出错: {e}")
logging.info(f"--- 第 {i} 次循环完成 ---")
logging.info(f"已完成 {ITERATIONS} 次循环。脚本执行结束")
if __name__ == "__main__":
main()
speed.py
import requests
import json
import urllib.parse
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
import logging # 导入日志模块
# --- 配置项 ---
CLASH_CONTROLLER_HOST = "127.0.0.1" # 控制器位于本机
CLASH_CONTROLLER_PORT = 9090
CLASH_API_BASE_URL = f"http://{CLASH_CONTROLLER_HOST}:{CLASH_CONTROLLER_PORT}"
LATENCY_TEST_URL = "https://github.com" # 更新测试URL
LATENCY_TEST_TIMEOUT_MS = 5000 # 毫秒
CONCURRENT_CONNECTIONS = 10 # 并发测试数
# 需要排除的代理组名称列表
# 这些通常是策略组或特殊代理而非独立节点
EXCLUDE_PROXY_GROUPS = [
"DIRECT",
"REJECT",
"GLOBAL", # 默认已在API中排除
"🇨🇳国内网站或资源",
"🌵其它规则外",
"🎬Netflix等国外流媒体",
"📦ChatGPT",
"📹Youtube",
"📺爱奇艺等国内流媒体",
"🚧代理",
# 可在此添加其他需要排除的组名
]
# --- speed.py日志配置 ---
# 为该脚本单独配置日志
# 确保当speed.py被导入时，其输出写入speed.log而非clash_manager.log
logging.basicConfig(
filename='clash.log',
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
def get_all_proxy_names():
"""从Clash API获取所有代理名称，排除已知组"""
try:
response = requests.get(f"{CLASH_API_BASE_URL}/proxies", timeout=5)
response.raise_for_status() # 对HTTP错误（4xx或5xx）抛出异常
proxies_data = response.json()
all_names = proxies_data.get("proxies", {}).keys()
# 过滤组名
filtered_names = [name for name in all_names if name not in EXCLUDE_PROXY_GROUPS]
logging.info(f"成功获取 {len(filtered_names)} 个可测试代理名称")
return filtered_names
except requests.exceptions.ConnectionError:
logging.error(f"无法连接至Clash API接口 {CLASH_API_BASE_URL}。请确保Clash正在运行")
return []
except requests.exceptions.Timeout:
logging.error(f"连接Clash API超时（5秒）")
return []
except requests.exceptions.RequestException as e:
logging.error(f"获取代理名称时发生意外错误: {e}")
return []
def test_proxy_latency(proxy_name):
"""通过Clash API测试单个代理延迟
返回元组 (proxy_name, latency) 或失败时返回 (proxy_name, None)
"""
encoded_proxy_name = urllib.parse.quote(proxy_name)
url = f"{CLASH_API_BASE_URL}/proxies/{encoded_proxy_name}/delay"
params = {
"url": LATENCY_TEST_URL,
"timeout": LATENCY_TEST_TIMEOUT_MS
}
try:
# requests超时单位为秒，需转换毫秒
response = requests.get(url, params=params, timeout=(LATENCY_TEST_TIMEOUT_MS / 1000) + 1)
response.raise_for_status()
latency_data = response.json()
latency = latency_data.get("delay")
logging.info(f"代理: {proxy_name} - 延迟: {latency}ms")
return proxy_name, latency
except requests.exceptions.RequestException as e:
logging.warning(f"测试'{proxy_name}'时出错: {e}")
return proxy_name, None
def get_top_proxies(num_results=5):
"""
并发测试Clash代理速度并返回前N个最快独立代理
返回:
list: 包含'name'和'latency'的字典列表
若无可用代理或出错则返回空列表
"""
logging.info("开始通过External Controller API并发测试Clash代理速度...")
logging.info(f"测试URL: {LATENCY_TEST_URL}")
logging.info(f"并发测试数: {CONCURRENT_CONNECTIONS}。请稍候...")
proxy_names_to_test = get_all_proxy_names()
if not proxy_names_to_test:
logging.warning("未找到可测试代理或获取代理名称时出错")
return []
logging.info(f"发现 {len(proxy_names_to_test)} 个独立代理待测试")
proxy_latencies = {}
with ThreadPoolExecutor(max_workers=CONCURRENT_CONNECTIONS) as executor:
future_to_proxy = {executor.submit(test_proxy_latency, name): name for name in proxy_names_to_test}
for future in as_completed(future_to_proxy):
proxy_name = future_to_proxy[future]
try:
name, latency = future.result()
if latency is not None:
proxy_latencies[name] = latency
else:
logging.info(f"代理: {name} - 测试失败或超时（详见警告日志）")
except Exception as exc:
logging.error(f"代理 {proxy_name} 测试时发生意外异常: {exc}")
sorted_proxies = sorted([item for item in proxy_latencies.items() if item[1] is not None],
key=lambda item: item[1])
top_proxies_list = []
if not sorted_proxies:
logging.warning("未完成任何成功的独立代理测试")
else:
logging.info(f"--- 前 {num_results} 个最快独立代理 ---")
for i, (name, latency) in enumerate(sorted_proxies[:num_results]):
top_proxies_list.append({"name": name, "latency": latency})
logging.info(f" {i+1}. {name}: {latency}ms")
logging.info(f"速度测试完成。已识别前 {num_results} 个最快代理")
return top_proxies_list
if __name__ == "__main__":
# 直接运行speed.py时仍会输出至speed.log
top_5_proxies = get_top_proxies(num_results=5)
print("\n函数返回的前5个代理:", top_5_proxies)