Automatisierung der Clash-Proxy-Verwaltung | Original, von KI übersetzt

Home PDF

Dieser Beitrag beschreibt ein Python-Skript, clash.py, das entwickelt wurde, um die Verwaltung Ihrer Clash-Proxy-Konfiguration zu automatisieren. Es übernimmt alles von der regelmäßigen Aktualisierung der Proxy-Konfiguration durch Herunterladen neuer Einstellungen und dem Neustart des Clash-Dienstes bis hin zur intelligenten Auswahl und Umschaltung auf den schnellsten verfügbaren Proxy innerhalb einer bestimmten Gruppe. Ergänzend dazu ermöglicht das Modul speed.py die gleichzeitige Latenzmessung einzelner Clash-Proxys, um sicherzustellen, dass Ihre Verbindung immer über den optimalen Server geleitet wird.

clash.py

import os
import subprocess
import time
import shutil
import argparse
import logging
import requests
import json
import urllib.parse

# Annahme: speed.py befindet sich im selben Verzeichnis oder ist über PYTHONPATH erreichbar
from speed import get_top_proxies 

# --- Konfiguration ---
CLASH_CONTROLLER_HOST = "127.0.0.1"
CLASH_CONTROLLER_PORT = 9090
CLASH_API_BASE_URL = f"http://{CLASH_CONTROLLER_HOST}:{CLASH_CONTROLLER_PORT}"
# Der Name der Proxy-Gruppe, der der beste einzelne Proxy zugewiesen wird.
# Stellen Sie sicher, dass diese Gruppe in Ihrer Clash-Konfiguration existiert.
TARGET_PROXY_GROUP = "🚧Proxy" 

def setup_logging():
    """Konfiguriert die grundlegende Protokollierung für das Skript."""
    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):
    """Setzt systemweite Proxy-Umgebungsvariablen."""
    os.environ["GLOBAL_PROXY"] = global_proxy_address # Für Konsistenz, falls benötigt
    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}"
    # Diese müssen normalerweise nicht explizit auf "false" gesetzt werden,
    # bleiben aber aus Kompatibilitätsgründen erhalten.
    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"Systemweiter Proxy gesetzt auf: {global_proxy_address}")

def stop_system_proxy():
    """Löscht systemweite Proxy-Umgebungsvariablen."""
    os.environ["http_proxy"] = ""
    os.environ["HTTP_PROXY"] = ""
    os.environ["https_proxy"] = ""
    os.environ["HTTPS_PROXY"] = ""
    os.environ["HTTP_PROXY_REQUEST_FULLURI"] = "true" # Auf Standard zurücksetzen
    os.environ["HTTPS_PROXY_REQUEST_FULLURI"] = "true"
    os.environ["ALL_PROXY"] = ""
    logging.info("Systemweiter Proxy gestoppt (Umgebungsvariablen gelöscht).")

def switch_clash_proxy_group(group_name, proxy_name):
    """
    Wechselt den aktiven Proxy in einer bestimmten Clash-Proxy-Gruppe zu einem neuen 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"Erfolgreich gewechselt: '{group_name}' zu '{proxy_name}'.")
        return True
    except requests.exceptions.ConnectionError:
        logging.error(f"Fehler: Verbindung zur Clash-API unter {CLASH_API_BASE_URL} fehlgeschlagen.")
        logging.error("Stellen Sie sicher, dass Clash läuft und der external-controller konfiguriert ist.")
        return False
    except requests.exceptions.Timeout:
        logging.error(f"Fehler: Timeout bei der Verbindung zur Clash-API während des Proxy-Wechsels für '{group_name}'.")
        return False
    except requests.exceptions.RequestException as e:
        logging.error(f"Ein unerwarteter Fehler ist beim Wechseln des Proxys für '{group_name}' aufgetreten: {e}")
        return False

def main():
    """Hauptfunktion zur Verwaltung der Clash-Konfiguration, Neustart und Auswahl des besten Proxys."""
    setup_logging()
    
    parser = argparse.ArgumentParser(description="Clash-Konfigurations- und Verwaltungsskript.")
    parser.add_argument("--minutes", type=int, default=10, help="Minuten zwischen Updates (Standard: 10)")
    parser.add_argument("--iterations", type=int, default=1000, help="Anzahl der Iterationen (Standard: 1000)")
    parser.add_argument(
        "--config-url", 
        type=str, 
        default=os.getenv("CLASH_DOWNLOAD_URL"),
        help="URL zum Herunterladen der Clash-Konfiguration. Standardmäßig wird die Umgebungsvariable CLASH_DOWNLOAD_URL verwendet, falls gesetzt."
    )
    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("Fehler: Keine Konfigurations-URL angegeben. Bitte setzen Sie die Umgebungsvariable CLASH_DOWNLOAD_URL oder verwenden Sie das Argument --config-url.")
        return # Beenden, falls keine URL verfügbar ist

    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"--- Starte Iteration {i} von {ITERATIONS} ---")

        # Schritt 1: Bestehende Proxy-Einstellungen stoppen
        stop_system_proxy()
        
        # Schritt 2: Clash-Konfiguration herunterladen und aktualisieren
        try:
            logging.info(f"Lade neue Konfiguration von: {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-Konfiguration erfolgreich aktualisiert!")
        except subprocess.CalledProcessError as e:
            logging.error(f"Fehler beim Herunterladen oder Verschieben der Konfigurationsdatei: {e.stderr.decode().strip()}")
            logging.error("Überspringe zur nächsten Iteration.")
            time.sleep(10) # Kurze Wartezeit vor dem erneuten Versuch
            continue
        except Exception as e:
            logging.error(f"Ein unerwarteter Fehler ist während der Konfigurationsaktualisierung aufgetreten: {e}")
            logging.error("Überspringe zur nächsten Iteration.")
            time.sleep(10)
            continue

        # Schritt 3: Clash im Hintergrund starten
        clash_process = None
        try:
            # Es ist wichtig, dass Clash mit aktiviertem external-controller startet
            # Dies wird normalerweise in der config.yaml konfiguriert.
            clash_process = subprocess.Popen([clash_executable_path], 
                                             stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            logging.info(f"Clash gestartet mit PID {clash_process.pid}")
            # Clash etwas Zeit geben, um vollständig zu initialisieren
            time.sleep(5) 
        except FileNotFoundError:
            logging.critical(f"Clash-Executable nicht gefunden unter: {clash_executable_path}")
            logging.critical("Bitte stellen Sie sicher, dass der Pfad korrekt ist und Clash installiert ist.")
            return # Kritischer Fehler, Skript beenden
        except Exception as e:
            logging.error(f"Fehler beim Starten von Clash: {e}")
            logging.error("Überspringe zur nächsten Iteration.")
            if clash_process: clash_process.terminate()
            time.sleep(10)
            continue

        # Schritt 4: Proxy-Geschwindigkeiten testen und den besten auswählen
        best_proxy_name = None
        try:
            logging.info("Teste Proxy-Geschwindigkeiten, um den besten zu finden...")
            top_proxies = get_top_proxies(num_results=1) # Nur den besten Proxy abrufen
            if top_proxies:
                best_proxy_name = top_proxies[0]['name']
                logging.info(f"Bester Proxy identifiziert: '{best_proxy_name}' mit Latenz {top_proxies[0]['latency']}ms")
            else:
                logging.warning("Keine erfolgreichen Proxy-Tests. Kann für diese Iteration keinen besten Proxy auswählen.")
        except Exception as e:
            logging.error(f"Fehler während des Proxy-Geschwindigkeitstests: {e}")

        # Schritt 5: Clashs Proxy-Gruppe auf den besten Proxy umschalten (falls gefunden)
        if best_proxy_name:
            # Vor dem Setzen des System-Proxys sicherstellen, dass Clash korrekt eingerichtet ist.
            # Setzt den systemweiten Proxy auf den lokalen HTTP-Proxy von Clash.
            # Clash läuft normalerweise auf Port 7890 (oder ähnlich, überprüfen Sie Ihre Konfiguration).
            clash_local_proxy_address = f"{CLASH_CONTROLLER_HOST}:7890" # Anpassen, falls Ihr Clash-HTTP-Port abweicht
            start_system_proxy(clash_local_proxy_address)
            
            if not switch_clash_proxy_group(TARGET_PROXY_GROUP, best_proxy_name):
                logging.error(f"Fehler beim Wechseln der Clash-Gruppe '{TARGET_PROXY_GROUP}' zu '{best_proxy_name}'.")
        else:
            logging.warning("Kein bester Proxy gefunden, überspringe Proxy-Gruppenwechsel und System-Proxy-Einrichtung für diese Iteration.")
            
        # Schritt 6: Auf die angegebene Dauer warten
        logging.info(f"Warte {SLEEP_SECONDS / 60} Minuten vor der nächsten Iteration...")
        time.sleep(SLEEP_SECONDS)

        # Schritt 7: Clash-Prozess stoppen
        if clash_process:
            logging.info("Beende Clash-Prozess...")
            clash_process.terminate()
            try:
                clash_process.wait(timeout=10) # Clash etwas Zeit für einen sauberen Stopp geben
                logging.info("Clash erfolgreich gestoppt.")
            except subprocess.TimeoutExpired:
                logging.warning("Clash hat sich nicht ordnungsgemäß beendet, Prozess wird beendet.")
                clash_process.kill()
                clash_process.wait() # Sicherstellen, dass der Prozess vollständig beendet ist
            except Exception as e:
                logging.error(f"Fehler beim Warten auf das Beenden von Clash: {e}")
        
        logging.info(f"--- Iteration {i} abgeschlossen ---")

    logging.info(f"{ITERATIONS} Iterationen abgeschlossen. Skript beendet.")

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 # Importiert das logging-Modul

— Konfiguration —

CLASH_CONTROLLER_HOST = “127.0.0.1” # Verwendet 127.0.0.1, da der Controller auf demselben Rechner läuft CLASH_CONTROLLER_PORT = 9090 CLASH_API_BASE_URL = f”http://{CLASH_CONTROLLER_HOST}:{CLASH_CONTROLLER_PORT}” LATENCY_TEST_URL = “https://github.com” # Aktualisierte Test-URL LATENCY_TEST_TIMEOUT_MS = 5000 # Millisekunden CONCURRENT_CONNECTIONS = 10 # Anzahl gleichzeitiger Tests

Liste bekannter Proxy-Gruppennamen, die vom Geschwindigkeitstest ausgeschlossen werden sollen

Dies sind typischerweise keine einzelnen Knoten, sondern Policy-Gruppen oder spezielle Proxys.

EXCLUDE_PROXY_GROUPS = [ “DIRECT”, “REJECT”, “GLOBAL”, # Standardmäßig bereits in der API ausgeschlossen “🇨


Back 2025.06.30 Donate