Automating Clash Proxy Management | Original
This post details a Python script, clash.py, designed to automate the management of your Clash proxy configuration. It handles everything from periodically downloading updated proxy configurations and restarting the Clash service to intelligently selecting and switching to the fastest available proxy within a designated group. Complementing clash.py, the speed.py module facilitates concurrent latency testing of individual Clash proxies, ensuring your connection always routes through the optimal server.
clash.py
import os
import subprocess
import time
import shutil
import argparse
import logging
import requests
import json
import urllib.parse
# Assuming speed.py is in the same directory or accessible in PYTHONPATH
from speed import get_top_proxies
# --- Configuration ---
CLASH_CONTROLLER_HOST = "127.0.0.1"
CLASH_CONTROLLER_PORT = 9090
CLASH_API_BASE_URL = f"http://{CLASH_CONTROLLER_HOST}:{CLASH_CONTROLLER_PORT}"
# The group proxy name to which the best individual proxy will be assigned.
# Make sure this group exists in your Clash configuration.
TARGET_PROXY_GROUP = "🚧Proxy"
def setup_logging():
"""Configures basic logging for the script."""
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):
"""Sets system-wide proxy environment variables."""
os.environ["GLOBAL_PROXY"] = global_proxy_address # Set for consistency if needed elsewhere
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}"
# These typically don't need to be explicitly set to "false" with modern tools,
# but keeping for compatibility with your original script's intent.
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"System-wide proxy set to: {global_proxy_address}")
def stop_system_proxy():
"""Clears system-wide proxy environment variables."""
os.environ["http_proxy"] = ""
os.environ["HTTP_PROXY"] = ""
os.environ["https_proxy"] = ""
os.environ["HTTPS_PROXY"] = ""
os.environ["HTTP_PROXY_REQUEST_FULLURI"] = "true" # Revert to default
os.environ["HTTPS_PROXY_REQUEST_FULLURI"] = "true"
os.environ["ALL_PROXY"] = ""
logging.info("System-wide proxy stopped (environment variables cleared).")
def switch_clash_proxy_group(group_name, proxy_name):
"""
Switches the active proxy in a specified Clash proxy group to a new proxy.
"""
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"Successfully switched '{group_name}' to '{proxy_name}'.")
return True
except requests.exceptions.ConnectionError:
logging.error(f"Error: Could not connect to Clash API at {CLASH_API_BASE_URL} to switch proxy.")
logging.error("Ensure Clash is running and its external-controller is configured.")
return False
except requests.exceptions.Timeout:
logging.error(f"Error: Connection to Clash API timed out while switching proxy for '{group_name}'.")
return False
except requests.exceptions.RequestException as e:
logging.error(f"An unexpected error occurred while switching proxy for '{group_name}': {e}")
return False
def main():
"""Main function to manage Clash config, restart, and select best proxy."""
setup_logging()
parser = argparse.ArgumentParser(description="Clash configuration and management script.")
parser.add_argument("--minutes", type=int, default=10, help="Minutes between updates (default: 10)")
parser.add_argument("--iterations", type=int, default=1000, help="Number of iterations (default: 1000)")
parser.add_argument(
"--config-url",
type=str,
default=os.getenv("CLASH_DOWNLOAD_URL"),
help="URL to download Clash configuration from. Defaults to CLASH_DOWNLOAD_URL environment variable if set, otherwise a hardcoded 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("Error: No configuration download URL provided. Please set CLASH_DOWNLOAD_URL environment variable or use --config-url argument.")
return # Exit if no URL is available
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"--- Starting Iteration {i} of {ITERATIONS} ---")
# Step 1: Stop any existing system proxy settings
stop_system_proxy()
# Step 2: Download and update Clash config
try:
logging.info(f"Downloading new config from: {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 config updated successfully!")
except subprocess.CalledProcessError as e:
logging.error(f"Failed to download or move config file: {e.stderr.decode().strip()}")
logging.error("Skipping to next iteration.")
time.sleep(10) # Wait a bit before retrying
continue
except Exception as e:
logging.error(f"An unexpected error occurred during config update: {e}")
logging.error("Skipping to next iteration.")
time.sleep(10)
continue
# Step 3: Start Clash in the background
clash_process = None
try:
# It's crucial that Clash starts with the external-controller enabled and accessible
# This is usually configured within the config.yaml itself.
clash_process = subprocess.Popen([clash_executable_path],
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
logging.info(f"Clash started with PID {clash_process.pid}")
# Give Clash a moment to fully initialize and open its API port
time.sleep(5)
except FileNotFoundError:
logging.critical(f"Clash executable not found at: {clash_executable_path}")
logging.critical("Please ensure the path is correct and Clash is installed.")
return # Critical error, exit script
except Exception as e:
logging.error(f"Failed to start Clash: {e}")
logging.error("Skipping to next iteration.")
if clash_process: clash_process.terminate()
time.sleep(10)
continue
# Step 4: Test proxy speeds and select the best one
best_proxy_name = None
try:
logging.info("Testing proxy speeds to find the best one...")
top_proxies = get_top_proxies(num_results=1) # Get only the single best proxy
if top_proxies:
best_proxy_name = top_proxies[0]['name']
logging.info(f"Identified best proxy: '{best_proxy_name}' with latency {top_proxies[0]['latency']}ms")
else:
logging.warning("No successful proxy tests. Cannot select a best proxy for this iteration.")
except Exception as e:
logging.error(f"Error during proxy speed testing: {e}")
# Step 5: Switch Clash's proxy group to the best proxy (if found)
if best_proxy_name:
# Before setting system proxy, ensure Clash is set up correctly.
# Set the system-wide proxy to point to Clash's local HTTP proxy.
# Clash typically runs its HTTP proxy on port 7890 (or similar, check your config).
clash_local_proxy_address = f"{CLASH_CONTROLLER_HOST}:7890" # Adjust if your Clash HTTP port is different
start_system_proxy(clash_local_proxy_address)
if not switch_clash_proxy_group(TARGET_PROXY_GROUP, best_proxy_name):
logging.error(f"Failed to switch Clash group '{TARGET_PROXY_GROUP}' to '{best_proxy_name}'.")
else:
logging.warning("No best proxy found, skipping proxy group switch and system proxy setup for this iteration.")
# Step 6: Wait for the specified duration
logging.info(f"Waiting for {SLEEP_SECONDS / 60} minutes before next iteration...")
time.sleep(SLEEP_SECONDS)
# Step 7: Stop Clash process
if clash_process:
logging.info("Terminating Clash process...")
clash_process.terminate()
try:
clash_process.wait(timeout=10) # Give Clash a bit more time to shut down gracefully
logging.info("Clash stopped successfully.")
except subprocess.TimeoutExpired:
logging.warning("Clash did not terminate gracefully, killing process.")
clash_process.kill()
clash_process.wait() # Ensure process is fully killed
except Exception as e:
logging.error(f"Error while waiting for Clash to stop: {e}")
logging.info(f"--- Iteration {i} completed ---")
logging.info(f"Completed {ITERATIONS} iterations. Script finished.")
if __name__ == "__main__":
main()
speed.py
import requests
import json
import urllib.parse
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
import logging # Import the logging module
# --- Configuration ---
CLASH_CONTROLLER_HOST = "127.0.0.1" # Use 127.0.0.1 as the controller is on the same machine
CLASH_CONTROLLER_PORT = 9090
CLASH_API_BASE_URL = f"http://{CLASH_CONTROLLER_HOST}:{CLASH_CONTROLLER_PORT}"
LATENCY_TEST_URL = "https://github.com" # Updated test URL
LATENCY_TEST_TIMEOUT_MS = 5000 # Milliseconds
CONCURRENT_CONNECTIONS = 10 # Number of concurrent tests
# List of known group proxy names to exclude from speed testing
# These are typically not individual nodes but rather policy groups or special proxies.
EXCLUDE_PROXY_GROUPS = [
"DIRECT",
"REJECT",
"GLOBAL", # Already excluded by default in the API
"🇨🇳国内网站或资源",
"🌵其它规则外",
"🎬Netflix等国外流媒体",
"📦ChatGPT",
"📹Youtube",
"📺爱奇艺等国内流媒体",
"🚧Proxy",
# Add any other group names you want to exclude here
]
# --- Logging Setup for speed.py ---
# Configure logging for this specific script
# This ensures that when speed.py is imported and its functions are called,
# its output goes to speed.log, separate from 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'
)
# Optionally, if you also want to see output in the console, add a StreamHandler:
# console_handler = logging.StreamHandler()
# console_handler.setLevel(logging.INFO)
# formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
# console_handler.setFormatter(formatter)
# logging.getLogger().addHandler(console_handler)
# --- Script Logic ---
def get_all_proxy_names():
"""Fetches all proxy names from the Clash API, excluding known groups."""
try:
response = requests.get(f"{CLASH_API_BASE_URL}/proxies", timeout=5)
response.raise_for_status() # Raise an exception for HTTP errors (4xx or 5xx)
proxies_data = response.json()
all_names = proxies_data.get("proxies", {}).keys()
# Filter out the group names
filtered_names = [name for name in all_names if name not in EXCLUDE_PROXY_GROUPS]
logging.info(f"Successfully fetched {len(filtered_names)} testable proxy names.")
return filtered_names
except requests.exceptions.ConnectionError:
logging.error(f"Could not connect to Clash API at {CLASH_API_BASE_URL}. Ensure Clash is running.")
return []
except requests.exceptions.Timeout:
logging.error(f"Connection to Clash API timed out after 5 seconds.")
return []
except requests.exceptions.RequestException as e:
logging.error(f"An unexpected error occurred while fetching proxy names: {e}")
return []
def test_proxy_latency(proxy_name):
"""Tests the latency of a single proxy using the Clash API.
Returns a tuple (proxy_name, latency) or (proxy_name, None) on failure.
"""
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 timeout is in seconds, convert milliseconds
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: {proxy_name} - Latency: {latency}ms")
return proxy_name, latency
except requests.exceptions.RequestException as e:
logging.warning(f"Error testing '{proxy_name}': {e}")
return proxy_name, None
def get_top_proxies(num_results=5):
"""
Tests Clash proxy speeds concurrently and returns the top N fastest individual proxies.
Returns:
list: A list of dictionaries, each containing 'name' and 'latency' for the top proxies.
Returns an empty list if no testable proxies are found or an error occurs.
"""
logging.info("Starting Clash proxy speed test via External Controller API (concurrently)...")
logging.info(f"Using test URL: {LATENCY_TEST_URL}")
logging.info(f"Running {CONCURRENT_CONNECTIONS} tests at a time. This might take a moment...")
proxy_names_to_test = get_all_proxy_names()
if not proxy_names_to_test:
logging.warning("No testable proxies found or an error occurred during proxy name retrieval.")
return []
logging.info(f"Found {len(proxy_names_to_test)} individual proxies 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"Proxy: {name} - Test Failed or Timed Out (details in warning logs)")
except Exception as exc:
logging.error(f"Proxy {proxy_name} generated an unexpected exception during testing: {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("No successful individual proxy tests were completed.")
else:
logging.info(f"--- Top {num_results} Fastest Individual Proxies ---")
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"Finished speed testing. Top {num_results} proxies identified.")
return top_proxies_list
if __name__ == "__main__":
# When speed.py is run directly, it will still output to speed.log
# and print to console (if console_handler is uncommented).
top_5_proxies = get_top_proxies(num_results=5)
print("\nTop 5 proxies returned by function:", top_5_proxies)