Spaces:
Sleeping
Sleeping
| import random | |
| import requests | |
| import time | |
| import threading | |
| from freeGPT import gpt3 | |
| def fetch_proxies(): | |
| """Fetch a list of proxy servers from proxyscrape.com. | |
| Returns: | |
| list: A list of proxy servers in the format "IP:Port". | |
| """ | |
| url = "https://api.proxyscrape.com/v2/?request=displayproxies&protocol=http&timeout=1000&country=all&ssl=all&anonymity=all" | |
| response = requests.get(url) | |
| if response.status_code == 200: | |
| proxy_list = response.text.split("\r\n")[:-1] | |
| return proxy_list | |
| else: | |
| print(f"Error fetching proxies: {response.status_code}") | |
| return [] | |
| def test_proxy(proxy, prompt, timeout): | |
| """Test the given proxy server with a specified prompt and timeout. | |
| Args: | |
| proxy (str): The proxy server in the format "IP:Port". | |
| prompt (str): The test prompt to be used for testing. | |
| timeout (int): The maximum time in seconds allowed for the test. | |
| """ | |
| try: | |
| start_time = time.time() | |
| res = gpt3.Completion.create(prompt=prompt, proxy=proxy) | |
| end_time = time.time() | |
| response_time = end_time - start_time | |
| if response_time < timeout: | |
| response_time = int(response_time*1000) | |
| print(f'proxy: {proxy} [{response_time}ms] ✅') | |
| add_working_proxy((proxy)) | |
| except Exception as e: | |
| pass | |
| def add_working_proxy(proxy): | |
| """Add a working proxy server to the global working_proxies list. | |
| Args: | |
| proxy (str): The proxy server in the format "IP:Port". | |
| """ | |
| global working_proxies | |
| working_proxies.append(proxy) | |
| def remove_proxy(proxy): | |
| """Remove a proxy server from the global working_proxies list. | |
| Args: | |
| proxy (str): The proxy server in the format "IP:Port". | |
| """ | |
| global working_proxies | |
| if proxy in working_proxies: | |
| working_proxies.remove(proxy) | |
| def get_working_proxies(prompt, timeout=5): | |
| """Fetch and test proxy servers, adding working proxies to the global working_proxies list. | |
| Args: | |
| prompt (str): The test prompt to be used for testing. | |
| timeout (int, optional): The maximum time in seconds allowed for testing. Defaults to 5. | |
| """ | |
| proxy_list = fetch_proxies() | |
| threads = [] | |
| for proxy in proxy_list: | |
| thread = threading.Thread(target=test_proxy, args=( | |
| proxy, prompt, timeout)) | |
| threads.append(thread) | |
| thread.start() | |
| for t in threads: | |
| t.join(timeout) | |
| def update_working_proxies(): | |
| """Continuously update the global working_proxies list with working proxy servers.""" | |
| global working_proxies | |
| test_prompt = "What is the capital of France?" | |
| while True: | |
| working_proxies = [] # Clear the list before updating | |
| get_working_proxies(test_prompt) | |
| print('proxies updated') | |
| time.sleep(1800) # Update proxies list every 30 minutes | |
| def get_random_proxy(): | |
| """Get a random working proxy server from the global working_proxies list. | |
| Returns: | |
| str: A random working proxy server in the format "IP:Port". | |
| """ | |
| global working_proxies | |
| return random.choice(working_proxies) | |