Complete code refactoring

main.py

@@ -7,103 +7,110 @@ import sys
 import subprocess
 import json
 
-PROXIES_LIST = "https://nnp.nnchan.ru/mahoproxy.php?u=https://api.sandvpn.com/fetch-free-proxys"
+PROXIES_LIST_URL = "https://nnp.nnchan.ru/mahoproxy.php?u=https://api.sandvpn.com/fetch-free-proxys"
 SPEEDTEST_URL = "http://212.183.159.230/5MB.zip"
 
-def get_proxies():
+def fetch_proxies():
-    request = requests.get(PROXIES_LIST)
+    """Fetch the list of proxies from the defined URL."""
-    if request.status_code == 200:
+    response = requests.get(PROXIES_LIST_URL)
-        return request.json()
+    response.raise_for_status()
-    else:
+    return response.json()
-        request.raise_for_status()
 
+def is_valid_proxy(proxy):
+    """Check if the proxy is valid and not from Russia (youtube is slowed down there)."""
+    return proxy["host"] and proxy["country"] != "Russia"
 
-def get_best_proxy(proxy_json):
+def construct_proxy_string(proxy):
-    proxy_times = []
+    """Construct a proxy string from the proxy dictionary."""
-    for i, item in enumerate(proxy_json):
+    if proxy.get("username"):
-        if item["host"] != "" and item["country"] != "Russia":
+        return f'{proxy["username"]}:{proxy["password"]}@{proxy["host"]}:{proxy["port"]}'
-            proxy_str = ""
+    return f'{proxy["host"]}:{proxy["port"]}'
-            print(f'Testing {item["host"]}:{item["port"]}')
-            if item["username"]:
-                proxy_str = f'{item["username"]}:{item["password"]}@{item["host"]}:{item["port"]}'
-            else:
-                proxy_str = f'{item["host"]}:{item["port"]}'
-            with io.BytesIO() as f:
-                start = time.perf_counter()
-                try:
-                    r = requests.get(SPEEDTEST_URL, stream=True, proxies={"http": proxy_str}, timeout=20)
-                except:
-                    print("Proxy is dead, skiping...")
-                    del proxy_json[i]
-                    continue
-                total_length = r.headers.get('content-length')
-                dl = 0
-                if total_length is None: # no content length header
-                    print("no content")
-                    f.write(r.content)
-                elif int(total_length) == 5242880:
-                    for chunk in r.iter_content(1024):
-                        dl += len(chunk)
-                        f.write(chunk)
-                        done = int(30 * dl / int(total_length))
-                        if done > 3 and (dl//(time.perf_counter() - start) / 100000) < 1.0:
-                            print("\nProxy is too slow, skiping...")
-                            start = 500
-                            break 
-                        sys.stdout.write("\r[%s%s] %s Mbps" % ('=' * done, ' ' * (30-done), dl//(time.perf_counter() -
-                        start) / 100000))
-                else:
-                    del proxy_json[i]
-                    continue
-                item.update({"time": round(time.perf_counter() - start, 2)})
-                proxy_times.append(item)
-                print("\n")
 
-        else:
+def test_proxy(proxy):
-            del proxy_json[i]
+    """Test the proxy by measuring the download time."""
+    proxy_str = construct_proxy_string(proxy)
+    print(f'Testing {proxy_str}')
 
