-
-
Save AnderRV/e45c0c7f74e89f13c9a609385950bfc4 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import requests | |
import random | |
from threading import Timer | |
def reset_proxy(proxy): | |
print("reset_proxy", proxy) | |
unchecked.add(proxy) | |
working.discard(proxy) | |
not_working.discard(proxy) | |
def set_working(proxy): | |
print("set_working", proxy) | |
unchecked.discard(proxy) | |
working.add(proxy) | |
not_working.discard(proxy) | |
def set_not_working(proxy): | |
print("set_not_working", proxy) | |
unchecked.discard(proxy) | |
working.discard(proxy) | |
not_working.add(proxy) | |
# move to unchecked after a certain time | |
Timer(20.0, reset_proxy, [proxy]).start() | |
def get_random_proxy(): | |
# create a tuple from unchecked and working sets | |
available_proxies = tuple(unchecked.union(working)) | |
if not available_proxies: | |
raise Exception("no proxies available") | |
return random.choice(available_proxies) | |
VALID_STATUSES = [200, 301, 302, 307, 404] | |
session = requests.Session() | |
proxies_list = open("rotating_proxies_list.txt", "r").read().strip().split("\n") | |
unchecked = set(proxies_list[0:10]) # limited to 10 to avoid too many requests | |
# unchecked = set(proxies_list) | |
working = set() | |
not_working = set() | |
def get(url, proxy = None): | |
if not proxy: | |
proxy = get_random_proxy() | |
try: | |
# Send proxy requests to the final URL | |
response = session.get(url, proxies={'http': f"http://{proxy}"}, timeout=30) | |
if response.status_code in VALID_STATUSES: | |
set_working(proxy) | |
else: | |
set_not_working(proxy) | |
return response | |
except Exception as e: | |
set_not_working(proxy) | |
raise e # raise exception | |
def check_proxies(): | |
for proxy in list(unchecked): | |
get("http://ident.me/", proxy) | |
check_proxies() | |
print("unchecked ->", unchecked) # unchecked -> set() | |
print("working ->", working) # working -> {"152.0.209.175:8080", ...} | |
print("not_working ->", not_working) # not_working -> {"167.71.5.83:3128", ...} | |
def main(): | |
result = get("http://ident.me/") | |
print(result.status_code) # 200 | |
print(result.text) # 152.0.209.175 | |
main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment