import json
import datetime
import requests
import re
import random
import time
from concurrent.futures import ThreadPoolExecutor
from typing import Dict, List, Optional, Set, Tuple, Union
class ProxyFinder:
"""Finds and validates proxies from various online sources"""
def __init__(self, verbose: bool = False):
"""Initialize ProxyFinder with optional verbose logging"""
self.verbose = verbose
self.api: Dict[str, List[str]] = {
'socks4': [
"https://api.proxyscrape.com/?request=displayproxies&proxytype=socks4&timeout=10000&country=all&simplified=true",
"https://www.proxy-list.download/api/v1/get?type=socks4",
"https://api.openproxylist.xyz/socks4.txt",
'https://openproxy.space/list/socks4',
'https://proxyspace.pro/socks4.txt',
"https://sunny9577.github.io/proxy-scraper/generated/socks4_proxies.txt",
'https://cdn.rei.my.id/proxy/SOCKS4',
"https://raw.githubusercontent.com/TheSpeedX/PROXY-List/master/socks4.txt",
"https://raw.githubusercontent.com/roosterkid/openproxylist/main/SOCKS4_RAW.txt",
'https://raw.githubusercontent.com/monosans/proxy-list/main/proxies/socks4.txt'
],
'socks5': [
"https://api.proxyscrape.com/v2/?request=getproxies&protocol=socks5&timeout=10000&country=all&simplified=true",
"https://www.proxy-list.download/api/v1/get?type=socks5",
"https://api.openproxylist.xyz/socks5.txt",
'https://openproxy.space/list/socks5',
'https://spys.me/socks.txt',
'https://proxyspace.pro/socks5.txt',
"https://sunny9577.github.io/proxy-scraper/generated/socks5_proxies.txt",
'https://cdn.rei.my.id/proxy/SOCKS5',
'https://raw.githubusercontent.com/manuGMG/proxy-365/main/SOCKS5.txt',
"https://raw.githubusercontent.com/TheSpeedX/PROXY-List/master/socks5.txt",
"https://raw.githubusercontent.com/hookzof/socks5_list/master/proxy.txt"
],
'http': [
'https://raw.githubusercontent.com/officialputuid/KangProxy/KangProxy/http/http.txt',
"https://github.com/TheSpeedX/PROXY-List/raw/refs/heads/master/http.txt",
"https://api.proxyscrape.com/?request=displayproxies&proxytype=http&timeout=10000&country=all&simplified=true",
"https://www.proxy-list.download/api/v1/get?type=http",
"https://api.openproxylist.xyz/http.txt",
'https://openproxy.space/list/http',
'https://proxyspace.pro/http.txt',
"https://sunny9577.github.io/proxy-scraper/generated/http_proxies.txt",
'https://cdn.rei.my.id/proxy/HTTP',
'https://raw.githubusercontent.com/UptimerBot/proxy-list/master/proxies/http.txt',
'https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/http.txt'
],
'https': [
'https://raw.githubusercontent.com/Firdoxx/proxy-list/main/https',
'https://raw.githubusercontent.com/roosterkid/openproxylist/main/HTTPS_RAW.txt',
'https://raw.githubusercontent.com/aslisk/proxyhttps/main/https.txt',
'https://raw.githubusercontent.com/ErcinDedeoglu/proxies/main/proxies/https.txt',
'https://raw.githubusercontent.com/zloi-user/hideip.me/main/https.txt',
'https://raw.githubusercontent.com/vakhov/fresh-proxy-list/master/https.txt',
'https://raw.githubusercontent.com/Vann-Dev/proxy-list/main/proxies/https.txt'
],
'mixed': [
'https://github.com/jetkai/proxy-list/blob/main/online-proxies/txt/proxies.txt',
'https://raw.githubusercontent.com/mertguvencli/http-proxy-list/main/proxy-list/data.txt',
'https://raw.githubusercontent.com/a2u/free-proxy-list/master/free-proxy-list.txt',
'https://raw.githubusercontent.com/mishakorzik/Free-Proxy/main/proxy.txt',
'http://rootjazz.com/proxies/proxies.txt',
'https://multiproxy.org/txt_all/proxy.txt',
'https://proxy-spider.com/api/proxies.example.txt'
]
}
self.proxy_dict: Dict[str, List[str]] = {'socks4': [], 'socks5': [], 'http': [], 'https': []}
self.max_workers = 20 # Maximum workers for parallel requests
def log(self, *args):
"""Log messages if verbose mode is enabled"""
if self.verbose:
print(*args)
def extract_proxy(self, line: str) -> Optional[str]:
"""
Extracts the first occurrence of an IP:port from a line.
"""
match = re.search(r'(\d{1,3}(?:\.\d{1,3}){3}:\d{2,5})', line)
if match:
return match.group(1)
return None
def fetch_from_url(self, url: str, proxy_type: str) -> List[str]:
"""
Fetches proxies from a given URL for the specified type.
Returns a list of valid proxies.
"""
proxy_list = []
try:
r = requests.get(url, timeout=5)
if r.status_code == requests.codes.ok:
for line in r.text.splitlines():
proxy = self.extract_proxy(line)
if proxy:
proxy_list.append(proxy)
self.log(f"Got {len(proxy_list)} {proxy_type} proxies from {url}")
return proxy_list
except Exception as e:
self.log(f"Failed to retrieve from {url}: {str(e)}")
return []
def fetch_proxies_parallel(self, proxy_type: str) -> List[str]:
"""
Fetch proxies in parallel for a specific type from all sources.
"""
if proxy_type not in self.api:
return []
all_proxies = []
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
futures = [executor.submit(self.fetch_from_url, url, proxy_type)
for url in self.api[proxy_type]]
for future in futures:
all_proxies.extend(future.result())
return list(set(all_proxies)) # Remove duplicates
def get_geonode_proxies(self) -> Dict[str, List[str]]:
"""
Retrieves proxies from geonode API
"""
result = {'http': [], 'socks4': [], 'socks5': [], 'https': []}
try:
url = 'https://proxylist.geonode.com/api/proxy-list?limit=500&sort_by=lastChecked&sort_type=desc'
response = requests.get(url, timeout=10)
if response.status_code == 200:
data = response.json()
for p in data.get('data', []):
for protocol in p.get('protocols', []):
protocol = protocol.lower()
# Map 'https' to 'http' in our dictionary
if protocol == 'https':
result['http'].append(f"{p['ip']}:{p['port']}")
elif protocol in result:
result[protocol].append(f"{p['ip']}:{p['port']}")
self.log(f"Got {sum(len(v) for v in result.values())} proxies from GeoNode")
except Exception as e:
self.log(f"Failed to fetch from GeoNode: {str(e)}")
return result
def get_checkerproxy_archive(self) -> Dict[str, List[str]]:
"""
Gets proxies from checkerproxy.net archive
"""
result = {'http': [], 'socks5': []}
for q in range(5): # Check only last 5 days to be faster
day = datetime.date.today() + datetime.timedelta(days=-q)
formatted_date = f'{day.year}-{day.month}-{day.day}'
try:
r = requests.get(f'https://checkerproxy.net/api/archive/{formatted_date}', timeout=5)
if r.text != '[]':
json_result = json.loads(r.text)
for i in json_result:
# Skip internal IPs
if re.match(r"172\.|192\.168\.|10\.", i['ip']):
continue
addr = i.get('addr')
if not addr:
continue
if i['type'] in [1, 2]:
result['http'].append(addr)
if i['type'] == 4:
result['socks5'].append(addr)
self.log(f"Got {len(result['http'])} http and {len(result['socks5'])} socks5 proxies from CheckerProxy for {formatted_date}")
except Exception as e:
self.log(f"Failed to get archive for {formatted_date}: {str(e)}")
return result
def get_proxies(self, proxy_types: List[str] = None) -> Dict[str, List[str]]:
"""
Get proxies of the specified types. If None, get all types.
Returns a dictionary with proxy lists for each type.
"""
if proxy_types is None:
proxy_types = ['http', 'https', 'socks4', 'socks5']
# Reset proxy dictionary
self.proxy_dict = {'socks4': [], 'socks5': [], 'http': [], 'https': []}
self.log("Starting proxy retrieval process")
# Fetch from regular sources in parallel for each type
for ptype in proxy_types:
if ptype in self.api:
self.log(f"Processing {ptype} proxy sources")
proxies = self.fetch_proxies_parallel(ptype)
self.proxy_dict[ptype].extend(proxies)
# Add proxies from GeoNode
geonode_proxies = self.get_geonode_proxies()
for ptype, proxies in geonode_proxies.items():
if ptype in proxy_types:
self.proxy_dict[ptype].extend(proxies)
# Add proxies from CheckerProxy
checker_proxies = self.get_checkerproxy_archive()
for ptype, proxies in checker_proxies.items():
if ptype in proxy_types:
self.proxy_dict[ptype].extend(proxies)
# Process "mixed" sources if any proxy type is requested
if proxy_types:
self.log("Processing mixed proxy sources")
for url in self.api.get('mixed', []):
try:
proxies = self.fetch_from_url(url, 'mixed')
# Distribute mixed proxies equally among requested types
if proxies:
chunks = len(proxy_types)
chunk_size = len(proxies) // chunks if chunks > 0 else 0
for i, ptype in enumerate(proxy_types):
start = i * chunk_size
end = start + chunk_size if i < chunks - 1 else len(proxies)
self.proxy_dict[ptype].extend(proxies[start:end])
except Exception as e:
self.log(f"Failed to process mixed proxy source: {str(e)}")
# Remove duplicates for all types
for key in self.proxy_dict:
original_count = len(self.proxy_dict[key])
self.proxy_dict[key] = list(set(self.proxy_dict[key]))
new_count = len(self.proxy_dict[key])
self.log(f"Removed {original_count - new_count} duplicate {key} proxies")
self.log("Proxy retrieval process completed")
return self.proxy_dict
def get_random_proxy(self, proxy_type: str = None) -> Optional[str]:
"""
Returns a random proxy of the specified type.
If type is None, returns a random proxy from any type.
"""
if proxy_type and proxy_type in self.proxy_dict and self.proxy_dict[proxy_type]:
return random.choice(self.proxy_dict[proxy_type])
elif not proxy_type:
# Combine all proxy types and get a random one
all_proxies = []
for ptype in self.proxy_dict:
all_proxies.extend(self.proxy_dict[ptype])
if all_proxies:
return random.choice(all_proxies)
return None
def get_random_proxies(self, count: int = 10, proxy_type: str = None) -> List[str]:
"""
Returns a list of random proxies of the specified type.
If type is None, returns random proxies from any type.
"""
if proxy_type and proxy_type in self.proxy_dict:
proxies = self.proxy_dict[proxy_type]
else:
# Combine all proxy types
proxies = []
for ptype in self.proxy_dict:
proxies.extend(self.proxy_dict[ptype])
# Get random proxies up to count or as many as available
if not proxies:
return []
return random.sample(proxies, min(count, len(proxies)))
if __name__ == "__main__":
# Example usage
finder = ProxyFinder(verbose=True)
proxies = finder.get_proxies(['http', 'socks5'])
print("\nSummary:")
for ptype, proxy_list in proxies.items():
print(f"{ptype}: {len(proxy_list)} proxies")
print("\nRandom HTTP proxy:", finder.get_random_proxy('http'))
print("\nRandom SOCKS5 proxies:", finder.get_random_proxies(5, 'socks5'))