-    best_five_proxies = sorted(proxy_times, key=lambda x: x['time'])[:5]
+    start_time = time.perf_counter()
-    return best_five_proxies
+    try:
+        response = requests.get(SPEEDTEST_URL, stream=True, proxies={"http": proxy_str}, timeout=5)
+        response.raise_for_status()  # Ensure we raise an error for bad responses
+
+        total_length = response.headers.get('content-length')
+        if total_length is None or int(total_length) != 5242880:
+            print("No content or unexpected content size.")
+            return None
+
+        with io.BytesIO() as f:
+            download_time, downloaded_bytes = download_with_progress(response, f, total_length, start_time)
+            return {"time": download_time, **proxy}  # Include original proxy info
+    except requests.RequestException:
+        print("Proxy is dead, skipping...")
+        return None
+
+def download_with_progress(response, f, total_length, start_time):
+    """Download content from the response with progress tracking."""
+    downloaded_bytes = 0
+    for chunk in response.iter_content(1024):
+        downloaded_bytes += len(chunk)
+        f.write(chunk)
+        done = int(30 * downloaded_bytes / int(total_length))
+        if done > 3 and (downloaded_bytes//(time.perf_counter() - start_time) / 100000) < 1.0:
+            print("\nProxy is too slow, skipping...")
+            return float('inf'), downloaded_bytes
+        sys.stdout.write("\r[%s%s] %s Mbps" % ('=' * done, ' ' * (30-done), downloaded_bytes//(time.perf_counter() -
+        start_time) / 100000))
+    sys.stdout.write("\n")
+    return round(time.perf_counter() - start_time, 2), downloaded_bytes
+
+def get_best_proxies(proxies):
+    """Return the top five proxies based on speed."""
+    proxy_times = [test_proxy(proxy) for proxy in proxies if is_valid_proxy(proxy)]
+    return sorted(filter(None, proxy_times), key=lambda x: x['time'])[:5]
+
+def save_proxies_to_file(proxies, filename="proxy.json"):
+    """Save the best proxies to a JSON file."""
+    with open(os.path.join(os.path.dirname(__file__), filename), "w") as f:
+        json.dump(proxies, f, indent=4)
 
 def update_proxies():
-  proxy_json = get_proxies()
+    """Update the proxies list and save the best ones."""
-  best_proxy_json = get_best_proxy(proxy_json)
+    proxies = fetch_proxies()
-  with open(os.path.abspath(__file__).split("main.py")[0]+"proxy.json", "w") as f:
+    best_proxies = get_best_proxies(proxies)
-    f.write(json.dumps(best_proxy_json, indent=4))
+    save_proxies_to_file(best_proxies)
-  print("All done.")
+    print("All done.")
-
 
 def run_yt_dlp():
-  proxy_url = ""
+    """Run yt-dlp with a randomly selected proxy."""
-  error =  True
+    while True:
-  while error:
+        with open("proxy.json", "r") as f:
-    with open(os.path.abspath(__file__).split("main.py")[0]+"proxy.json", "r") as f:
+            proxy = random.choice(json.load(f))
-      proxy_dict = random.choice(json.load(f))
+            proxy_str = construct_proxy_string(proxy)
-      print(f"Using proxy based in {proxy_dict['city']}, {proxy_dict['country']}")
+            print(f"Using proxy from {proxy['city']}, {proxy['country']}")
-      if proxy_dict["username"]:
-        proxy_url = f'{proxy_dict["username"]}:{proxy_dict["password"]}@{proxy_dict["host"]}:{proxy_dict["port"]}'
-      else:
-        proxy_url = f'{proxy_dict["host"]}:{proxy_dict["port"]}'
-    subprocess.run(f"yt-dlp --color always --proxy  '{proxy_url}' {" ".join([str(arg) for arg in sys.argv])} 2>&1 | tee tempout", shell=True)
-    with open("tempout", 'r') as log_fl:
-      if 'Sign in to' in  log_fl.readlines()[-1]:
-          error = True
-          print("Got 'Sign in to confirm' error. Trying again with another proxy...")
-      else:
-          error = False
-  os.remove("tempout") 
 
+            if execute_yt_dlp_command(proxy_str):
+                break  # Exit loop if command was successful
+            print("Got 'Sign in to confirm' error. Trying again with another proxy...")
+
+def execute_yt_dlp_command(proxy_str):
+    """Execute the yt-dlp command with the given proxy."""
+    command = f"yt-dlp --color always --proxy  '{proxy_str}' {" ".join([str(arg) for arg in sys.argv])} 2>&1 | tee tempout"
+    subprocess.run(command, shell=True)
+    with open("tempout", 'r') as log_fl:
+        return 'Sign in to' not in log_fl.read()
 
 def main():
