Как асинхронно проверить прокси в python
? Я взяла за пример библиотеку proxy-checker и написала код с нужными мне параметрами. Он полностью рабочий, но одна проверка занимает от 10 до 20 секунд и так как метод не асинхронный, во время выполнения на PyQt5
интерфейс подвисает и отвисает на секунду только между каждой проверкой.
Можно ли данный код переписать с асинхронностью или возможно есть другие, уже готовые, асинхронные методы проверки прокси?
Я пробовала использовать:
loop = asyncio.get_event_loop()
res = await loop.run_in_executor(None, ..., ...)
Интерфейс не зависал, но при этом почему-то не возвращались результаты выполнения, к тому же это не является прямым решением.
Вот весь мой код:
import re
import time
import pycurl
import random
from io import BytesIO
proxy_judges = ['http://proxyjudge.us/azenv.php', 'http://mojeip.net.pl/asdfa/azenv.php']
def send_query(proxy=False, url=None, user=None, password=None):
response = BytesIO()
c = pycurl.Curl()
c.setopt(c.URL, url or random.choice(proxy_judges))
c.setopt(c.WRITEDATA, response)
c.setopt(c.TIMEOUT, 5)
if user is not None and password is not None:
c.setopt(c.PROXYUSERPWD, f"{user}:{password}")
c.setopt(c.SSL_VERIFYHOST, 0)
c.setopt(c.SSL_VERIFYPEER, 0)
if proxy:
c.setopt(c.PROXY, proxy)
try:
c.perform()
except:
return False
if c.getinfo(c.HTTP_CODE) != 200:
return False
timeout = round(c.getinfo(c.CONNECT_TIME) * 1000)
response = response.getvalue().decode('iso-8859-1')
return {
'timeout': timeout,
'response': response
}
def get_country(ip):
r = send_query(url='https://ip2c.org/' + ip)
if r and r['response'][0] == '1':
r = r['response'].split(';')
return [r[3], r[1]]
return ['-', '-']
def check_proxy(proxy, check_country=True, check_address=False, user=None, password=None):
timeout, protocols = 0, {}
for protocol in ['http', 'socks4', 'socks5']:
r = send_query(proxy=protocol + '://' + proxy, user=user, password=password)
if not r:
continue
protocols[protocol] = r
timeout += r['timeout']
if len(protocols) == 0:
return False
r = protocols[random.choice(list(protocols.keys()))]['response']
timeout = timeout // len(protocols)
results = {
'protocols': list(protocols.keys()),
'timeout': timeout
}
if check_country:
country = get_country(proxy.split(':')[0])
results['country'] = country[0]
results['country_code'] = country[1]
if check_address:
remote_regex = r'REMOTE_ADDR = (\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})'
remote_addr = re.search(remote_regex, r)
if remote_addr:
remote_addr = remote_addr.group(1)
results['remote_address'] = remote_addr
return results
if __name__ == '__main__':
all_proxies = open('proxies.txt', 'r', encoding='utf-8').read().split('\n')
for proxy in all_proxies:
cur_proxy = proxy.split(':')
ip_port = cur_proxy[0] + ':' + cur_proxy[1]
proxy_log, proxy_pass = cur_proxy[2], cur_proxy[3]
start_time = time.time()
res = check_proxy(ip_port, check_country=True, user=proxy_log, password=proxy_pass)
if res:
protocol = str(res['protocols'][0])
timeout = str(res['timeout'])
country = str(res['country'])
code = str(res['country_code'])
print("Прошло %s секунд" % (round(time.time() - start_time)))
print(f'{proxy}: timeout {timeout} ({protocol}) | {country} ({code})')
Пример вывода:
Прошло 15 секунд
...: timeout 86 (http) | Russian Federation (RU)
Прошло 16 секунд
...: timeout 89 (http) | - (-)
Прошло 12 секунд
...: timeout 87 (http) | Russian Federation (RU)
Прошло 13 секунд
...: timeout 83 (http) | Russian Federation (RU)
threading
случаем не то, что Вам нужно?