1- import random
2- import requests
3- import time
4- import threading
1+ import random
2+ import requests
3+ import time
4+ import threading
55import socket
6- import redis
7-
8-
9- # Создаем подключение к redis
10- r = redis .StrictRedis (host = 'localhost' , port = 6379 , db = 0 )
116
127
138def fetch_proxies ():
14- url = "https://api.proxyscrape.com/v2/?request=displayproxies&protocol=http"
9+ """Fetch a list of proxy servers from proxyscrape.com.
10+
11+ Returns:
12+ list: A list of proxy servers in the format "IP:Port".
13+ """
14+ url = "https://www.proxy-list.download/api/v1/get?type=https"
1515 response = requests .get (url )
1616 if response .status_code == 200 :
1717 return response .text .split ("\r \n " )[:- 1 ]
1818 print (f"Error fetching proxies: { response .status_code } )
19- return []
19+ return []
2020
2121
2222def test_proxy (proxy , prompt , timeout ):
23+ """Test the given proxy server with a specified prompt and timeout.
24+
25+ Args:
26+ proxy (str): The proxy server in the format "IP:Port".
27+ prompt (str): The test prompt to be used for testing.
28+ timeout (int): The maximum time in seconds allowed for the test.
29+ """
2330 try :
24- ip , port = proxy .split (':' )
31+ # Split IP and Port
32+ ip , port = proxy .split (':' )
33+
34+ # Create a socket object
2535 sock = socket .socket (socket .AF_INET , socket .SOCK_STREAM )
36+
37+ # Start the timer
2638 start_time = time .time ()
39+
40+ # Connect to the proxy server
2741 sock .connect ((ip , int (port )))
42+
43+ # Stop the timer and calculate the elapsed time
2844 end_time = time .time ()
2945 elapsed_time = end_time - start_time
46+
47+ # Print the elapsed time
48+ #print(f"Elapsed time: {elapsed_time} seconds")
49+
50+ # Close the socket
3051 sock .close ()
31-
52+
53+ # Check if the elapsed time is below the timeout
3254 if elapsed_time < timeout :
3355 print (f"proxy: { proxy } { elapsed_time } )
34- r . rpush ( 'working_proxies' , proxy ) # Сохраняем рабочего прокси в redis
56+ add_working_proxy ( proxy )
3557 except Exception as e :
3658 pass
3759
3860
39- def get_working_proxies (prompt , timeout = 1 ):
40- proxy_list = fetch_proxies ()
41- threads = []
42- r .delete ('working_proxies' ) # Очищаем список рабочих прокси в redis перед обновлением
61+ def add_working_proxy (proxy ):
62+ """Add a working proxy server to the global working_proxies list.
63+
64+ Args:
65+ proxy (str): The proxy server in the format "IP:Port".
66+ """
67+ global working_proxies
68+ working_proxies .append (proxy )
69+
70+
71+ def remove_proxy (proxy ):
72+ """Remove a proxy server from the global working_proxies list.
73+
74+ Args:
75+ proxy (str): The proxy server in the format "IP:Port".
76+ """
77+ global working_proxies
78+ if proxy in working_proxies :
79+ working_proxies .remove (proxy )
80+
81+
82+ def get_working_proxies (prompt , timeout = 5 ):
83+ """Fetch and test proxy servers, adding working proxies to the global working_proxies list.
4384
85+ Args:
86+ prompt (str): The test prompt to be used for testing.
87+ timeout (int, optional): The maximum time in seconds allowed for testing. Defaults to 5.
88+ """
89+ proxy_list = fetch_proxies ()
90+ threads = []
91+
4492 for proxy in proxy_list :
45- thread = threading .Thread (target = test_proxy , args = (proxy , prompt , timeout ))
93+ thread = threading .Thread (target = test_proxy , args = (
94+ proxy , prompt , timeout ))
4695 threads .append (thread )
4796 thread .start ()
48-
97+
4998 for t in threads :
50- t .join (timeout )
99+ t .join (timeout )
51100
52101
53102def update_working_proxies ():
103+ """Continuously update the global working_proxies list with working proxy servers."""
104+ global working_proxies
54105 test_prompt = "What is the capital of France?"
55-
106+
56107 while True :
108+ working_proxies = [] # Clear the list before updating
57109 get_working_proxies (test_prompt )
58110 print ('proxies updated' )
59- time .sleep (1800 ) # Обновляем список прокси каждые 30 минут
60-
111+ time .sleep (1800 ) # Update proxies list every 30 minutes
112+
113+
114+ def get_random_proxy ():
115+ """Get a random working proxy server from the global working_proxies list.
61116
62- def get_random_proxy ():
63- # Получаем случайного прокси из рабочих
64- working_proxies = r .lrange ('working_proxies' , 0 , - 1 )
65- return random .choice (working_proxies )
117+ Returns:
118+ str: A random working proxy server in the format "IP:Port".
119+ """
120+ global working_proxies
121+ return random .choice (working_proxies )
0 commit comments