Skip to content

Improve proxy testing performance with concurrent execution in proxy_… #22

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Improve proxy testing performance with concurrent execution in proxy_… #22

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 22 additions & 18 deletions proxy_tester_menu.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
import time
import requests
import os
from concurrent.futures import ThreadPoolExecutor, as_completed

DEFAULT_DOMAIN = "www.google.com"

Expand All @@ -18,30 +19,33 @@ def test_proxy(ip, domain=DEFAULT_DOMAIN):
return False


def test_ip_addresses(ip_file, output_file, domain=DEFAULT_DOMAIN):
def test_ip_addresses_concurrent(ip_file, output_file, domain=DEFAULT_DOMAIN):
with open(ip_file, "r") as file:
ip_addresses = file.read().splitlines()

results = []

for i, ip in enumerate(ip_addresses, start=1):
is_working = test_proxy(ip, domain=domain)
status = "Active" if is_working else "Inactive"
results.append((ip, status))

progress = i / len(ip_addresses) * 100
print(
f'\rTesting - Progress: {progress:.1f}% | {"." * (i % 4)} ',
end="",
flush=True,
)
time.sleep(0.1) # Add a slight delay to simulate animation

# Using ThreadPoolExecutor to test multiple proxies concurrently for better performance
with ThreadPoolExecutor(max_workers=10) as executor:
# Submit each proxy test to the executor and store the future object along with the IP
future_to_ip = {executor.submit(test_proxy, ip, domain): ip for ip in ip_addresses}
for future in as_completed(future_to_ip):
ip = future_to_ip[future]
try:
# Get the result of the future and determine if the proxy is active or inactive
is_working = future.result()
status = "Active" if is_working else "Inactive"
results.append((ip, status))
except Exception as e:
# Handle any exceptions that occur during the proxy test
print(f"Error testing proxy {ip}: {e}")

# Write the results to the output file
with open(output_file, "w") as file:
for ip, status in results:
file.write(f"{ip}\t{status}\n")

print("\rTesting complete. ")
print("Testing complete.")


def display_menu():
Expand All @@ -68,10 +72,10 @@ def get_input_filename(choice):
def check_input_files():
http_file = "http.txt"
https_file = "https.txt"
if not os.path.exists(http_file) or not os.path.isfile(http_file):
if not os.path.isfile(http_file):
print(f"Error: '{http_file}' file is missing or not found.")
return False
if not os.path.exists(https_file) or not os.path.isfile(https_file):
if not os.path.isfile(https_file):
print(f"Error: '{https_file}' file is missing or not found.")
return False
return True
Expand Down Expand Up @@ -99,7 +103,7 @@ def main():
domain = get_domain_choice()
print(f"Using domain: {domain}")

test_ip_addresses(input_file, output_file, domain=domain)
test_ip_addresses_concurrent(input_file, output_file, domain=domain)

print("Testing complete. Results saved to", output_file)

Expand Down