agenticworkflowsspace/proxyvault

1import json2import os3from proxy_scraper import scrape_proxies4from proxy_tester import test_proxies5from colorama import Fore, Style, init6 7init(autoreset=True)8 9# Global list to store working proxies for incremental saving10results_list = []11 12def save_callback(result):13    """Callback function called every time a working proxy is found."""14    global results_list15    results_list.append(result)16    17    # Save as text (ip:port) - append mode18    with open("working_proxies.txt", "a") as f:19        f.write(f"{result['proxy']}\n")20        21    # Save as JSON - overwrite with full current list to keep it valid JSON22    with open("working_proxies.json", "w") as f:23        json.dump(results_list, f, indent=4)24 25def main():26    print(f"{Fore.MAGENTA}=== Mass Proxy Scraper & Tester with Whoer Info ==={Style.RESET_ALL}\n")27    28    # 1. Scrape Proxies29    proxies = scrape_proxies()30    31    if not proxies:32        print(f"{Fore.RED}No proxies scraped. Exiting.{Style.RESET_ALL}")33        return34 35    # Clear previous results36    if os.path.exists("working_proxies.txt"):37        os.remove("working_proxies.txt")38    if os.path.exists("working_proxies.json"):39        os.remove("working_proxies.json")40 41    # 2. Test Proxies42    # We pass the save_callback to update the file in real-time43    working_proxies = test_proxies(proxies, max_concurrent=100, callback=save_callback)44    45    # 3. Final Summary46    if working_proxies:47        print(f"\n{Fore.GREEN}Finished. Total working proxies: {len(working_proxies)}")48        print(f"{Fore.GREEN}Results saved to 'working_proxies.txt' and 'working_proxies.json'{Style.RESET_ALL}")49    else:50        print(f"\n{Fore.RED}No working proxies found.{Style.RESET_ALL}")51 52if __name__ == "__main__":53    main()54