auto/manual/clash/switch_proxy.py

import httpx
import time
import json
import uuid
import logging
import websockets
from typing import Optional, List
from websockets import WebSocketCommonProtocol

logging.basicConfig(level=logging.INFO)


class ClashProxyManager:
    def __init__(self):
        self.clash_api_base_url = [
            "http://192.168.31.201:58001/api",
            "http://192.168.31.201:58002/api",
            "http://192.168.31.201:58003/api",
            "http://192.168.31.201:58004/api",
            "http://192.168.31.201:58005/api",
            "http://192.168.31.201:58006/api",
            "http://192.168.31.201:58007/api",
            "http://192.168.31.201:58008/api",
            "http://192.168.31.201:58009/api",
            "http://192.168.31.201:58010/api",
        ]
        self.proxy_keyword = ['sg', 'SG', '新加坡', '马来西亚']
        self.all_proxies = []
        self.selected_proxies = []

    def get_all_proxies(self, clash_tool_url: str) -> List[str]:
        url = f"{clash_tool_url}/proxies"
        try:
            response = httpx.get(url)
            response.raise_for_status()
            proxies = response.json()
            logging.info("Available proxies:")
            # 输出读取的所有代理信息
            # for proxy_name, proxy_info in proxies['proxies'].items():
            #     logging.info(f"Name: {proxy_name}, Type: {proxy_info.get('type', 'Unknown')}")
            return list(proxies['proxies'].keys())
        except Exception as e:
            logging.error(f"Failed to get proxies: {e}")
            return []

    def filter_proxy(self):
        for keyword in self.proxy_keyword:
            for item in self.all_proxies:
                if keyword.lower() in item.lower():
                    self.selected_proxies.append(item)

    def switch_proxy(self, proxy_name: str, clash_api_base_url: str) -> None:
        url = f"{clash_api_base_url}/proxies/GLOBAL"
        data = {"name": proxy_name}
        try:
            response = httpx.put(url, json=data)
            if response.status_code == 204:
                logging.info(f"Switched to proxy: {proxy_name}")
            else:
                logging.error(f"Failed to switch proxy: {response.status_code} - {proxy_name}")
        except Exception as e:
            logging.error(f"Failed to switch proxy: {e}")

    def main(self) -> None:
        # 读取所有代理
        if not self.all_proxies:
            for clash_tool_url in self.clash_api_base_url:
                proxies = self.get_all_proxies(clash_tool_url)
                if proxies:
                    self.all_proxies = proxies
                    break

        if not self.all_proxies:
            logging.error("Failed to get all proxies")
            return

        # 通过关键词过滤出需要的代理
        self.filter_proxy()
        if not self.selected_proxies:
            logging.error("Failed to filter proxies")
            return

        # 遍历所有的线路api, 切换代理
        for select_proxy, clash_api_base_url in zip(self.selected_proxies, self.clash_api_base_url):
            self.switch_proxy(select_proxy, clash_api_base_url)


if __name__ == "__main__":
    manager = ClashProxyManager()
    manager.main()