auto/manual/MyGrassio/random_proxy.py

# -*- coding: utf-8 -*-
'''
切换到随机代理
'''
import random

import httpx
import time
import logging
import subprocess
from typing import Optional, List

logging.basicConfig(level=logging.INFO)

BASE_URL = "http://192.168.31.201"

PORT_LIST = [
    ["52001", "53001"],
    ["52002", "53002"],
    ["52003", "53003"],
    ["52004", "53004"],
    ["52005", "53005"],
    ["52006", "53006"],
    ["52007", "53007"],
    ["52008", "53008"],
    ["52009", "53009"],
    ["52011", "53012"],
    ["52012", "53012"]
]


class ClashProxyManager:
    def __init__(self, base_url, base_port):
        self.key_group = 0
        self.base_url = base_url
        self.base_port = base_port
        self.all_proxies = []

    def get_all_proxies(self, clash_tool_url: str) -> List[str]:
        url = f"{clash_tool_url}/api/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')}")
            proxy_list = list(proxies['proxies'].keys())
            filtered_list = [item for item in proxy_list if item not in {'REJECT', 'GLOBAL', 'DIRECT'}]
            return filtered_list
        except Exception as e:
            logging.error(f"Failed to get proxies: {e}")
            return []

    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)
            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 update_configs(self):
        for base_port in self.base_port:
            url_and_port = self.base_url + ":" + base_port[0]
            key = "/api/configs"
            url = url_and_port + key

            headers = {
                "accept": "application/json, text/plain, */*",
                "accept-encoding": "gzip, deflate, br, zstd",
                "accept-language": "zh-CN,zh",
                "connection": "keep-alive",
                "content-type": "application/json",
                "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"',
                "sec-fetch-dest": "empty",
                "sec-fetch-mode": "cors",
                "sec-fetch-site": "same-origin",
                "sec-gpc": "1",
                "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36",
            }

            # 请求体数据
            data = {"mode": "Global"}  # 替换为实际的请求数据

            # 使用 httpx 发送 PATCH 请求
            try:
                with httpx.Client() as client:
                    response = client.patch(url, headers=headers, json=data)
                    if response.status_code == 204:
                        print(f"{url} OK")
                    else:
                        print("响应内容:", response.text)
            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 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:
                    self.all_proxies = proxies
                    break

        if not self.all_proxies:
            logging.error("Failed to get all proxies")
            return
        else:
            logging.info(f"Found {self.all_proxies} proxies.")
            logging.info(f"Found {len(self.all_proxies)} proxies.")

        # 遍历所有的线路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:
                # 选一个代理
                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:
                    print(f"{url_and_port} 切换 {choose_proxy} 检测失败")
                    time.sleep(1)


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