自動化されたクラッシュプロキシ管理 | オリジナル、AI翻訳
この投稿では、Clashプロキシ設定の自動管理を目的としたPythonスクリプトclash.pyについて詳しく説明します。このスクリプトは、定期的に更新されたプロキシ設定をダウンロードし、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 = "🚧Proxy"
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環境変数が設定されていない場合はハードコードされた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:
# external-controllerが有効でアクセス可能な状態でClashを起動することが重要
# これは通常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) # 最速のプロキシ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プロキシ（通常ポート7890）に設定
clash_local_proxy_address = f"{CLASH_CONTROLLER_HOST}:7890" # Clash HTTPポートが異なる場合は調整
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
```python 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” # コントローラは同じマシン上にあるため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”, “📺愛奇藝などの国内ストリーミング”, “🚧Proxy”, # 除外したい他のグループ名をここに追加 ]
— speed.py用のロギング設定 —
このスクリプト専用のロギングを設定
speed.pyがインポートされ関数が呼び出された場合、
その出力はclash_manager.logとは別のspeed.logに記録される
logging.basicConfig( filename=’clash.log’, level=logging.INFO, format=’%(asctime)s - %(levelname)s - %(message)s’, datefmt=’%Y-%m-%d %H:%M:%S’ )
必要に応じて、コンソールにも出力したい場合は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)
— スクリプトロジック —
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,