zhitou_trade/iptest.py

import winreg
import ctypes
import re
import sys
import time
import requests
import logging

# 日志配置：同时输出到文件和控制台
logger = logging.getLogger(__name__)

def get_init_ip():
    return requests.get('https://checkip.amazonaws.com').text.strip()

def get_proxy_ip(proxy_ip,proxy_port):
    """测试代理ip，并返回代理IP的IP值,由于暂时只能使用http连接，不支持https连接"""
    while True:
        try:
            url = 'https://checkip.amazonaws.com'
            proxyMeta = 'http://45qac9:n2tc94fc@{}:{}'.format(proxy_ip,proxy_port)
            proxies = { 'http':proxyMeta,"https": proxyMeta}
            return requests.get(url, proxies=proxies).text
        except Exception as e:
            logger.info(f'出错了{e}')
            time.sleep(2)



def set_global_proxy(proxy_ip, proxy_port, enable=True):
    """设置/关闭全局代理"""
    key = None
    try:
        if ctypes.windll.shell32.IsUserAnAdmin() == 0:
            logging.error("需要管理员权限来修改系统代理设置")
            return False

        key = winreg.OpenKey(
            winreg.HKEY_CURRENT_USER,
            r"Software\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
            0, winreg.KEY_WRITE | winreg.KEY_READ
        )

        # 备份当前设置
        try:
            old_proxy = winreg.QueryValueEx(key, "ProxyServer")[0]
            old_enable = winreg.QueryValueEx(key, "ProxyEnable")[0]
            old_override = winreg.QueryValueEx(key, "ProxyOverride")[0]
        except WindowsError:
            old_proxy = ""
            old_enable = 0
            old_override = "<local>"

        if enable:
            proxy_string = f"{proxy_ip}:{proxy_port}"
            winreg.SetValueEx(key, "ProxyServer", 0, winreg.REG_SZ, proxy_string)
            winreg.SetValueEx(key, "ProxyEnable", 0, winreg.REG_DWORD, 1)
            bypass = "<local>;" + ";".join(["/api", "/static"])
            winreg.SetValueEx(key, "ProxyOverride", 0, winreg.REG_SZ, bypass)
        else:
            winreg.SetValueEx(key, "ProxyEnable", 0, winreg.REG_DWORD, 0)

        # 刷新系统设置
        internet_set_option = ctypes.windll.Wininet.InternetSetOptionW
        internet_set_option(0, 37, 0, 0)  # INTERNET_OPTION_REFRESH
        internet_set_option(0, 39, 0, 0)  # INTERNET_OPTION_SETTINGS_CHANGED

        logging.info("代理设置已" + ("启用" if enable else "关闭"))
        if enable:
            logging.info(f"代理服务器: {proxy_ip}:{proxy_port}")
        return True

    except Exception as e:
        logging.error(f"设置代理时发生错误: {str(e)}")
        if key is not None:
            try:
                winreg.SetValueEx(key, "ProxyServer", 0, winreg.REG_SZ, old_proxy)
                winreg.SetValueEx(key, "ProxyEnable", 0, winreg.REG_DWORD, old_enable)
                winreg.SetValueEx(key, "ProxyOverride", 0, winreg.REG_SZ, old_override)
                logging.info("已恢复原始代理设置")
            except:
                logging.warning("警告：无法恢复原始代理设置")
        return False
    finally:
        if key:
            winreg.CloseKey(key)


def get_ip_data():
    """从API获取代理IP和端口"""
    url = "http://api.tianqiip.com/getip"
    params = {
        "secret": "d8wqfdf0qhrnxgne",  # 替换为你的提取秘钥
        "num": 1,
        "yys": "电信",
        "type": "json",
        "lb": "\n",
        # "region": "1,2,3",
        "port": 1,
        "time": 5,
        "ts": 1,
        "ys": 1,
        "cs": 1,
        "sign": "386ff88188185bc6070ec011266745b3",  # 用户签名
        "mr": 1
    }
    response = requests.get(url,params)
    if response.status_code == 200:
        data = response.json()
        if data.get("code") == 1000:
            logging.info("获取代理IP成功")
            return data.get("data")[0]



def reset_proxy_to_default():
    """恢复系统代理为默认（关闭）"""
    key = winreg.OpenKey(
        winreg.HKEY_CURRENT_USER,
        r"Software\\Microsoft\\Windows\\CurrentVersion\\Internet Settings",
        0, winreg.KEY_WRITE
    )
    try:
        winreg.SetValueEx(key, "ProxyEnable", 0, winreg.REG_DWORD, 0)
        winreg.SetValueEx(key, "ProxyServer", 0, winreg.REG_SZ, "")
        winreg.SetValueEx(key, "ProxyOverride", 0, winreg.REG_SZ, "<local>")
    finally:
        winreg.CloseKey(key)
    internet_set_option = ctypes.windll.Wininet.InternetSetOptionW
    internet_set_option(0, 37, 0, 0)
    internet_set_option(0, 39, 0, 0)


def is_admin():
    """检查是否以管理员身份运行"""
    try:
        return ctypes.windll.shell32.IsUserAnAdmin()
    except:
        return False


def fetch_whitelist():
    """
    获取当前IP白名单，返回IP列表
    """
    url = "http://api.tianqiip.com/white/fetch?key=Maxinxin&brand=2&sign=386ff88188185bc6070ec011266745b3"
    try:
        response = requests.get(url, timeout=10)
        data = response.json()
        if data.get("code") == 200:
            # 返回IP列表
            return data.get("data")
        else:
            print(f"获取白名单失败: {data.get('msg', 'Unknown error')}")
            return []
    except Exception as e:
        print(f"请求异常: {e}")
        return []


if __name__ == "__main__":
    if not is_admin():
        ctypes.windll.shell32.ShellExecuteW(None, "runas", sys.executable, " ".join(sys.argv), None, 1)
        sys.exit(0)

    ip_fetch_count = 0  # 计数器
    while True:
        ip_fetch_count += 1
        start_time = time.time()
        reset_proxy_to_default()
        my_ip = get_init_ip()
        whitelist = fetch_whitelist()
        print(whitelist)
        if my_ip in whitelist:
            logging.info(f"当前IP {my_ip} 在白名单中")
        else:
            logging.info(f"当前IP {my_ip} 不在白名单中")
        print(1)
        item = get_ip_data()
        proxy_ip = item['ip']
        proxy_port = item['port']
        set_global_proxy(proxy_ip, proxy_port, enable=True)
        try:

            my_ip1 = get_proxy_ip(proxy_ip,proxy_port)
            logging.info(f"第{ip_fetch_count}次获取IP，当前IP: {my_ip1}")
        except Exception as e:
            logging.error(f"第{ip_fetch_count}次获取IP失败: {e}")
        logging.info(f"本次代理切换耗时: {time.time() - start_time:.2f}秒")
        time.sleep(300)