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.
- http
- https
- socks5 (socks5h)
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
}
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())
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