-  try:
+    """Main function to handle script arguments and execute the appropriate command."""
-    if "update" in sys.argv:
+    try:
-      update_proxies()
+        if "update" in sys.argv:
-    elif len(sys.argv) == 1:
+            update_proxies()
-      print("usage: main.py update | <yt-dlp args> \n\nScript for starting yt-dlp with best free proxy\n\nCommands:\n update   Update best proxy\n")
+        elif len(sys.argv) < 2:
-    else:
+            print("usage: main.py update | <yt-dlp args> \nScript for starting yt-dlp with best free proxy\nCommands:\n update   Update best proxy")
-      sys.argv.pop(0)
+        else:
-      run_yt_dlp()
+            sys.argv.pop(0)
-  except KeyboardInterrupt:
+            run_yt_dlp()
-    print("Canceled by user")
+    except KeyboardInterrupt:
+        print("Canceled by user")
 
-main()
+if __name__ == "__main__":
+    main()

proxy.json

@@ -1,29 +1,6 @@
 [
     {
-        "city": "Chicago 2",
+        "time": 7.21,
-        "country": "USA",
-        "host": "107.152.38.99",
-        "new": false,
-        "paid": false,
-        "password": "L5KT7rSkCrxn8mm4HngTbK8f4k",
-        "port": "3128",
-        "premium": true,
-        "username": "chicago_squid_user",
-        "time": 15.72
-    },
-    {
-        "city": "New York 2",
-        "country": "USA",
-        "host": "107.152.44.132",
-        "new": false,
-        "paid": false,
-        "password": "Xs6QjgCMeiGsX68KMdjnRJbmLmD",
-        "port": "3128",
-        "premium": true,
-        "username": "nc_squid_user",
-        "time": 17.03
-    },
-    {
         "city": "New York 1",
         "country": "USA",
         "host": "107.152.43.179",
@@ -32,31 +9,54 @@
         "password": "pJTg94JCukVjz2b",
         "port": "3128",
         "premium": true,
-        "username": "ny_premium",
+        "username": "ny_premium"
-        "time": 20.92
     },
     {
-        "city": "Hesse",
+        "time": 7.99,
-        "country": "Germany",
+        "city": "Chicago 2",
-        "host": "162.19.230.101",
+        "country": "USA",
+        "host": "107.152.38.99",
         "new": false,
         "paid": false,
-        "password": "enfo8BM3YDbGfyANYJRCKyL7hGrH",
+        "password": "L5KT7rSkCrxn8mm4HngTbK8f4k",
         "port": "3128",
-        "premium": false,
+        "premium": true,
-        "username": "hesse_squid_user",
+        "username": "chicago_squid_user"
-        "time": 62594.22
     },
     {
-        "city": "London",
+        "time": 12.86,
-        "country": "UK",
+        "city": "New York 2",
-        "host": "83.147.17.103",
+        "country": "USA",
+        "host": "107.152.44.132",
         "new": false,
         "paid": false,
-        "password": "dsRLomjPDynq7mJQmhKX7QJCLid",
+        "password": "Xs6QjgCMeiGsX68KMdjnRJbmLmD",
+        "port": "3128",
+        "premium": true,
+        "username": "nc_squid_user"
+    },
+    {
+        "time": 13.09,
+        "city": "Los Angeles",
+        "country": "USA",
+        "host": "107.152.47.150",
+        "new": false,
+        "paid": false,
+        "password": "!K3o!nHrs4M5Lxtg",
         "port": "3128",
         "premium": false,
-        "username": "london_squid_user",
+        "username": "la_premium"
-        "time": 62614.32
+    },
+    {
+        "time": Infinity,
+        "city": "Chicago 3",
+        "country": "USA",
+        "host": "107.152.38.47",
+        "new": false,
+        "paid": false,
+        "password": "yirXePSAqCE8jhLg88FBTfCz",
+        "port": "3128",
+        "premium": false,
+        "username": "chicago_squid_user_free"
     }
 ]