Thank you very much for your demonstration code snippet. I finally came up with the following working method:
$ cat working-version-1.py
import time
import sys
import pycurl
from io import BytesIO
from blessed import Terminal
import subprocess
last_calc_time = None
download_start_time = None
interrupted = False
#def fetch_proxies():
# command = 'echo "show stat" | sudo socat stdio /var/run/haproxy.sock 2>/dev/null | awk -F, \'$1=="socks5" && !($2~/^(FRONTEND|BACKEND)$/) {print $2,$74}\''
# process = subprocess.Popen(command, stdout=subprocess.PIPE, shell=True)
# output, error = process.communicate()
# result_dict = {} # Initialize default empty dictionary
# if error is not None:
# return result_dict # Return the empty dictionary in case of error
# lines = output.decode('utf-8').split('\n')
# for line in lines:
# if line != "":
# key_value = line.split(' ')
# result_dict[key_value[0]] = key_value[1]
# return result_dict
def fetch_proxies():
# Starting a subprocess with specified arguments
process = subprocess.run(['sudo', 'socat', 'stdio', '/var/run/haproxy.sock'],
input="show stat\n".encode(),
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=True)
# Getting the output
output = process.stdout
# Initializing an empty dictionary to store the results
result_dict = {}
# Decoding and splitting the output into lines
lines = output.decode().split('\n')
# Parsing each line
for line in lines:
parts = line.split(',')
if len(parts) > 1 and parts[0] == 'socks5' and parts[1] not in ['FRONTEND', 'BACKEND']:
# Assuming the desired information is at index 73
result_dict[parts[1]] = parts[73]
return result_dict
def test_proxy(proxy, url):
global last_calc_time, download_start_time, interrupted
buffer = BytesIO()
c = pycurl.Curl()
c.setopt(pycurl.URL, url)
c.setopt(pycurl.WRITEDATA, buffer)
c.setopt(pycurl.PROXY, proxy)
c.setopt(pycurl.PROXYTYPE, pycurl.PROXYTYPE_SOCKS5_HOSTNAME)
c.setopt(pycurl.NOPROGRESS, False)
c.setopt(pycurl.XFERINFOFUNCTION, progress)
c.setopt(pycurl.TIMEOUT, 5)
download_start_time = time.time()
last_calc_time = download_start_time
try:
c.perform()
except pycurl.error as e:
pass
except KeyboardInterrupt:
print("Script interrupted by user.")
interrupted = True # set the flag to True when interrupted
average_speed = c.getinfo(pycurl.SPEED_DOWNLOAD) / 1024
return average_speed
def progress(download_t, download_d, upload_t, upload_d):
global last_calc_time, download_start_time
current_time = time.time()
if current_time - last_calc_time >= 2:
elapsed_time = current_time - download_start_time
current_speed = download_d / elapsed_time / 1024
print(f"Current Speed: {current_speed:.2f} kB/s")
last_calc_time = current_time
def listener(terminal):
with terminal.cbreak():
key = terminal.inkey(timeout=0.05)
if key == '\x1b':
sys.exit("Esc key detected!")
proxy_data = fetch_proxies()
url = "http://ipv4.download.thinkbroadband.com/1GB.zip"
term = Terminal()
for key, value in proxy_data.items():
if interrupted:
sys.exit("Exiting due to interrupt.")
print(f"Proxy: {key}")
listener(term) # Listen for escape key press after each proxy test
try:
average_speed = test_proxy(value, url)
print(f"Average Speed: {average_speed:.2f} kB/s")
except KeyboardInterrupt:
sys.exit("Exiting due to interrupt.")
The test results are as follows:
(datasci) werner@X10DAi:~/Desktop/proxy-speed$ python working-version-1.py
Proxy: SG_ssr_futeapbquf5m.nodelist.club_1453_018d83c677e05e71f172014fc3f45e39
Current Speed: 671.90 kB/s
^[Current Speed: 7220.37 kB/s
Average Speed: 8817.69 kB/s
Proxy: HK_ssr_wo8o8npg4fny.nodelist.club_1303_b5bf85111d0f51c517ec7302d3f33ce1
Esc key detected!