Re: Making Python Script Exit Completely with One Ctrl+C

On Fri, Dec 8, 2023 at 9:00 AM Dan Fandrich via curl-users
<[email protected]> wrote:
>
> On Fri, Dec 08, 2023 at 08:39:28AM +0800, Hongyi Zhao via curl-users wrote:
> > The script, named **proxy-speed.py**, is employed to measure and
> > display the speed of different proxies. It runs through a list of
> > proxies, giving the current and average speed for each, one at a time.
> > However, the issue arises when I try to terminate the script
> > prematurely using Ctrl+C. The KeyboardInterrupt appears to only pause
> > the current task, and the script instantaneously proceeds with the
> > next proxy on the list, clearly illustrating that it doesn’t exit
> > entirely.
> >
> > Here’s an example illustrating the issue:
> >
> >
> > ```
> >   File "/home/werner/Desktop/proxy-speed/proxy-speed.py", line 54, in progress
> >     def progress(download_t, download_d, upload_t, upload_d):
> >
> > KeyboardInterrupt
> > Average Speed: 3944.97 kB/s
> >
> >     Proxy: SG_ssr_futeapbquf5m.nodelist.club_1356_8ce27b1301fcbcb77cc5abb28cb127a6
> >     ^CTraceback (most recent call last):
> >       File "/home/werner/Desktop/proxy-speed/proxy-speed.py", line 54,
> > in progress
> >         def progress(download_t, download_d, upload_t, upload_d):
> >
> >     KeyboardInterrupt
> >     Average Speed: 72.81 kB/s
> > ```
>
> The call trace shows the exception being raised within the progress() function.
> One reasonable response to this by pycurl would be to implicitly treat that as a
> progress abort return code, but it looks like it does not do that. Instead, you
> will probably need a try/catch block in that function that returns an error
> code to pycurl. In the C binding this is done by returning a nonzero value from
> that function; it's probably similar in pycurl. This will then return an error
> from the c.perform() function that you can catch in the main loop.
>
> In any case, this is unlikely to be a libcurl problem.

I tried the following method but in vain:

```
import subprocess
import time
import pycurl
from io import BytesIO
import sys

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 a default empty dictionary

    if error is not None:
        return result_dict  # Return the empty dictionary in case of an 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 test_proxy(proxy, url):
    global last_calc_time, download_start_time, kb_interrupt

    # Make sure the flag is reset before testing the next proxy
    kb_interrupt = False

    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. Exiting.")
        sys.exit(0)

    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, kb_interrupt
    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

    try:
        pass
    except KeyboardInterrupt:
        kb_interrupt = True

proxy_data = fetch_proxies()
url = "http://ipv4.download.thinkbroadband.com/1GB.zip"

kb_interrupt = False

for key, value in proxy_data.items():
    print(f"Proxy: {key}")

    if kb_interrupt:
        print("Script terminated by user. Exiting.")
        sys.exit(0)

    try:
        average_speed = test_proxy(value, url)
        print(f"Average Speed: {average_speed:.2f} kB/s")
    except KeyboardInterrupt:
        print("Keyboard interrupt encountered during proxy testing. Exiting.")
        break
```

See the test results below for more information:

(datasci) werner@X10DAi:~/Desktop/proxy-speed$ python temp-test.py
Proxy: SG_ssr_futeapbquf5m.nodelist.club_1453_018d83c677e05e71f172014fc3f45e39
Current Speed: 3118.90 kB/s
Current Speed: 9673.65 kB/s
Average Speed: 11194.13 kB/s
Proxy: HK_ssr_wo8o8npg4fny.nodelist.club_1303_b5bf85111d0f51c517ec7302d3f33ce1
^CTraceback (most recent call last):
  File "/home/werner/Desktop/proxy-speed/temp-test.py", line 57, in progress
    def progress(download_t, download_d, upload_t, upload_d):

KeyboardInterrupt
Average Speed: 84.27 kB/s
Proxy: SG_ssr_futeapbquf5m.nodelist.club_1354_ddfba110eddfdb7037f389e9b5917477
Current Speed: 3319.48 kB/s
Current Speed: 9845.56 kB/s
Average Speed: 11209.70 kB/s
Proxy: HK_ssr_fwqvvo60u1mj.nodelist.club_1423_dba3b0996744e7b82eae7634626a5ba9
^CTraceback (most recent call last):
  File "/home/werner/Desktop/proxy-speed/temp-test.py", line 57, in progress
    def progress(download_t, download_d, upload_t, upload_d):


Regards,
Zhao

>
> > This repeat pattern continues for every Ctrl+C interruption made.
> > Hence, my goal is to modify the script so that a single Ctrl+C command
> > would result in a comprehensive termination of the script.
> >
> > Any guidance or suggestions that you could offer regarding this issue
> > would be greatly appreciated. If there is any further information
> > needed to better understand the issue, please let me know.
> >
> > The content of **proxy-speed.py** is as follows:
> >
> > ```
> > import subprocess
> > import time
> > import pycurl
> > from io import BytesIO
> >
> > 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 test_proxy(proxy, url):
> >     global last_calc_time, download_start_time
> >
> >     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. Exiting.")
> >         sys.exit(0)  # immediately exit the script
> >
> >     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
> >
> > proxy_data = fetch_proxies()
> > url = "http://ipv4.download.thinkbroadband.com/1GB.zip"
> >
> > for key, value in proxy_data.items():
> >     print(f"Proxy: {key}")
> >     try:
> >         average_speed = test_proxy(value, url)
> >         print(f"Average Speed: {average_speed:.2f} kB/s")
> >     except KeyboardInterrupt:
> >         print("Script interrupted by user. Exiting.")
> >         break  # exit the for loop
> > ```
> >
> > See here [1] for the related discussion.
> >
> > Thank you very much for your help in advance.
> >
> > [1] https://discuss.python.org/t/making-python-script-exit-completely-with-one-ctrl-c/40529/1
> >
> > Best Regards,
> > Zhao
> >
> > P.S: If possible, could you please provide both an explanation and
> > snippet of the solution so that I can understand and learn from it for
> > future references.
> > --
> > Assoc. Prof. Hongsheng Zhao <[email protected]>
> > Theory and Simulation of Materials
> > Hebei Vocational University of Technology and Engineering
> > No. 473, Quannan West Street, Xindu District, Xingtai, Hebei province
> --
> Unsubscribe: https://lists.haxx.se/mailman/listinfo/curl-users
> Etiquette:   https://curl.se/mail/etiquette.html



-- 
Assoc. Prof. Hongsheng Zhao <[email protected]>
Theory and Simulation of Materials
Hebei Vocational University of Technology and Engineering
No. 473, Quannan West Street, Xindu District, Xingtai, Hebei province
-- 
Unsubscribe: https://lists.haxx.se/mailman/listinfo/curl-users
Etiquette:   https://curl.se/mail/etiquette.html