README.md

ProxPy

This is a work-in-progress python module that serves as a wrapper around requests with proxy support

I do not hold any responsibility for any actions taken using this module

---

ProxPy allows you to effortlessly use multiple proxies imported from a text file without having to deal with any configuration.

Multithreading is fully supported with automatic proxy assignment to each thread, ban/dead detection, multiple workers per proxy and recycling after all proxies have been marked as dead or banned.

Currently supported proxies:

• http
• https
• socks5 (socks5h)

Variables:

class ProxyType(Enum):
    """
    enum specifying type of proxies being used
    """
    
    socks5h = 0
    socks5 = 1
    https = 2
    http = 3

class Judge(Enum):
    """
    Proxy judge enum
    """
    
    azenv = "https://azenv.net"
    proxyjudge = "http://proxyjudge.info"

class Options:
    """
    options holder, you can set all the possible things here
    """

    def __init__(self, username="", password="", judge=Judge.azenv, proxy_check_threads=50,
                 check_timeout=4,
                 debug=False,
                 show_progress=True,
                 workers_per_proxy=5):
        self.judge = judge
        self.proxy_check_threads = proxy_check_threads
        self.check_timeout = check_timeout
        self.debug = debug
        self.username = username
        self.password = password
        self.show_progress = show_progress
        self.workers_per_proxy = workers_per_proxy

class Proxy:
    """
    class for one proxy, holds all workers bound to it, banned/dead status, raw address in a string and address formatted according to requests module
    """

    def __init__(self, proxy_address, proxy_type: ProxyType = ProxyType.https, user=None, password=None):
        self.proxy = proxy_address
        self.dead = False
        self.banned = False
        self.workers = []
        self.dict_proxy = {
            "http": f"{proxy_type.name}://{user}:{password}@{proxy_address}",
            "https": f"{proxy_type.name}://{user}:{password}@{proxy_address}",
        }
        self.type = proxy_type

class ProxyList:
    """
    Holder of proxy lists for all supported proxy types
    """
    def __init__(self):
        self.size = 0
        self.dict = {
            ProxyType.http: [],
            ProxyType.https: [],
            ProxyType.socks5: [],
            ProxyType.socks5h: [],
        }

    #
    # ---**METHODS**---
    #
    def add_proxy(self, proxy: str, proxy_type: ProxyType = ProxyType.https):
        """
        add a proxy from string to the list
        :param proxy: proxy in string format
        :param proxy_type: type of proxy
        :return: None
        """
        self.dict[proxy_type].append(Proxy(proxy, proxy_type, opts.username, opts.password))
        self.size += 1

    def add_proxy_class(self, proxy: Proxy):
        """
        add a proxy as an instance of Proxy class
        :param proxy: proxy to be added, must have Proxy.type set!
        :return: None
        """
        self.dict[proxy.type].append(proxy)
        self.size += 1

    def clear_proxies(self, proxy_type: ProxyType = ProxyType.https):
        """
        clears a proxy list of specific type
        :param proxy_type: type of proxy list to clear
        :return: None 
        """
        self.size -= len(self.dict[proxy_type])
        self.dict[proxy_type] = []

    def update_size(self):
        """
        recalculate the number of proxies in total (after removing, adding without using the add_proxy method)
        :return: None
        """
        self.size = 0
        for plist in self.dict.values():
            self.size += len(plist)

# Internal variables:
# these are used by all the functions, can be accessed freely, should not be modified directly

proxy_list = ProxyList()
opts = Options()

_root = tk.Tk()
_resetting = False
_checked = []
_host_ip = ""

proxy_stats = {
    "suc": 0,
    "fail": 0,
    "alive": 0,
    "dead": 0
}

Methods:

All with simplified functionality

def prequest(method, url, **kwargs):
    """
    Sends a request to url with method method, using proxy bound to current thread
    :param method: 'post' / 'get' / 'put' / ...
    :param url: target url
    :param kwargs: additional arguments (headers, payload, ...), if session is specified, it will be used to send the request
    :return: requests.request
    """
    return request(method, url, **kwargs)

def ppost(url, params=None, **kwargs):
    """
    send POST request to a url with a proxy bound to current thread
    :param url: target url
    :param params: request parameters
    :param kwargs: additional arguments (headers, payload, ...), if session is specified, it will be used to send the request
    :return: requests.models.Response
    """
    return prequest("post", url, params=params, **kwargs)

def pget(url, params=None, **kwargs):
    """
    send GET request to a url with a proxy bound to current thread
    :param url: target url
    :param params: request parameters
    :param kwargs: additional arguments (headers, payload, ...), if session is specified, it will be used to send the request
    :return: requests.models.Response
    """
    return prequest("get", url, params=params, **kwargs)

def get_external_ip(p: Proxy = None):
    """
    determines the current external ip being used to connect to the internet
    :param p: proxy to connect through
    :return: None if connection error, else ipv4 in string
    """
    return ipv4 if connection.ok else None

def check_proxy(prx: str):
    """
    checks a proxy in string format [0-255].[0-255].[0-255].[0-255]:[0-65535]
    :param prx: proxy as a string
    :return: True if valid format else False
    """
    return True of proxy.valid_format else False

def get_new_proxy(dead=False, banned=False):
    """
    binds a new proxy from the internal proxy list to the current thread while possibly marking the old one as banned or dead
    :param dead: current proxy dead?
    :param banned: current proxy banned?
    :return: None
    """
    this_thread.bind_new_proxy(get_proxy_from_list())

def test_proxy(prx: Proxy):
    """
    checks if proxy is working and anonymous
    :param prx: proxy: Proxy class
    :return: Proxy if working else False
    """
    return prx if prx.working else False

def parse_proxies(proxy_file, proxy_type: ProxyType = ProxyType.https):
    """
    parses proxies from a file
    :param proxy_type: type of proxies to be parsed
    :param proxy_file: VALID path to a text file containing proxies
    :return: None
    """
    for valid_proxy in proxy_file:
        proxies.append(proxy)

def check_proxies():
    """
    checks all the parsed proxies - all the types
    :return: None
    """
    for proxy in proxies:
        if proxy.not_working:
            proxies.remove(proxy)

def reset_proxies():
    """
    resets all the proxies, locks other threads from doing so as well
    :return: None
    """
    for proxy in proxies:
        proxy.banned, proxy.dead = False, False

def get_proxy():
    """
    returns the first available proxy from proxies list, resets if all banned
    :return: Proxy class or None if proxy list is empty
    """
    for proxy in proxies:
        if proxy.available:
            return proxy

def update_options(options):
    """
    changes internal options to provided ones
    :param options: Options class with all options set
    :return: None
    """
    global opts
    opts = options

def open_proxy_file(proxy_type: ProxyType = ProxyType.https):
    """
    opens a file dialog to choose the proxy file, parses it afterwards
    :return: None
    """
    parse_proxies(open_file_dialog())

Please note:

As this is an early release, reverse compatibility will most likely NOT be present at the moment

Some things may crash / behave oddly - if so, please report them on github