[Python]代理批量检测延迟工具

[Python]代理批量检测延迟工具

成品:https://download.csdn.net/download/G971005287W/90524014

import sys
import time
import threading
import tkinter as tk
from tkinter import ttk, messagebox, filedialog


try:
    import requests
except ImportError:
    print("错误: 无法导入requests模块，请确保已正确安装requests")
    print("可以使用以下命令安装: pip install requests")
    sys.exit(1)

import concurrent.futures  # 导入线程池模块


class ProxyCheckerApp:
    def __init__(self, root):
        self.root = root
        self.root.title("代理检测工具 - 简易版")
        self.root.geometry("1600x800")  # 扩大窗口以适应列表框
        
        # 创建变量
        self.proxy_list_var = tk.StringVar(value="代理列表")
        self.url_var = tk.StringVar(value="http://example.com")
        self.status_var = tk.StringVar(value="就绪")
        self.result_var = tk.StringVar(value="未检测")
        self.last_check_times = {}  # 添加字典来存储每个代理的最后检测时间
        
        # 创建界面
        self.create_widgets()
    
    def create_widgets(self):
        # 创建主框架
        main_frame = ttk.Frame(self.root, padding="20")
        main_frame.grid(row=0, column=0, sticky=tk.NSEW)  # 修改此处，使用grid布局
        
        # 创建输入区域
        input_frame = ttk.LabelFrame(main_frame, text="代理设置", padding="10")
        input_frame.grid(row=0, column=0, sticky=tk.NSEW, pady=(0, 10))  # 修改此处，使用grid布局
    
        # 代理列表
        ttk.Label(input_frame, text="代理列表文件路径:").grid(row=0, column=0, sticky=tk.W, padx=5, pady=5)
        ttk.Entry(input_frame, textvariable=self.proxy_list_var, width=70, state='readonly').grid(row=0, column=1, sticky=tk.W+tk.E, padx=5, pady=5)  # 设置为只读
        browse_btn = ttk.Button(input_frame, text="浏览", command=self.browse_proxy_list)  # 添加浏览按钮
        browse_btn.grid(row=0, column=2, padx=5)
        
        # 手动添加代理
        ttk.Label(input_frame, text="名称:").grid(row=1, column=0, sticky=tk.W, padx=5, pady=5)
        self.name_var = tk.StringVar()
        ttk.Entry(input_frame, textvariable=self.name_var, width=20).grid(row=1, column=1, sticky=tk.W, padx=5, pady=5)
        
        ttk.Label(input_frame, text="IP:").grid(row=2, column=0, sticky=tk.W, padx=5, pady=5)
        self.ip_var = tk.StringVar()
        ttk.Entry(input_frame, textvariable=self.ip_var, width=20).grid(row=2, column=1, sticky=tk.W, padx=5, pady=5)
        
        ttk.Label(input_frame, text="端口:").grid(row=3, column=0, sticky=tk.W, padx=5, pady=5)
        self.port_var = tk.StringVar()
        ttk.Entry(input_frame, textvariable=self.port_var, width=20).grid(row=3, column=1, sticky=tk.W, padx=5, pady=5)
        
        ttk.Label(input_frame, text="用户名:").grid(row=4, column=0, sticky=tk.W, padx=5, pady=5)
        self.username_var = tk.StringVar()
        ttk.Entry(input_frame, textvariable=self.username_var, width=20).grid(row=4, column=1, sticky=tk.W, padx=5, pady=5)
        
        ttk.Label(input_frame, text="密码:").grid(row=5, column=0, sticky=tk.W, padx=5, pady=5)
        self.password_var = tk.StringVar()
        ttk.Entry(input_frame, textvariable=self.password_var, width=20).grid(row=5, column=1, sticky=tk.W, padx=5, pady=5)
        
        add_proxy_btn = ttk.Button(input_frame, text="添加代理", command=self.add_proxy)
        add_proxy_btn.grid(row=6, column=0, columnspan=2, pady=10)
        
        # 测试URL
        ttk.Label(input_frame, text="测试URL:").grid(row=7, column=0, sticky=tk.W, padx=5, pady=5)
        ttk.Entry(input_frame, textvariable=self.url_var, width=70).grid(row=7, column=1, sticky=tk.W+tk.E, padx=5, pady=5)
        
        # 创建按钮区域
        button_frame = ttk.Frame(main_frame)
        button_frame.grid(row=1, column=0, sticky=tk.NSEW, pady=10)  # 修改此处，使用grid布局
        
        # 检测按钮
        self.check_btn = ttk.Button(button_frame, text="检测代理", command=self.check_proxies)
        self.check_btn.grid(row=0, column=0, padx=5)
        
        # 清除按钮
        self.clear_btn = ttk.Button(button_frame, text="清除结果", command=self.clear_result)
        self.clear_btn.grid(row=0, column=1, padx=5)
        
        # 创建结果区域
        result_frame = ttk.LabelFrame(main_frame, text="检测结果", padding="10")
        result_frame.grid(row=0, column=1, sticky=tk.NSEW, pady=10)  # 修改此处，使用grid布局
    
        # 结果列表框
        self.result_listbox = tk.Listbox(result_frame, width=100, height=30)  # 增加高度
        self.result_listbox.grid(row=0, column=0, sticky=tk.NSEW, pady=20)  # 修改此处，使用grid布局
    
        # 状态栏
        status_bar = ttk.Label(self.root, textvariable=self.status_var, relief=tk.SUNKEN, anchor=tk.W)
        status_bar.grid(row=1, column=0, columnspan=2, sticky=tk.NSEW)  # 修改此处，使用grid布局
    
        # 定时检测设置
        ttk.Label(input_frame, text="检测间隔(秒):").grid(row=8, column=0, sticky=tk.W, padx=5, pady=5)
        self.interval_var = tk.StringVar(value="60")
        ttk.Entry(input_frame, textvariable=self.interval_var, width=10).grid(row=8, column=1, sticky=tk.W, padx=5, pady=5)
        
        self.auto_check_var = tk.BooleanVar()
        ttk.Checkbutton(input_frame, text="自动检测", variable=self.auto_check_var).grid(row=8, column=2, padx=5)
    
    def browse_proxy_list(self):
        # 打开文件选择对话框
        file_path = filedialog.askopenfilename(filetypes=[("Text files", "*.txt"), ("All files", "*.*")])
        if file_path:
            self.proxy_list_var.set(file_path)
    
    def add_proxy(self):
        name = self.name_var.get()
        ip = self.ip_var.get()
        port = self.port_var.get()
        username = self.username_var.get()
        password = self.password_var.get()
        
        if not all([name, ip, port]):
            messagebox.showwarning("警告", "请填写名称、IP和端口")
            return
        
        proxy_entry = f"{name} {ip} {port} {username} {password}"
        with open("proxies.txt", "a") as file:
            file.write(proxy_entry + "\n")
        
        messagebox.showinfo("成功", "代理已添加到proxies.txt")
        self.proxy_list_var.set("proxies.txt")
    
    def check_proxies(self):
        # 禁用检测按钮，防止重复点击
        self.check_btn.config(state=tk.DISABLED)
        self.result_listbox.delete(0, tk.END)  # 清空结果列表框
        self.status_var.set("正在检测代理...")
        
        # 读取代理列表文件，指定编码为utf-8
        proxy_list_path = self.proxy_list_var.get()
        try:
            with open(proxy_list_path, 'r', encoding='utf-8') as file:
                proxies = file.readlines()
        except FileNotFoundError:
            messagebox.showerror("错误", "代理列表文件未找到")
            self.check_btn.config(state=tk.NORMAL)
            return
        except Exception as e:
            messagebox.showerror("错误", f"读取代理列表文件时发生错误: {str(e)}")
            self.check_btn.config(state=tk.NORMAL)
            return
        
        # 提前将所有代理名称添加到结果列表框中
        self.proxy_results = {proxy.strip(): "等待检测..." for proxy in proxies if proxy.strip()}
        for proxy in self.proxy_results:
            self.result_listbox.insert(tk.END, f"⌛ {proxy} | 状态: {self.proxy_results[proxy]}")
            self.last_check_times[proxy.strip()] = None  # 初始化最后检测时间为None
        
        # 创建一个新线程来执行检测任务
        threading.Thread(target=self.run_proxy_checks, daemon=True).start()

    def run_proxy_checks(self):
        # 创建线程池执行检测
        with concurrent.futures.ThreadPoolExecutor() as executor:
            futures = {executor.submit(self.test_proxy, proxy): proxy for proxy in self.proxy_results}
            for future in concurrent.futures.as_completed(futures):
                proxy = futures[future]
                try:
                    result, color = future.result()
                    self.update_result(proxy, result, color)
                except Exception as e:
                    result = f"⚠️ 发生未知错误: {str(e)} | {proxy}"
                    color = "#ffcccc"  # 浅红色
                    self.update_result(proxy, result, color)

        self.status_var.set("检测完成")
        self.check_btn.config(state=tk.NORMAL)
        
        # 如果自动检测已启用，则重新设置定时器
        if self.auto_check_var.get():
            interval = int(self.interval_var.get())
            self.root.after(interval * 1000, self.check_proxies)

    def test_proxy(self, proxy):
        proxy = proxy.strip()
        if not proxy:
            return f"⚠️ 代理格式错误 | {proxy}", "#ffcccc"  # 浅红色
        
        # 解析代理信息
        parts = proxy.split()
        if len(parts) < 3:
            return f"⚠️ 代理格式错误 | {proxy}", "#ffcccc"  # 浅红色
        name, ip, port, *auth = parts
        proxy_url = f"http://{ip}:{port}"
        
        if auth:
            username, password = auth[0], auth[1] if len(auth) > 1 else ""
            proxy_url = f"http://{username}:{password}@{ip}:{port}"
        
        proxies_dict = {
            "http": proxy_url,
            "https": proxy_url
        }
        
        try:
            # 记录请求开始时间
            start_time = time.time()
            
            # 发送测试请求
            response = requests.get(self.url_var.get(),
                                    proxies=proxies_dict,
                                    timeout=10,
                                    headers={'User-Agent': 'Mozilla/5.0'})
            
            # 计算延迟（毫秒）
            latency = (time.time() - start_time) * 1000
            
            # 检查响应状态
            if response.status_code == 200:
                result = f"✅ 代理可用 | 延迟: {latency:.2f}ms | {proxy}"
                color = "#ccffcc"  # 浅绿色
            else:
                result = f"⚠️ 代理返回异常状态码: {response.status_code} | {proxy}"
                color = "#ffffcc"  # 浅黄色
        except requests.exceptions.ProxyError:
            result = f"❌ 代理认证失败或无法连接代理服务器 | {proxy}"
            color = "#ffcccc"  # 浅红色
        except requests.exceptions.ConnectTimeout:
            result = f"⌛ 连接超时 | {proxy}"
            color = "#ffcccc"  # 浅红色
        except requests.exceptions.ConnectionError:
            result = f"🔌 网络连接错误 | {proxy}"
            color = "#ffcccc"  # 浅红色
        except Exception as e:
            result = f"⚠️ 发生未知错误: {str(e)} | {proxy}"
            color = "#ffcccc"  # 浅红色
        
        # 更新最后检测时间
        self.last_check_times[proxy] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        
        return result, color

    def update_result(self, proxy, result, color):
        # 找到代理对应的行并更新结果
        for i in range(self.result_listbox.size()):
            if proxy in self.result_listbox.get(i):
                self.result_listbox.delete(i)
                self.result_listbox.insert(i, f"{result} | 最后检测时间: {self.last_check_times[proxy]}")
                self.result_listbox.itemconfig(i, {'bg': color})
                break

    def clear_result(self):
        self.result_listbox.delete(0, tk.END)
        self.status_var.set("就绪")

if __name__ == "__main__":
    root = tk.Tk()
    app = ProxyCheckerApp(root)
    root.mainloop()