JimWas-Proxy-Checker.py

#!/usr/bin/env python3
"""
Proxy Checker Script
Checks HTTP and SOCKS5 proxies from a text file and outputs working ones to a new file.
"""

import requests
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.parse import urlparse
import argparse
import sys

class ProxyChecker:
    def __init__(self, input_file, output_file, timeout=10, max_workers=50, test_url="http://httpbin.org/ip"):
        self.input_file = input_file
        self.output_file = output_file
        self.timeout = timeout
        self.max_workers = max_workers
        self.test_url = test_url
        self.working_proxies = []
        self.total_proxies = 0
        self.checked_proxies = 0
        self.lock = threading.Lock()
        
    def load_proxies(self):
        """Load proxies from input file"""
        proxies = []
        try:
            with open(self.input_file, 'r', encoding='utf-8', errors='ignore') as f:
                for line in f:
                    line = line.strip()
                    if line and not line.startswith('#'):
                        # Support different formats:
                        # ip:port
                        # protocol://ip:port
                        # ip:port:username:password
                        if '://' not in line:
                            # Assume HTTP if no protocol specified
                            line = 'http://' + line
                        proxies.append(line)
        except FileNotFoundError:
            print(f"Error: Input file '{self.input_file}' not found.")
            sys.exit(1)
        except Exception as e:
            print(f"Error reading input file: {e}")
            sys.exit(1)
            
        return proxies
    
    def check_proxy(self, proxy_url):
        """Check if a single proxy is working"""
        try:
            parsed = urlparse(proxy_url)
            
            # Determine proxy type and format
            if parsed.scheme.lower() in ['socks5', 'socks5h']:
                proxy_dict = {
                    'http': proxy_url,
                    'https': proxy_url
                }
            elif parsed.scheme.lower() in ['http', 'https']:
                proxy_dict = {
                    'http': proxy_url,
                    'https': proxy_url
                }
            else:
                # Default to HTTP
                proxy_dict = {
                    'http': proxy_url,
                    'https': proxy_url
                }
            
            # Make test request
            response = requests.get(
                self.test_url,
                proxies=proxy_dict,
                timeout=self.timeout,
                headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}
            )
            
            if response.status_code == 200:
                # Additional check: verify we're actually using the proxy
                # by checking if the response contains different IP than our real IP
                return True, proxy_url, response.elapsed.total_seconds()
            else:
                return False, proxy_url, None
                
        except requests.exceptions.ProxyError:
            return False, proxy_url, "Proxy Error"
        except requests.exceptions.ConnectTimeout:
            return False, proxy_url, "Connection Timeout"
        except requests.exceptions.ReadTimeout:
            return False, proxy_url, "Read Timeout"
        except requests.exceptions.ConnectionError:
            return False, proxy_url, "Connection Error"
        except Exception as e:
            return False, proxy_url, str(e)
    
    def extract_ip_port(self, proxy_url):
        """Extract just IP:PORT from proxy URL"""
        try:
            parsed = urlparse(proxy_url)
            if parsed.hostname and parsed.port:
                return f"{parsed.hostname}:{parsed.port}"
            else:
                # Handle cases where URL parsing fails
                # Try to extract IP:PORT manually
                clean_url = proxy_url.replace('http://', '').replace('https://', '').replace('socks5://', '').replace('socks5h://', '')
                if '@' in clean_url:
                    # Remove username:password@ part
                    clean_url = clean_url.split('@')[1]
                # Take only the IP:PORT part (before any path)
                ip_port = clean_url.split('/')[0]
                return ip_port
        except Exception:
            return proxy_url
    
    def update_progress(self, working_proxy=None):
        """Update and display progress"""
        with self.lock:
            self.checked_proxies += 1
            progress = (self.checked_proxies / self.total_proxies) * 100
            working_count = len(self.working_proxies)
            
            if working_proxy:
                proxy_url, response_time = working_proxy
                ip_port = self.extract_ip_port(proxy_url)
                # Clear the current line and show the working proxy
                print(f"\r{' ' * 80}", end='')  # Clear line
                print(f"\r✓ WORKING: {ip_port} (Response: {response_time:.2f}s)")
                # Also save to file immediately
                self.save_single_proxy(ip_port, response_time)
            
            # Show progress on the same line
            print(f"\rProgress: {progress:.1f}% ({self.checked_proxies}/{self.total_proxies}) | Working: {working_count}", end='', flush=True)
    
    def save_single_proxy(self, proxy_url, response_time):
        """Save a single working proxy to the output file immediately"""
        try:
            # Check if this is the first proxy being written
            file_exists = False
            try:
                with open(self.output_file, 'r') as f:
                    file_exists = True
            except FileNotFoundError:
                pass
            
            # Open in append mode
            with open(self.output_file, 'a', encoding='utf-8') as f:

                # Write the working proxy
                f.write(f"{proxy_url}\n")
                f.flush()  # Ensure immediate write to disk
                
        except Exception as e:
            print(f"\nError saving proxy to file: {e}")
    
    def finalize_output_file(self):
        """Add final statistics to the output file"""
        try:
            with open(self.output_file, 'a', encoding='utf-8') as f:
                f.write(f"\n# Scanning completed on {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
                f.write(f"# Total working proxies: {len(self.working_proxies)}\n")
                    
        except Exception as e:
            print(f"\nError finalizing output file: {e}")
    
    def run(self):
        """Main execution method"""
        print(f"Loading proxies from '{self.input_file}'...")
        proxies = self.load_proxies()
        self.total_proxies = len(proxies)
        
        if self.total_proxies == 0:
            print("No proxies found in input file.")
            return
            
        print(f"Found {self.total_proxies} proxies to check.")
        print(f"Using {self.max_workers} threads with {self.timeout}s timeout.")
        print("Testing against:", self.test_url)
        print(f"Working proxies will be saved to: '{self.output_file}'")
        print("-" * 70)
        print("✓ Working proxies will be displayed and saved in real-time:")
        print("-" * 70)
        
        start_time = time.time()
        
        # Check proxies using thread pool
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            # Submit all tasks
            future_to_proxy = {executor.submit(self.check_proxy, proxy): proxy for proxy in proxies}
            
            # Process completed tasks
            for future in as_completed(future_to_proxy):
                is_working, proxy_url, result = future.result()
                
                if is_working:
                    with self.lock:
                        self.working_proxies.append((proxy_url, result))
                    # Pass the working proxy info to update_progress
                    self.update_progress(working_proxy=(proxy_url, result))
                else:
                    self.update_progress()
        
        elapsed_time = time.time() - start_time
        
        # Display final results
        print(f"\n\n" + "="*70)
        print("SCAN COMPLETED")
        print("="*70)
        print(f"Time elapsed: {elapsed_time:.2f} seconds")
        print(f"Total proxies checked: {self.total_proxies}")
        print(f"Working proxies found: {len(self.working_proxies)}")
        print(f"Success rate: {(len(self.working_proxies)/self.total_proxies)*100:.1f}%")
        
        # Finalize the output file
        if self.working_proxies:
            self.finalize_output_file()
            print(f"All working proxies have been saved to '{self.output_file}'")
        else:
            print("No working proxies found.")


def main():
    parser = argparse.ArgumentParser(
        description="Check HTTP and SOCKS5 proxies from a text file",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python proxy_checker.py proxies.txt working_proxies.txt
  python proxy_checker.py -i proxies.txt -o working.txt -t 15 -w 100
  
Input file format (one proxy per line):
  192.168.1.1:8080
  http://192.168.1.1:8080
  socks5://192.168.1.1:1080
  192.168.1.1:8080:username:password
        """
    )
    
    parser.add_argument('input_file', nargs='?', default='proxies.txt',
                       help='Input file containing proxies (default: proxies.txt)')
    parser.add_argument('output_file', nargs='?', default='working_proxies.txt',
                       help='Output file for working proxies (default: working_proxies.txt)')
    parser.add_argument('-t', '--timeout', type=int, default=10,
                       help='Timeout in seconds for each proxy test (default: 10)')
    parser.add_argument('-w', '--workers', type=int, default=50,
                       help='Number of concurrent threads (default: 50)')
    parser.add_argument('-u', '--url', default='http://httpbin.org/ip',
                       help='Test URL (default: http://httpbin.org/ip)')
    
    args = parser.parse_args()
    
    # Create and run proxy checker
    checker = ProxyChecker(
        input_file=args.input_file,
        output_file=args.output_file,
        timeout=args.timeout,
        max_workers=args.workers,
        test_url=args.url
    )
    
    try:
        checker.run()
    except KeyboardInterrupt:
        print("\n\nOperation cancelled by user.")
        sys.exit(1)


if __name__ == "__main__":
    main()