Gold Proxy Checker (Python script)

Code:

from Queue import Queue
from threading import Thread

import argparse
import requests
import sys
import time

def process_proxy():
    try:
        while True:
            proxy = queue.get()
            if proxy == "STOP":
                return
            
            save_valid_proxy(check_proxy(proxy))
            queue.task_done()

    except:
        pass

def check_proxy(proxy, timeout = 5, url = "https://google.com"):
    proxies = {
            "http": "http://" + proxy,
            "https": "https://" + proxy
    }
    try:
        r = requests.get(url, proxies = proxies, timeout = timeout)

    except IOError:
        return False
                
    return proxy

def save_valid_proxy(proxy):
    if proxy:
        OUT_F.write(proxy + "\n")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description = "Check a proxy list for working proxies.")
    parser.add_argument("infile", help = "The proxy list file.")
    parser.add_argument("outfile", help = "The output file.")
    parser.add_argument("-t", "--threads", type = int, help = "Set the number of threads running concurrently.")
    parser.add_argument("-v", "--verbose", action = "store_true", help = "Be verbose?")
    args = parser.parse_args()

    IN_F = open(args.infile, "r")
    OUT_F = open(args.outfile, "w")
    # number of threads running at once
    if args.threads:
        concurrent_threads = args.threads

    else:
        concurrent_threads = 250

    if args.verbose:
        print("Loading proxy list from: {}".format(args.infile))
        print("Saving valid proxies list to: {}".format(args.outfile))
        print("Running: {} threads...".format(concurrent_threads))
    
    queue = Queue(concurrent_threads * 2)

    for i in range(0, concurrent_threads):
        thread = Thread(target = process_proxy)
        thread.setDaemon(True)
        thread.start()

    start = time.time()
    try:
        for proxy in IN_F:
            queue.put(proxy.strip())

        # queue.join()

    except KeyboardInterrupt:
        pass

    print("Closing down, please wait. ({} seconds run time)".format(time.time() - start))
    queue.put("STOP")
        
    IN_F.close()
    OUT_F.close()
    sys.exit()

Code:

'''
    Proxy Checker v2
        -=m0dem=-

    Requirements:
        requests (HTTP library)
'''

from Queue import Queue
from threading import Thread

import argparse
import requests
import sys
import time


def process_proxy():
    try:
        while True:
            proxy = queue.get()
            if proxy == "STOP":
                return
            
            save_valid_proxy(check_proxy(proxy))
            queue.task_done()

    except:
        pass

def check_proxy(proxy, timeout = 5):
    proxies = {
            "http": "http://" + proxy,
            "https": "https://" + proxy
    }
    try:
        # see if the proxy actually works
        ip = get_external_ip(proxies = proxies)           
        if ip == ORIG_IP:
            if VERBOSE:
                print(ip)
                
            return False

    except IOError:
        return False
                
    return proxy

def save_valid_proxy(proxy):
    if proxy:
        OUT_F.write(proxy + "\n")

def get_external_ip(proxies = None):
    headers = {"User-Agent": "Mozilla/5.0"}
    if proxies:
        r = requests.get(IP_CHECK, proxies = proxies, headers = headers)

    else:
        r = requests.get(IP_CHECK, headers = headers)
        
    return str(r.text)


DEFAULT_THREADS = 200
IP_CHECK = "https://api.ipify.org"
IN_F = None
OUT_F = None
VERBOSE = False
ORIG_IP = None


if __name__ == "__main__":
    # handle all the command-line argument stuff
    parser = argparse.ArgumentParser(description = "Check a proxy list for working proxies.")
    parser.add_argument("infile", type = str, help = "input proxy list file")
    parser.add_argument("outfile", type = str, help = "output proxy list file")
    parser.add_argument("-t", "--threads", type = int, default = DEFAULT_THREADS, help = "set the number of threads running concurrently (default {})".format(DEFAULT_THREADS))
    parser.add_argument("-v", "--verbose", action = "store_true", help = "say lots of useless stuff (sometimes)")
    args = parser.parse_args()

    try:
        IN_F = open(args.infile, "r")

    except IOError:
        print("Invalid proxy list filename.")
        sys.exit()

    OUT_F = open(args.outfile, "w")
        
    # number of threads running at once
    CONCURRENT_THREADS = args.threads

    if args.verbose:
        VERBOSE = True
        print("Loading proxy list from: {}".format(args.infile))
        print("Saving valid proxies list to: {}".format(args.outfile))
        print("Running: {} threads...".format(CONCURRENT_THREADS))
        
    # let's begin the whole process
    ORIG_IP = get_external_ip()
    if VERBOSE:
        print("Your original external IP address is: {}".format(ORIG_IP))
        print("Checking proxies...")
    
    queue = Queue(CONCURRENT_THREADS * 2)

    for i in range(0, CONCURRENT_THREADS):
        thread = Thread(target = process_proxy)
        thread.daemon = True
        thread.start()

    start = time.time()
    try:
        for proxy in IN_F:
            queue.put(proxy.strip())

        # queue.join()

    except KeyboardInterrupt:
        pass

    # make sure everything is closed down
    print("Closing down, please wait. ({} seconds run time)".format(time.time() - start))
    queue.put("STOP")

    IN_F.close()
    OUT_F.close()
    sys.exit()