Automatización de la Gestión de Proxies en Clash | Original, traducido por IA
Esta publicación detalla un script de Python, clash.py, diseñado para automatizar la gestión de tu configuración de proxy Clash. Maneja todo, desde descargar periódicamente configuraciones de proxy actualizadas y reiniciar el servicio Clash hasta seleccionar y cambiar inteligentemente al proxy disponible más rápido dentro de un grupo designado. Complementando a clash.py, el módulo speed.py facilita pruebas de latencia concurrentes de proxies individuales de Clash, asegurando que tu conexión siempre se enrute a través del servidor óptimo.
clash.py
import os
import subprocess
import time
import shutil
import argparse
import logging
import requests
import json
import urllib.parse
# Asumiendo que speed.py está en el mismo directorio o accesible en PYTHONPATH
from speed import get_top_proxies
# --- Configuración ---
CLASH_CONTROLLER_HOST = "127.0.0.1"
CLASH_CONTROLLER_PORT = 9090
CLASH_API_BASE_URL = f"http://{CLASH_CONTROLLER_HOST}:{CLASH_CONTROLLER_PORT}"
# El nombre del grupo de proxy al que se asignará el mejor proxy individual.
# Asegúrate de que este grupo exista en tu configuración de Clash.
TARGET_PROXY_GROUP = "🚧Proxy"
def setup_logging():
"""Configura el registro básico para el 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):
"""Establece variables de entorno de proxy a nivel del sistema."""
os.environ["GLOBAL_PROXY"] = global_proxy_address # Establece para consistencia si se necesita en otro lugar
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}"
# Estos normalmente no necesitan ser explícitamente establecidos como "false" con herramientas modernas,
# pero se mantienen para compatibilidad con la intención original de tu script.
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"Proxy a nivel del sistema establecido en: {global_proxy_address}")
def stop_system_proxy():
"""Borra las variables de entorno de proxy a nivel del sistema."""
os.environ["http_proxy"] = ""
os.environ["HTTP_PROXY"] = ""
os.environ["https_proxy"] = ""
os.environ["HTTPS_PROXY"] = ""
os.environ["HTTP_PROXY_REQUEST_FULLURI"] = "true" # Revertir a predeterminado
os.environ["HTTPS_PROXY_REQUEST_FULLURI"] = "true"
os.environ["ALL_PROXY"] = ""
logging.info("Proxy a nivel del sistema detenido (variables de entorno borradas).")
def switch_clash_proxy_group(group_name, proxy_name):
"""
Cambia el proxy activo en un grupo de proxy de Clash especificado a un nuevo 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"Se cambió exitosamente '{group_name}' a '{proxy_name}'.")
return True
except requests.exceptions.ConnectionError:
logging.error(f"Error: No se pudo conectar a la API de Clash en {CLASH_API_BASE_URL} para cambiar el proxy.")
logging.error("Asegúrate de que Clash esté ejecutándose y su external-controller esté configurado.")
return False
except requests.exceptions.Timeout:
logging.error(f"Error: La conexión a la API de Clash se agotó mientras se cambiaba el proxy para '{group_name}'.")
return False
except requests.exceptions.RequestException as e:
logging.error(f"Ocurrió un error inesperado al cambiar el proxy para '{group_name}': {e}")
return False
def main():
"""Función principal para gestionar la configuración de Clash, reiniciar y seleccionar el mejor proxy."""
setup_logging()
parser = argparse.ArgumentParser(description="Script de configuración y gestión de Clash.")
parser.add_argument("--minutes", type=int, default=10, help="Minutos entre actualizaciones (predeterminado: 10)")
parser.add_argument("--iterations", type=int, default=1000, help="Número de iteraciones (predeterminado: 1000)")
parser.add_argument(
"--config-url",
type=str,
default=os.getenv("CLASH_DOWNLOAD_URL"),
help="URL para descargar la configuración de Clash. Por defecto, usa la variable de entorno CLASH_DOWNLOAD_URL si está configurada, de lo contrario, una URL codificada."
)
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 se proporcionó URL de descarga de configuración. Por favor, configura la variable de entorno CLASH_DOWNLOAD_URL o usa el argumento --config-url.")
return # Salir si no hay URL disponible
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"--- Iniciando Iteración {i} de {ITERATIONS} ---")
# Paso 1: Detener cualquier configuración de proxy del sistema existente
stop_system_proxy()
# Paso 2: Descargar y actualizar la configuración de Clash
try:
logging.info(f"Descargando nueva configuración desde: {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("¡Configuración de Clash actualizada exitosamente!")
except subprocess.CalledProcessError as e:
logging.error(f"Fallo al descargar o mover el archivo de configuración: {e.stderr.decode().strip()}")
logging.error("Saltando a la siguiente iteración.")
time.sleep(10) # Esperar un poco antes de reintentar
continue
except Exception as e:
logging.error(f"Ocurrió un error inesperado durante la actualización de configuración: {e}")
logging.error("Saltando a la siguiente iteración.")
time.sleep(10)
continue
# Paso 3: Iniciar Clash en segundo plano
clash_process = None
try:
# Es crucial que Clash inicie con el external-controller habilitado y accesible
# Esto normalmente se configura dentro del config.yaml mismo.
clash_process = subprocess.Popen([clash_executable_path],
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
logging.info(f"Clash iniciado con PID {clash_process.pid}")
# Darle a Clash un momento para inicializar completamente y abrir su puerto API
time.sleep(5)
except FileNotFoundError:
logging.critical(f"Ejecutable de Clash no encontrado en: {clash_executable_path}")
logging.critical("Por favor, asegúrate de que la ruta sea correcta y Clash esté instalado.")
return # Error crítico, salir del script
except Exception as e:
logging.error(f"Fallo al iniciar Clash: {e}")
logging.error("Saltando a la siguiente iteración.")
if clash_process: clash_process.terminate()
time.sleep(10)
continue
# Paso 4: Probar velocidades de proxy y seleccionar el mejor
best_proxy_name = None
try:
logging.info("Probando velocidades de proxy para encontrar el mejor...")
top_proxies = get_top_proxies(num_results=1) # Obtener solo el mejor proxy
if top_proxies:
best_proxy_name = top_proxies[0]['name']
logging.info(f"Proxy mejor identificado: '{best_proxy_name}' con latencia {top_proxies[0]['latency']}ms")
else:
logging.warning("No hubo pruebas de proxy exitosas. No se puede seleccionar un mejor proxy para esta iteración.")
except Exception as e:
logging.error(f"Error durante la prueba de velocidad de proxy: {e}")
# Paso 5: Cambiar el grupo de proxy de Clash al mejor proxy (si se encontró)
if best_proxy_name:
# Antes de configurar el proxy del sistema, asegurarse de que Clash esté configurado correctamente.
# Establecer el proxy del sistema para apuntar al proxy HTTP local de Clash.
# Clash normalmente ejecuta su proxy HTTP en el puerto 7890 (o similar, verifica tu configuración).
clash_local_proxy_address = f"{CLASH_CONTROLLER_HOST}:7890" # Ajustar si tu puerto HTTP de Clash es diferente
start_system_proxy(clash_local_proxy_address)
if not switch_clash_proxy_group(TARGET_PROXY_GROUP, best_proxy_name):
logging.error(f"Fallo al cambiar el grupo de Clash '{TARGET_PROXY_GROUP}' a '{best_proxy_name}'.")
else:
logging.warning("No se encontró mejor proxy, omitiendo el cambio de grupo de proxy y configuración de proxy del sistema para esta iteración.")
# Paso 6: Esperar la duración especificada
logging.info(f"Esperando {SLEEP_SECONDS / 60} minutos antes de la siguiente iteración...")
time.sleep(SLEEP_SECONDS)
# Paso 7: Detener el proceso de Clash
if clash_process:
logging.info("Terminando proceso de Clash...")
clash_process.terminate()
try:
clash_process.wait(timeout=10) # Darle a Clash un poco más de tiempo para apagarse correctamente
logging.info("Clash detenido exitosamente.")
except subprocess.TimeoutExpired:
logging.warning("Clash no terminó correctamente, matando proceso.")
clash_process.kill()
clash_process.wait() # Asegurarse de que el proceso esté completamente terminado
except Exception as e:
logging.error(f"Error mientras se esperaba que Clash se detuviera: {e}")
logging.info(f"--- Iteración {i} completada ---")
logging.info(f"Completadas {ITERATIONS} iteraciones. Script finalizado.")
if __name__ == "__main__":
main()
speed.py
```python import requests import json import urllib.parse import time from concurrent.futures import ThreadPoolExecutor, as_completed import logging # Importar el módulo de registro
— Configuración —
CLASH_CONTROLLER_HOST = “127.0.0.1” # Usar 127.0.0.1 ya que el controlador está en la misma máquina CLASH_CONTROLLER_PORT = 9090 CLASH_API_BASE_URL = f”http://{CLASH_CONTROLLER_HOST}:{CLASH_CONTROLLER_PORT}” LATENCY_TEST_URL = “https://github.com” # URL de prueba actualizada LATENCY_TEST_TIMEOUT_MS = 5000 # Milisegundos CONCURRENT_CONNECTIONS = 10 # Número de pruebas concurrentes
Lista de nombres de grupos de proxy conocidos para excluir de las pruebas de velocidad
Estos normalmente no son nodos individuales sino grupos de políticas o proxies especiales.
EXCLUDE_PROXY_GROUPS = [ “DIRECT”, “REJECT”, “GLOBAL”, # Ya excluido por defecto en la API “🇨🇳国内网站或资源”, “🌵其它规则外”, “🎬Netflix等国外流媒体”, “📦ChatGPT”, “📹Youtube”, “📺爱奇艺等国内流媒体”, “🚧Proxy”, # Agregar cualquier otro nombre de grupo que quieras excluir aquí ]
— Configuración de Registro para speed.py —
Configurar el registro para este script específico
Esto asegura que cuando speed.py sea importado y sus funciones sean llamadas,
su salida vaya a speed.log, separada de 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’ )
Opcionalmente, si también quieres ver la salida en la consola, agrega un 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)
— Lógica del Script —
def get_all_proxy_names(): “"”Obtiene todos los nombres de proxy de la API de Clash, excluyendo grupos conocidos.””” try: response = requests.get(f”{CLASH_API_BASE_URL}/proxies”, timeout=5) response.raise_for_status() # Lanzar una excepción para errores HTTP (4xx o 5xx) proxies_data = response.json()
all_names = proxies_data.get("proxies", {}).keys()
# Filtrar los nombres de grupo
filtered_names = [name for name in all_names if name not in EXCLUDE_PROXY_GROUPS]
logging.info(f"Se obtuvieron exitosamente {len(filtered_names)} nombres de proxy probables.")
return filtered_names
except requests.exceptions.ConnectionError:
logging.error(f"No se pudo conectar a la API de Clash en {CLASH_API_BASE_URL}. Asegúrate de que Clash esté ejecutándose.")
return []
except requests.exceptions.Timeout:
logging.error(f"La conexión a la API de Clash se agotó después de 5 segundos.")
return []
except requests.exceptions.RequestException as e:
logging.error(f"Ocurrió un error inesperado al obtener los nombres de proxy: {e}")
return []
def test_proxy_latency(proxy_name): “"”Prueba la latencia de un solo proxy usando la API de Clash. Retorna una tupla (proxy_name, latency) o (proxy_name, None) en caso de fallo. “”” 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: # el timeout de requests está en segundos, convertir milisegundos 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} - Latencia: {latency}ms”) return proxy_name, latency except requests.exceptions.RequestException as e: logging.warning(f”Error probando ‘{proxy_name}’: {e}”) return proxy_name, None
def get_top_proxies(num_results=5): “”” Prueba las velocidades de los proxies de Clash concurrentemente y retorna los N proxies individuales más rápidos.
Retorna:
list: Una lista de diccionarios, cada uno contiene 'name' y 'latency' para los mejores proxies.
Retorna una lista vacía si no se encuentran proxies probables o ocurre un error.
"""
logging.info("Iniciando prueba de velocidad de proxies de Clash vía API Externa (concurrentemente)...")
logging.info(f"Usando URL de prueba: {LATENCY_TEST_URL}")
logging.info(f"Ejecutando {CONCURRENT_CONNECTIONS} pruebas a la vez. Esto puede tomar un momento...")
proxy_names_to_test = get_all_proxy_names()
if not proxy_names_to_test:
logging.warning("No se encontraron proxies probables o ocurrió un error durante la obtención de nombres de proxy.")
return []
logging.info(f"Se encontraron {len(proxy_names_to_test)} proxies individuales para probar.")
proxy_latencies = {}
with ThreadPoolExecutor(max_workers=CONCURRENT_CONNECTIONS) as executor:
future_to_proxy = {executor.submit(test_proxy_latency, name): name for name