Create proxy_gui_new.py

Author: localhost8083

proxy_gui_new.py

@@ -0,0 +1,390 @@
+import tkinter as tk
+from tkinter import ttk, filedialog, simpledialog
+import requests
+import socket as _socket
+from urllib.request import urlopen
+import csv
+import threading
+from concurrent.futures import ThreadPoolExecutor
+import feedparser
+import re
+import socks
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
+from tkinter.ttk import Progressbar
+from queue import Queue
+from tkinter import StringVar
+from tkinter import IntVar
+import concurrent
+
+
+class ProxyAdapter(HTTPAdapter):
+    def __init__(self, proxy, *args, **kwargs):
+        self.proxy = proxy
+        super(ProxyAdapter, self).__init__(*args, **kwargs)
+
+    def send(self, request, stream=False, timeout=None, verify=True, cert=None):
+        request.proxies = {
+            'http': self.proxy,
+            'https': self.proxy
+        }
+        return super(ProxyAdapter, self).send(request, stream=stream, timeout=timeout, verify=verify, cert=cert)
+
+
+class ProxyChecker:
+    def __init__(self):
+        self.window = tk.Tk()
+        self.window.title("Proxy Checker")
+        self.status_text = tk.StringVar()
+        self.status_text.set("Ready")
+        self.progress_queue = Queue()
+        self.stop_check = threading.Event()
+        self.proxies = []
+        self.find_valid_proxies_var = IntVar()
+        self.find_valid_proxies_var.trace(
+            "w", self.update_check_proxies_button)
+        self.create_widgets()
+
+    def update_check_proxies_button(self, *args):
+        if self.find_valid_proxies_var.get() or self.proxies:
+            # Make button visible
+            self.check_button.grid(
+                row=4, column=2, columnspan=2, padx=10, pady=10)
+            self.check_button.config(state='normal')
+        else:
+            self.check_button.grid(
+                row=4, column=2, columnspan=2, padx=10, pady=10)
+            self.check_button.config(state='disabled')
+
+    def update_status_label(self, *args):
+        self.status_label.config(text=self.status_text.get())
+
+    def update_progress_bar(self, increment=1):
+        self.progress_bar["value"] += increment
+        self.progress_bar.update_idletasks()
+
+    def create_widgets(self):
+        style = ttk.Style()
+        style.configure('TButton', font=('Helvetica', 10))
+        style.configure('TLabel', font=('Helvetica', 10))
+        style.configure('TEntry', font=('Helvetica', 10))
+
+        # Load Proxies button
+
+        self.load_button = ttk.Button(
+            self.window, text="Browse Proxy File", command=self.load_proxies)
+        self.load_button.grid(row=0, column=0, padx=10, pady=10, sticky='w')
+
+        # URL label and entry
+        self.url_label = ttk.Label(self.window, text="URL:")
+        self.url_label.grid(row=1, column=0, padx=10, sticky='w')
+
+        self.url_entry = ttk.Entry(self.window)
+        self.url_entry.grid(row=1, column=1, padx=10, pady=10, sticky='w')
+
+        # Load Proxies from URL button
+        self.load_url_button = ttk.Button(
+            self.window, text="Load Proxies from URL", command=self.start_load_proxies_from_url)
+        self.load_url_button.grid(
+            row=1, column=2, padx=10, pady=10, sticky='w')
+
+        # Stop button
+        self.stop_button = ttk.Button(
+            self.window, text="STOP", command=self.stop_check_proxies, state='disabled')
+        self.stop_button.grid(row=2, column=1, padx=10, pady=10, sticky='w')
+
+        # Status label
+        self.status_label = ttk.Label(
+            self.window, textvariable=self.status_text)
+        self.status_label.grid(row=7, column=0, padx=10, pady=10, sticky='w')
+
+        # Proxy Type label and combobox
+        self.proxy_type_label = ttk.Label(self.window, text="Proxy Type:")
+        self.proxy_type_label.grid(
+            row=3, column=1, padx=10, pady=10, sticky='w')
+
+        self.proxy_type_var = StringVar()
+        self.proxy_type_combobox = ttk.Combobox(
+            self.window, textvariable=self.proxy_type_var, state='readonly')
+        self.proxy_type_combobox['values'] = (
+            'All', 'HTTP', 'HTTPS', 'SOCKS4', 'SOCKS5')
+        self.proxy_type_combobox.current(0)
+        self.proxy_type_combobox.grid(
+            row=3, column=2, padx=10, pady=10, sticky='w')
+
+        # Check Proxies button
+        self.check_button = ttk.Button(
+            self.window, text="Check Proxies", command=self.start_check_proxies, state='disabled')
+        self.check_button.grid(row=4, column=2, padx=10, pady=10, sticky='w')
+
+        # Find Valid Proxies checkbox
+        self.find_valid_proxies_checkbox = ttk.Checkbutton(
+            self.window, text="Find Valid Proxies", variable=self.find_valid_proxies_var)
+        self.find_valid_proxies_checkbox.grid(
+            row=4, column=1, padx=10, pady=10, sticky='w')
+
+        # Valid Proxies label and text box
+        self.output_label = ttk.Label(self.window, text="Valid Proxies:")
+        self.output_label.grid(row=4, column=0, padx=10, pady=10, sticky='w')
+
+        self.output_text = tk.Text(
+            self.window, wrap=tk.WORD, width=60, height=10)
+        self.output_text.grid(row=5, column=0, padx=10,
+                              pady=10, columnspan=3, sticky='w')
+
+        # Export buttons
+        self.export_txt_button = ttk.Button(
+            self.window, text="Export as TXT", command=self.export_txt, state='disabled')
+        self.export_txt_button.grid(
+            row=6, column=0, padx=10, pady=10, sticky='w')
+        self.export_csv_button = ttk.Button(
+            self.window, text="Export as CSV", command=self.export_csv, state='disabled')
+        self.export_csv_button.grid(
+            row=6, column=1, padx=10, pady=10, sticky='w')
+
+        # Progress bar
+        self.progress_bar = ttk.Progressbar(self.window, mode="determinate")
+        self.progress_bar.grid(row=3, column=0, padx=10,
+                               pady=10, sticky='w', columnspan=3)
+
+    def load_proxies_from_url(self):
+        url = self.url_entry.get()
+        self.check_button.config(state='disabled')
+        if not url:
+            self.status_text.set("URL is empty.")
+            self.load_url_button.config(state='normal')
+            return
+
+        try:
+            if url.endswith('.txt'):
+                response = requests.get(url)
+                response.raise_for_status()
+                content = response.text
+            else:
+                feed = feedparser.parse(url)
+                if all('content' in entry for entry in feed.entries):
+                    content = '\n'.join(
+                        entry.content[0].value for entry in feed.entries)
+                else:
+                    content = '\n'.join(
+                        entry.summary for entry in feed.entries if 'summary' in entry)
+
+            proxies = self.extract_proxies_from_text(content)
+
+            if self.proxies:
+                self.check_button.config(state='normal')
+                self.status_text.set(f"Loaded {len(self.proxies)} proxies")
+            else:
+                self.status_text.set("No proxies found.")
+        except Exception as e:
+            self.status_text.set(f"Error: {str(e)}")
+
+        finally:
+            self.load_url_button.config(state='normal')
+
+    def extract_proxies_from_text(self, text):
+        proxies = []
+        lines = text.splitlines()
+        for line in lines:
+            matches = re.findall(r'\b(?:\d{1,3}\.){3}\d{1,3}\:\d{1,5}\b', line)
+            proxies.extend(matches)
+        return proxies
+
+    def load_proxies(self):
+        choice = filedialog.askopenfilename(
+            title="Select a file or cancel to enter proxies manually", filetypes=[("Text files", "*.txt")])
+
+        self.check_button.config(state='disabled')
+        if choice:
+            with open(choice, "r") as file:
+                self.proxies = file.read().splitlines()
+        else:
+            self.proxies = self.window.tk.splitlist(tk.simpledialog.askstring(
+                "Enter proxies", "Enter proxies separated by commas:\nExample: 127.0.0.1:8080,127.0.0.1:8081"))
+
+        if self.proxies:
+            self.check_button.config(state='normal')
+            self.status_text.set(f"Loaded {len(self.proxies)} proxies")
+
+    def check_proxies(self):
+        self.window.after_idle(self.status_text.set, "Checking proxies...")
+
+        self.valid_proxies = self.test_proxies(self.proxies)
+
+        self.output_text.delete(1.0, tk.END)
+        self.output_text.insert(tk.END, "\n".join(self.valid_proxies))
+
+        if self.valid_proxies:
+            self.export_txt_button.config(state='normal')
+            self.export_csv_button.config(state='normal')
+        else:
+            self.export_txt_button.config(state='disabled')
+            self.export_csv_button.config(state='disabled')
+
+        self.window.after_idle(self.check_button.config, state='normal')
+        self.window.after_idle(self.load_button.config, state='normal')
+        self.window.after_idle(self.stop_button.config, state='disabled')
+        self.status_text.set(
+            f"Checked proxies. {len(self.valid_proxies)} valid proxies found.")
+        self.update_status_label()
+
+    def toggle_check_proxies_button(self):
+        if self.find_valid_proxies_var.get():
+            self.check_button.config(state='disabled')
+        else:
+            if self.proxies:
+                self.check_button.config(state='normal')
+            else:
+                self.check_button.config(state='disabled')
+
+    def fetch_and_validate_proxies(self):
+        self.window.after_idle(self.status_text.set,
+                               "Fetching proxies from the internet...")
+        proxy_type = self.proxy_type_var.get()
+        proxy_type = proxy_type.lower() if proxy_type != 'All' else 'http'
+
+        try:
+            url = f'https://api.proxyscrape.com/v2/?request=displayproxies&protocol=http&timeout=10000&country=all&ssl=all&anonymity=all'
+            response = requests.get(url)
+            response.raise_for_status()
+            proxies = response.text.splitlines()
+            self.proxies = proxies
+            self.status_text.set(
+                f"Found {len(proxies)} proxies from the internet. Checking...")
+            self.check_proxies()
+        except Exception as e:
+            # self.status_text.set(f"Error fetching proxies: {str(e)}")
+            self.check_button.config(state='normal')
+            self.load_button.config(state='normal')
+            self.stop_button.config(state='disabled')
+
+    def start_check_proxies(self):
+        self.stop_check.clear()
+        self.check_button.config(state='disabled')
+        self.load_button.config(state='disabled')
+        self.stop_button.config(state='normal')
+        self.stop_check.clear()
+
+        if self.find_valid_proxies_var.get():
+            check_thread = threading.Thread(
+                target=self.fetch_and_validate_proxies)
+        else:
+            check_thread = threading.Thread(target=self.check_proxies)
+
+        check_thread.start()
+
+    def stop_check_proxies(self):
+        self.stop_check.set()
+
+    def is_proxy_valid(self, proxy):
+        if self.stop_check.is_set():
+            return False
+
+        proxy_url = f"http://{proxy}"
+        proxy_type = self.proxy_type_var.get()
+
+        if proxy_type in ('All', 'HTTP', 'HTTPS'):
+            session = requests.Session()
+            session.mount('http://', ProxyAdapter(proxy_url))
+            session.mount('https://', ProxyAdapter(proxy_url))
+            retry_strategy = Retry(
+                total=3,
+                backoff_factor=1,
+                status_forcelist=[429, 500, 502, 503, 504],
+                allowed_methods=["HEAD", "GET", "OPTIONS"]
+            )
+            session.mount("https://", HTTPAdapter(max_retries=retry_strategy))
+
+            try:
+                response = session.get("https://httpbin.org/ip", timeout=5)
+                if response.status_code == 200:
+                    # self.status_text.set(
+                    #     f"Testing: {proxy} - Valid (HTTP/HTTPS)")
+                    return True
+            except Exception:
+                pass
+
+        if proxy_type in ('All', 'SOCKS4', 'SOCKS5'):
+            proxy_ip, proxy_port = proxy.split(":")
+            socks_types = [socks.SOCKS4, socks.SOCKS5] if proxy_type == 'All' else [
+                getattr(socks, proxy_type)]
+
+            for socks_type in socks_types:
+                original_socket = _socket.socket
+                socks.set_default_proxy(socks_type, proxy_ip, int(proxy_port))
+                _socket.socket = socks.socksocket
+                try:
+                    response = urlopen("https://httpbin.org/ip", timeout=5)
+                    if response.status == 200:
+                        # self.status_text.set(
+                        #     f"Testing: {proxy} - Valid (SOCKS{socks_type - socks.SOCKS4 + 4})")
+                        return True
+                except Exception:
+                    pass
+                finally:
+                    _socket.socket = original_socket
+
+        self.progress_queue.put(1)
+        # self.status_text.set(f"Testing: {proxy} - Invalid")
+        return False
+
+    def start_load_proxies_from_url(self):
+        self.load_url_button.config(state='disabled')
+        self.status_text.set("Loading proxies from URL...")
+        load_thread = threading.Thread(target=self.load_proxies_from_url)
+        load_thread.start()
+
+    def export_txt(self):
+        file = filedialog.asksaveasfile(defaultextension=".txt", filetypes=[
+            ("Text files", "*.txt")])
+        if file:
+            file.write('\n'.join(self.valid_proxies))
+            file.close()
+
+    def export_csv(self):
+        file = filedialog.asksaveasfile(
+            defaultextension=".csv", filetypes=[("CSV files", "*.csv")])
+        if file:
+            writer = csv.writer(file)
+            for proxy in self.valid_proxies:
+                writer.writerow([proxy])
+            file.close()
+
+    def get_anonymity_level(self, proxy):
+        try:
+            with requests.get(self.test_url, proxies={"http": proxy, "https": proxy}, timeout=self.timeout) as response:
+                headers = response.headers
+                via_header = headers.get("Via")
+                x_forwarded_for_header = headers.get("X-Forwarded-For")
+
+                if not via_header and not x_forwarded_for_header:
+                    return "elite"
+                elif via_header and x_forwarded_for_header:
+                    return "anonymous"
+                else:
+                    return "transparent"
+        except Exception as e:
+            return None
+
+    def test_proxies(self, proxies):
+        self.progress_bar["maximum"] = len(proxies)
+        self.progress_bar["value"] = 0
+        valid_proxies = []
+        with ThreadPoolExecutor() as executor:
+            for proxy, valid in zip(proxies, executor.map(self.is_proxy_valid, proxies)):
+                if valid:
+                    valid_proxies.append(proxy)
+                    self.output_text.insert(tk.END, proxy + "\n")
+                    # Add this line to auto-scroll
+                    self.output_text.see(tk.END)
+                self.update_progress_bar(1)
+        self.progress_bar["value"] = len(valid_proxies)
+        return valid_proxies
+
+    def run(self):
+        self.window.mainloop()
+
+
+if __name__ == "__main__":
+    app = ProxyChecker()
+    app.run()