From 67118fb004f45ba72717dfc92e380b9e1cf615dd Mon Sep 17 00:00:00 2001 From: Peter Giters Date: Sun, 3 Nov 2024 15:55:34 +0300 Subject: [PATCH] Complete code refactoring --- main.py | 183 +++++++++++++++++++++++++++-------------------------- proxy.json | 78 +++++++++++------------ 2 files changed, 134 insertions(+), 127 deletions(-) diff --git a/main.py b/main.py index 4828938..e229a6e 100644 --- a/main.py +++ b/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(): - request = requests.get(PROXIES_LIST) - if request.status_code == 200: - return request.json() - else: - request.raise_for_status() - +def fetch_proxies(): + """Fetch the list of proxies from the defined URL.""" + response = requests.get(PROXIES_LIST_URL) + response.raise_for_status() + return response.json() -def get_best_proxy(proxy_json): - proxy_times = [] - for i, item in enumerate(proxy_json): - if item["host"] != "" and item["country"] != "Russia": - proxy_str = "" - 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: - del proxy_json[i] +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" - best_five_proxies = sorted(proxy_times, key=lambda x: x['time'])[:5] - return best_five_proxies +def construct_proxy_string(proxy): + """Construct a proxy string from the proxy dictionary.""" + if proxy.get("username"): + return f'{proxy["username"]}:{proxy["password"]}@{proxy["host"]}:{proxy["port"]}' + return f'{proxy["host"]}:{proxy["port"]}' + +def test_proxy(proxy): + """Test the proxy by measuring the download time.""" + proxy_str = construct_proxy_string(proxy) + print(f'Testing {proxy_str}') + + start_time = time.perf_counter() + 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() - best_proxy_json = get_best_proxy(proxy_json) - with open(os.path.abspath(__file__).split("main.py")[0]+"proxy.json", "w") as f: - f.write(json.dumps(best_proxy_json, indent=4)) - print("All done.") - + """Update the proxies list and save the best ones.""" + proxies = fetch_proxies() + best_proxies = get_best_proxies(proxies) + save_proxies_to_file(best_proxies) + print("All done.") def run_yt_dlp(): - proxy_url = "" - error = True - while error: - with open(os.path.abspath(__file__).split("main.py")[0]+"proxy.json", "r") as f: - proxy_dict = random.choice(json.load(f)) - print(f"Using proxy based in {proxy_dict['city']}, {proxy_dict['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") + """Run yt-dlp with a randomly selected proxy.""" + while True: + with open("proxy.json", "r") as f: + proxy = random.choice(json.load(f)) + proxy_str = construct_proxy_string(proxy) + print(f"Using proxy from {proxy['city']}, {proxy['country']}") + 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: - if "update" in sys.argv: - update_proxies() - elif len(sys.argv) == 1: - print("usage: main.py update | \n\nScript for starting yt-dlp with best free proxy\n\nCommands:\n update Update best proxy\n") - else: - sys.argv.pop(0) - run_yt_dlp() - except KeyboardInterrupt: - print("Canceled by user") + """Main function to handle script arguments and execute the appropriate command.""" + try: + if "update" in sys.argv: + update_proxies() + elif len(sys.argv) < 2: + print("usage: main.py update | \nScript for starting yt-dlp with best free proxy\nCommands:\n update Update best proxy") + else: + sys.argv.pop(0) + run_yt_dlp() + except KeyboardInterrupt: + print("Canceled by user") -main() +if __name__ == "__main__": + main() diff --git a/proxy.json b/proxy.json index 78da9f7..6cc3a13 100644 --- a/proxy.json +++ b/proxy.json @@ -1,29 +1,6 @@ [ { - "city": "Chicago 2", - "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 - }, - { + "time": 7.21, "city": "New York 1", "country": "USA", "host": "107.152.43.179", @@ -32,31 +9,54 @@ "password": "pJTg94JCukVjz2b", "port": "3128", "premium": true, - "username": "ny_premium", - "time": 20.92 + "username": "ny_premium" }, { - "city": "Hesse", - "country": "Germany", - "host": "162.19.230.101", + "time": 7.99, + "city": "Chicago 2", + "country": "USA", + "host": "107.152.38.99", "new": false, "paid": false, - "password": "enfo8BM3YDbGfyANYJRCKyL7hGrH", + "password": "L5KT7rSkCrxn8mm4HngTbK8f4k", "port": "3128", - "premium": false, - "username": "hesse_squid_user", - "time": 62594.22 + "premium": true, + "username": "chicago_squid_user" }, { - "city": "London", - "country": "UK", - "host": "83.147.17.103", + "time": 12.86, + "city": "New York 2", + "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", - "time": 62614.32 + "username": "la_premium" + }, + { + "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" } ] \ No newline at end of file