更新 随机切换代理

更新 切换分组代理

manual/clash/set_global_switch_proxy_random.py

@@ -1,36 +1,42 @@
+# -*- coding: utf-8 -*-
+'''
+切换到随机代理
+'''
 import random
 
 import httpx
 import time
-import json
-import uuid
 import logging
-import websockets
+import subprocess
 from typing import Optional, List
-from websockets import WebSocketCommonProtocol
 
 logging.basicConfig(level=logging.INFO)
 
-BASE_URL_LIST = [
-    "http://192.168.31.201:52001",
-    "http://192.168.31.201:52002",
-    "http://192.168.31.201:52003",
-    "http://192.168.31.201:52004",
-    "http://192.168.31.201:52005",
-    "http://192.168.31.201:52006",
-    "http://192.168.31.201:52007",
-    "http://192.168.31.201:52008",
-    "http://192.168.31.201:52009",
-    "http://192.168.31.201:52010",
+BASE_URL = "http://192.168.31.201"
+
+PORT_LIST = [
+    ["58001", "59001"],
+    ["58002", "59002"],
+    ["58003", "59003"],
+    ["58004", "59004"],
+    ["58005", "59005"],
+    ["58006", "59006"],
+    ["58007", "59007"],
+    ["58008", "59008"],
+    ["58009", "59009"],
+    ["58010", "59010"],
 ]
 
+TEST_URL = "https://httpbin.org/ip"
+
 
 class ClashProxyManager:
-    def __init__(self, base_url_list):
+    def __init__(self, base_url, base_port, test_url):
         self.key_group = 0
-        self.base_url_list = base_url_list
+        self.base_url = base_url
+        self.base_port = base_port
+        self.test_url = test_url
         self.all_proxies = []
-        self.filter_seeds = []
 
     def get_all_proxies(self, clash_tool_url: str) -> List[str]:
         url = f"{clash_tool_url}/api/proxies"
@@ -47,8 +53,9 @@ class ClashProxyManager:
             logging.error(f"Failed to get proxies: {e}")
             return []
 
-    def switch_proxy(self, proxy_name: str, clash_api_base_url: str) -> None:
-        url = f"{clash_api_base_url}/api/proxies/GLOBAL"
+    def switch_proxy(self, proxy_name: str, url_and_port: str) -> None:
+        logging.info("switch proxy")
+        url = f"{url_and_port}/api/proxies/GLOBAL"
         data = {"name": proxy_name}
         try:
             response = httpx.put(url, json=data)
@@ -60,9 +67,10 @@ class ClashProxyManager:
             logging.error(f"Failed to switch proxy: {e}")
 
     def update_configs(self):
-        for base_url in self.base_url_list:
+        for base_port in self.base_port:
+            url_and_port = self.base_url + ":" + base_port[0]
             key = "/api/configs"
-            url = base_url + key
+            url = url_and_port + key
 
             headers = {
                 "accept": "application/json, text/plain, */*",
@@ -70,9 +78,8 @@ class ClashProxyManager:
                 "accept-language": "zh-CN,zh",
                 "connection": "keep-alive",
                 "content-type": "application/json",
-                "host": "localhost:17888",
-                "origin": base_url,
-                "referer": base_url,
+                "origin": url_and_port,
+                "referer": url_and_port,
                 "sec-ch-ua": '"Not A(Brand";v="8", "Chromium";v="132", "Brave";v="132"',
                 "sec-ch-ua-mobile": "?0",
                 "sec-ch-ua-platform": '"macOS"',
@@ -97,13 +104,30 @@ class ClashProxyManager:
             except httpx.RequestError as exc:
                 print(f"请求失败: {exc}")
 
+    def check_proxy(self, proxy_url):
+        # proxy_url: 代理地址, 没有密码
+        # 测试目标地址:
+        command = ["curl", "-x", proxy_url, "ip.sb"]
+
+        # 执行命令并获取输出
+        try:
+            result = subprocess.run(command, capture_output=True, text=True, check=True)
+            # 打印命令的标准输出
+            print("Output:", result.stdout)
+            return True
+        except subprocess.CalledProcessError as e:
+            # 如果命令执行失败，打印错误信息
+            print("Error:", e.stderr)
+            return False
+
     def main(self) -> None:
         # 设置全局代理
         self.update_configs()
 
         # 读取所有代理
         if not self.all_proxies:
-            for base_url in self.base_url_list:
+            for port_list in self.base_port:
+                base_url = self.base_url + ":" + port_list[0]
                 clash_tool_url = f"{base_url}"
                 proxies = self.get_all_proxies(clash_tool_url)
                 if proxies:
@@ -114,25 +138,32 @@ class ClashProxyManager:
             logging.error("Failed to get all proxies")
             return
 
-
-        # 遍历所有的线路, 切换代理(需要所有线路不重复), 如果self.filter_seeds为空,则在self.all_proxies选择
-        # 切换完之后测试一下当前代理是否正常工作,否则切换下一个代理, 测试地址: https://api.ipify.org
-        switched_agent = []
-        for base_url in self.base_url_list:
-            clash_api_base_url = f"{base_url}"
-            choice_proxy = random.choice(self.all_proxies)
-            if choice_proxy in switched_agent:
-                continue
+        # 遍历所有的线路api, 切换不重复代理
+        # 切换后, 检测代理, 如果检测返回失败, 再次切换
+        used_proxy = []
+        for base_port in self.base_port:
+            url_and_port = self.base_url + ":" + base_port[0]
+            proxy_url = self.base_url + ":" + base_port[1]
             while True:
-                if self.filter_seeds:
-                    pass
+                # 选一个代理
+                choose_proxy = random.choice(self.all_proxies)
+                # 如果已经选过就继续下一个
+                if choose_proxy in used_proxy:
+                    continue
+
+                # 切换代理
+                self.switch_proxy(choose_proxy, url_and_port)
+
+                # 切换后检测代理
+                if self.check_proxy(proxy_url):
+                    print(f"代理 {choose_proxy} 切换成功，检测通过！")
+                    used_proxy.append(choose_proxy)  # 标记为已使用
+                    break  # 成功后退出当前代理的重试循环
                 else:
-                    self.switch_proxy(choice_proxy, clash_api_base_url)
-                    # 切换完之后测试一下当前代理是否正常工作,否则切换下一个代理
-                    switched_agent.append(choice_proxy)
-
+                    print(f"{url_and_port} 切换 {choose_proxy} 检测失败")
+                    time.sleep(1)
 
 
 if __name__ == "__main__":
-    manager = ClashProxyManager(BASE_URL_LIST)
+    manager = ClashProxyManager(BASE_URL, PORT_LIST, TEST_URL)
     manager.main()

manual/clash/switch_proxy_group_group.py

@@ -1,11 +1,12 @@
+# -*- coding: utf-8 -*-
+'''
+切换到指定代理组
+'''
 import httpx
 import time
-import json
-import uuid
 import logging
-import websockets
+import subprocess
 from typing import Optional, List
-from websockets import WebSocketCommonProtocol
 
 logging.basicConfig(level=logging.INFO)
 
@@ -115,18 +116,17 @@ class ClashProxyManager:
     def check_proxy(self, proxy_url):
         # proxy_url: 代理地址, 没有密码
         # 测试目标地址:
+        command = ["curl", "-x", proxy_url, "ip.sb"]
+
+        # 执行命令并获取输出
         try:
-            # 使用代理发送请求
-            response = httpx.get(
-                TEST_URL,
-                proxies={"http": proxy_url, "https": proxy_url},
-                timeout=5
-            )
-            response.raise_for_status()  # 检查请求是否成功
-            # 如果请求成功，返回 True
+            result = subprocess.run(command, capture_output=True, text=True, check=True)
+            # 打印命令的标准输出
+            print("Output:", result.stdout)
             return True
-        except Exception:
-            # 如果发生任何异常，返回 False
+        except subprocess.CalledProcessError as e:
+            # 如果命令执行失败，打印错误信息
+            print("Error:", e.stderr)
             return False
 
     def main(self) -> None:
@@ -173,9 +173,6 @@ class ClashProxyManager:
                     print(f"{url_and_port} 切换 {select_proxy} 检测失败")
                     time.sleep(1)  # 等待一段时间后重试
 
-                # 如果当前代理成功，直接进入下一个 base_port 的处理
-                break
-
 
 if __name__ == "__main__":
     manager = ClashProxyManager(BASE_URL, PORT_LIST, TEST_URL)