Automatisation de la Gestion des Proxies de Clash | Original, traduit par l'IA
Ce post détaille un script Python, clash.py, conçu pour automatiser la gestion de votre configuration de proxy Clash. Il gère tout, depuis le téléchargement périodique des configurations de proxy mises à jour et le redémarrage du service Clash jusqu’à la sélection intelligente et la bascule vers le proxy disponible le plus rapide au sein d’un groupe désigné. Complémentaire à clash.py, le module speed.py facilite le test de latence concurrent des proxies individuels de Clash, garantissant que votre connexion passe toujours par le serveur optimal.
clash.py
import os
import subprocess
import time
import shutil
import argparse
import logging
import requests
import json
import urllib.parse
# Supposant que speed.py est dans le même répertoire ou accessible dans 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}"
# Le nom du groupe de proxy auquel le meilleur proxy individuel sera assigné.
# Assurez-vous que ce groupe existe dans votre configuration Clash.
TARGET_PROXY_GROUP = "🚧Proxy"
def setup_logging():
"""Configure le logging de base pour le 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):
"""Définit les variables d'environnement du proxy système."""
os.environ["GLOBAL_PROXY"] = global_proxy_address # Défini pour cohérence si nécessaire ailleurs
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}"
# Ces variables n'ont généralement pas besoin d'être explicitement définies à "false" avec les outils modernes,
# mais conservées pour compatibilité avec l'intention originale de votre 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 système défini sur : {global_proxy_address}")
def stop_system_proxy():
"""Efface les variables d'environnement du proxy système."""
os.environ["http_proxy"] = ""
os.environ["HTTP_PROXY"] = ""
os.environ["https_proxy"] = ""
os.environ["HTTPS_PROXY"] = ""
os.environ["HTTP_PROXY_REQUEST_FULLURI"] = "true" # Revenir à la valeur par défaut
os.environ["HTTPS_PROXY_REQUEST_FULLURI"] = "true"
os.environ["ALL_PROXY"] = ""
logging.info("Proxy système arrêté (variables d'environnement effacées).")
def switch_clash_proxy_group(group_name, proxy_name):
"""
Bascule le proxy actif dans un groupe de proxy Clash spécifié vers un nouveau 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"Bascule réussie de '{group_name}' vers '{proxy_name}'.")
return True
except requests.exceptions.ConnectionError:
logging.error(f"Erreur : Impossible de se connecter à l'API Clash à {CLASH_API_BASE_URL} pour basculer le proxy.")
logging.error("Assurez-vous que Clash est en cours d'exécution et que son external-controller est configuré.")
return False
except requests.exceptions.Timeout:
logging.error(f"Erreur : Connexion à l'API Clash expirée lors de la bascule du proxy pour '{group_name}'.")
return False
except requests.exceptions.RequestException as e:
logging.error(f"Une erreur inattendue s'est produite lors de la bascule du proxy pour '{group_name}' : {e}")
return False
def main():
"""Fonction principale pour gérer la configuration de Clash, le redémarrage et la sélection du meilleur proxy."""
setup_logging()
parser = argparse.ArgumentParser(description="Script de configuration et de gestion de Clash.")
parser.add_argument("--minutes", type=int, default=10, help="Minutes entre les mises à jour (par défaut : 10)")
parser.add_argument("--iterations", type=int, default=1000, help="Nombre d'itérations (par défaut : 1000)")
parser.add_argument(
"--config-url",
type=str,
default=os.getenv("CLASH_DOWNLOAD_URL"),
help="URL pour télécharger la configuration de Clash. Par défaut, utilise la variable d'environnement CLASH_DOWNLOAD_URL si définie, sinon une URL codée en dur."
)
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("Erreur : Aucune URL de téléchargement de configuration fournie. Veuillez définir la variable d'environnement CLASH_DOWNLOAD_URL ou utiliser l'argument --config-url.")
return # Quitte si aucune URL n'est 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"--- Début de l'itération {i} sur {ITERATIONS} ---")
# Étape 1 : Arrête tous les paramètres de proxy système existants
stop_system_proxy()
# Étape 2 : Télécharge et met à jour la configuration de Clash
try:
logging.info(f"Téléchargement de la nouvelle configuration depuis : {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("Configuration de Clash mise à jour avec succès !")
except subprocess.CalledProcessError as e:
logging.error(f"Échec du téléchargement ou du déplacement du fichier de configuration : {e.stderr.decode().strip()}")
logging.error("Passage à l'itération suivante.")
time.sleep(10) # Attend un peu avant de réessayer
continue
except Exception as e:
logging.error(f"Une erreur inattendue s'est produite lors de la mise à jour de la configuration : {e}")
logging.error("Passage à l'itération suivante.")
time.sleep(10)
continue
# Étape 3 : Démarre Clash en arrière-plan
clash_process = None
try:
# Il est crucial que Clash démarre avec l'external-controller activé et accessible
# Ceci est généralement configuré dans le fichier config.yaml lui-même.
clash_process = subprocess.Popen([clash_executable_path],
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
logging.info(f"Clash démarré avec le PID {clash_process.pid}")
# Donne un moment à Clash pour s'initialiser complètement et ouvrir son port API
time.sleep(5)
except FileNotFoundError:
logging.critical(f"Exécutable Clash introuvable à l'emplacement : {clash_executable_path}")
logging.critical("Veuillez vérifier que le chemin est correct et que Clash est installé.")
return # Erreur critique, quitte le script
except Exception as e:
logging.error(f"Échec du démarrage de Clash : {e}")
logging.error("Passage à l'itération suivante.")
if clash_process: clash_process.terminate()
time.sleep(10)
continue
# Étape 4 : Teste les vitesses des proxies et sélectionne le meilleur
best_proxy_name = None
try:
logging.info("Test des vitesses des proxies pour trouver le meilleur...")
top_proxies = get_top_proxies(num_results=1) # Obtient uniquement le meilleur proxy
if top_proxies:
best_proxy_name = top_proxies[0]['name']
logging.info(f"Meilleur proxy identifié : '{best_proxy_name}' avec une latence de {top_proxies[0]['latency']}ms")
else:
logging.warning("Aucun test de proxy réussi. Impossible de sélectionner un meilleur proxy pour cette itération.")
except Exception as e:
logging.error(f"Erreur lors du test de vitesse des proxies : {e}")
# Étape 5 : Bascule le groupe de proxy de Clash vers le meilleur proxy (si trouvé)
if best_proxy_name:
# Avant de définir le proxy système, assure que Clash est correctement configuré.
# Définit le proxy système pour pointer vers le proxy HTTP local de Clash.
# Clash exécute généralement son proxy HTTP sur le port 7890 (ou similaire, vérifiez votre configuration).
clash_local_proxy_address = f"{CLASH_CONTROLLER_HOST}:7890" # Ajustez si votre port HTTP Clash est différent
start_system_proxy(clash_local_proxy_address)
if not switch_clash_proxy_group(TARGET_PROXY_GROUP, best_proxy_name):
logging.error(f"Échec de la bascule du groupe Clash '{TARGET_PROXY_GROUP}' vers '{best_proxy_name}'.")
else:
logging.warning("Aucun meilleur proxy trouvé, saute la bascule du groupe de proxy et la configuration du proxy système pour cette itération.")
# Étape 6 : Attend la durée spécifiée
logging.info(f"Attente de {SLEEP_SECONDS / 60} minutes avant la prochaine itération...")
time.sleep(SLEEP_SECONDS)
# Étape 7 : Arrête le processus Clash
if clash_process:
logging.info("Arrêt du processus Clash...")
clash_process.terminate()
try:
clash_process.wait(timeout=10) # Donne un peu plus de temps à Clash pour s'arrêter proprement
logging.info("Clash arrêté avec succès.")
except subprocess.TimeoutExpired:
logging.warning("Clash ne s'est pas arrêté proprement, kill du processus.")
clash_process.kill()
clash_process.wait() # S'assure que le processus est bien tué
except Exception as e:
logging.error(f"Erreur lors de l'attente de l'arrêt de Clash : {e}")
logging.info(f"--- Itération {i} terminée ---")
logging.info(f"{ITERATIONS} itérations terminées. Script terminé.")
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 # Importe le module logging
— Configuration —
CLASH_CONTROLLER_HOST = “127.0.0.1” # Utilise 127.0.0.1 car le contrôleur est sur la même machine CLASH_CONTROLLER_PORT = 9090 CLASH_API_BASE_URL = f”http://{CLASH_CONTROLLER_HOST}:{CLASH_CONTROLLER_PORT}” LATENCY_TEST_URL = “https://github.com” # URL de test mise à jour LATENCY_TEST_TIMEOUT_MS = 5000 # Millisecondes CONCURRENT_CONNECTIONS = 10 # Nombre de tests concurrents
Liste des noms de groupes de proxy connus à exclure des tests de vitesse
Ce ne sont généralement pas des nœuds individuels mais des groupes de politiques ou des proxies spéciaux.
EXCLUDE_PROXY_GROUPS = [ “DIRECT”, “REJECT”, “GLOBAL”, # Déjà exclu par défaut dans l’API “🇨🇳国内网站或资源”, “🌵其它规则外”, “🎬Netflix等国外流媒体”, “📦ChatGPT”, “📹Youtube”, “📺爱奇艺等国内流媒体”, “🚧Proxy”, # Ajoutez ici tout autre nom de groupe que vous souhaitez exclure ]
— Configuration du logging pour speed.py —
Configure le logging pour ce script spécifique
Cela garantit que lorsque speed.py est importé et que ses fonctions sont appelées,
sa sortie va dans speed.log, séparément 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’ )
Optionnellement, si vous souhaitez également voir la sortie dans la console, ajoutez 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)
— Logique du script —
def get_all_proxy_names(): “"”Récupère tous les noms de proxy depuis l’API Clash, en excluant les groupes connus.””” try: response = requests.get(f”{CLASH_API_BASE_URL}/proxies”, timeout=5) response.raise_for_status() # Lève une exception pour les erreurs HTTP (4xx ou 5xx) proxies_data = response.json()
all_names = proxies_data.get("proxies", {}).keys()
# Filtre les noms de groupes
filtered_names = [name for name in all_names if name not in EXCLUDE_PROXY_GROUPS]
logging.info(f"{len(filtered_names)} noms de proxy testables récupérés avec succès.")
return filtered_names
except requests.exceptions.ConnectionError:
logging.error(f"Impossible de se connecter à l'API Clash à {CLASH_API_BASE_URL}. Assurez-vous que Clash est en cours d'exécution.")
return []
except requests.exceptions.Timeout:
logging.error(f"Connexion à l'API Clash expirée après 5 secondes.")
return []
except requests.exceptions.RequestException as e:
logging.error(f"Une erreur inattendue s'est produite lors de la récupération des noms de proxy : {e}")
return []
def test_proxy_latency(proxy_name): “"”Teste la latence d’un seul proxy en utilisant l’API Clash. Retourne un tuple (nom_du_proxy, latence) ou (nom_du_proxy, None) en cas d’échec. “”” 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: # Le timeout de requests est en secondes, convertit les millisecondes 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} - Latence : {latency}ms”) return proxy_name, latency except requests.exceptions.RequestException as e: logging.warning(f”Erreur lors du test de ‘{proxy_name}’ : {e}”) return proxy_name, None
def get_top_proxies(num_results=5): “”” Teste les vitesses des proxies Clash en concurrence et retourne les N proxies individuels les plus rapides.
Retourne :
list : Une liste de dictionnaires, chacun contenant 'name' et 'latency' pour les meilleurs proxies.
Retourne une liste vide si aucun proxy testable n'est trouvé ou en cas d'erreur.
"""
logging.info("Début du test de vitesse des proxies Clash via l'API External Controller (en concurrence)...")
logging.info(f"Utilisation de l'URL de test : {LATENCY_TEST_URL}")
logging.info(f"Exécution de {CONCURRENT_CONNECTIONS} tests à la fois. Cela peut prendre un moment...")
proxy_names_to_test = get_all_proxy_names()
if not proxy_names_to_test:
logging.warning("Aucun