2

Как асинхронно проверить прокси в 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)
2
  • threading случаем не то, что Вам нужно?
    – Sunny Cove
    11 ноя 2021 в 20:47
  • не лишним было бы и ваш proxies.txt приложить чтобы не бороздить просторы в поисках списка живых прокси 12 ноя 2021 в 3:45

2 ответа 2

0
+100

Поскольку в метках вы поставили PyQt5, его и используем. А конкретнее - QThread:

import sys
import time
from PyQt5 import QtCore, QtWidgets
from PyQt5.QtCore import QObject, QThread, pyqtSignal
# ваш код прокси-чекера из вопроса пусть лежит в proxyChecker.py
from proxyChecker import check_proxy


# этот объект будет крутиться в отдельном потоке
class Worker(QObject):

    # у объекта будут два сигнала - поток завершен и сигнал с результатом проверки очередного прокси
    finished = pyqtSignal()
    progress = pyqtSignal(str)
    running = False

    # это будем использовать если надо будет остановить проверку
    def stop(self):
        self.running = False

    # это основной цикл проверки прокси
    def run(self):
        self.running = True
        all_proxies = open('proxies.txt', 'r', encoding='utf-8').read().split('\n')
        for proxy in all_proxies:
            self.progress.emit(f'Проверяем: {proxy}')
            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'])
                result = f"Прошло {round(time.time() - start_time)} секунд\n"
                result += f'{proxy}: timeout {timeout} ({protocol}) | {country} ({code})'
                self.progress.emit(result)
            else:
                result = f"Прошло {round(time.time() - start_time)} секунд\n"
                result += f'Недоступен'
                self.progress.emit(result)
            if not self.running:
                break
        # говорим основному потоку что мы отработали
        self.finished.emit()


class Ui_Form(object):
    def setupUi(self, Form):
        Form.setObjectName("Form")
        Form.resize(453, 408)
        self.verticalLayout = QtWidgets.QVBoxLayout(Form)
        self.verticalLayout.setObjectName("verticalLayout")
        self.verticalLayout_2 = QtWidgets.QVBoxLayout()
        self.verticalLayout_2.setObjectName("verticalLayout_2")
        self.textBrowser = QtWidgets.QTextBrowser(Form)
        self.textBrowser.setObjectName("textBrowser")
        self.verticalLayout_2.addWidget(self.textBrowser)
        self.verticalLayout.addLayout(self.verticalLayout_2)
        self.horizontalLayout = QtWidgets.QHBoxLayout()
        self.horizontalLayout.setObjectName("horizontalLayout")
        spacerItem = QtWidgets.QSpacerItem(40, 20, QtWidgets.QSizePolicy.Expanding, QtWidgets.QSizePolicy.Minimum)
        self.horizontalLayout.addItem(spacerItem)
        self.pushButton = QtWidgets.QPushButton(Form)
        self.pushButton.setObjectName("pushButton")
        self.horizontalLayout.addWidget(self.pushButton)
        self.pushButtonCancel = QtWidgets.QPushButton(Form)
        self.pushButtonCancel.setObjectName("pushButton2")
        self.pushButtonCancel.setEnabled(False)
        self.horizontalLayout.addWidget(self.pushButtonCancel)
        spacerItem1 = QtWidgets.QSpacerItem(40, 20, QtWidgets.QSizePolicy.Expanding, QtWidgets.QSizePolicy.Minimum)
        self.horizontalLayout.addItem(spacerItem1)
        self.verticalLayout.addLayout(self.horizontalLayout)
        self.retranslateUi(Form)
        QtCore.QMetaObject.connectSlotsByName(Form)

    def retranslateUi(self, Form):
        _translate = QtCore.QCoreApplication.translate
        Form.setWindowTitle(_translate("Form", "Example"))
        self.pushButton.setText(_translate("Form", "Start"))
        self.pushButtonCancel.setText(_translate("Form", "Stop"))



class MyWindow(QtWidgets.QWidget):
    def __init__(self, parent=None):
        super().__init__()
        self.ui = Ui_Form()
        self.ui.setupUi(self)
        # цепляем кнопочки к функциям запуска/остановки проверки
        self.ui.pushButton.clicked.connect(self.runLongTask)
        self.ui.pushButtonCancel.clicked.connect(self.breakLongTask)

    # функция будет отрабатывать когда поток проверки пришлет строку с результатом
    def reportProgress(self, n):
        self.ui.textBrowser.append(n)

    def breakLongTask(self):
        self.worker.stop()

    # при нажатии кнопы запуска надо создать поток проверки
    def runLongTask(self):
        # сделаем активной кнопку остановки проверки
        self.ui.pushButtonCancel.setEnabled(True)
        # создадим объект QThread
        self.thread = QThread()
        # создадим объект, который будет крутиться созданном в потоке
        self.worker = Worker()
        # запихиваем наш объект в новый поток
        self.worker.moveToThread(self.thread)
        # делаем обвязку сигналов:
        # Функция, вызываемая при запуске потока
        self.thread.started.connect(self.worker.run)
        # при получении от потока сигнала с результатом, обработаем его в reportProgress
        self.worker.progress.connect(self.reportProgress)
        # действия при завершении потока:
        # а) выйти из потока (завершить поток)
        self.worker.finished.connect(self.thread.quit)
        # б) запланировать удаление объекта, который крутился в потоке
        self.worker.finished.connect(self.worker.deleteLater)
        # в) запланировать удаление самого объекта QThread
        self.thread.finished.connect(self.thread.deleteLater)

        # Наш поток готов к запуску. Стартуем
        self.thread.start()

        # Пока поток работает, кнопка запуска будет недоступна
        self.ui.pushButton.setEnabled(False)

        # После завершения потока сделаем кнопку запуска активной а кнопку остановки неактивной
        self.thread.finished.connect(
            lambda: (
                self.ui.textBrowser.append("finished"),
                self.ui.pushButtonCancel.setEnabled(False),
                self.ui.pushButton.setEnabled(True)
                     )
        )


if __name__ == '__main__':
    app = QtWidgets.QApplication(sys.argv)
    window = MyWindow()
    window.show()
    sys.exit(app.exec())
5
  • Возможно ли в классе Worker как-то взаимодействовать с интерфейсом (self.ui)? Для вывода процесса проверки. 15 ноя 2021 в 0:31
  • Конечно можно. см. сигнал progress в Worker, и соответственно привязанный к нему слот reportProgress основного (gui) потока, выводящий результат проверки очередного прокси в textBrowser. При необходимости вы можете определить и свои сигналы с нужными вам параметрами и обрабатывать их в основном потоке 15 ноя 2021 в 2:56
  • да, это я сразу поняла, сейчас через emit выводится итоговый результат проверки. В примере всё выводится в одно текстовое поле, а как быть если нужно распределить текст по разным текстовым полям? Я подумала, что это можно сделать так: ibb.co/zFnBJjh, где n['type'] будет определять куда именно нужно выводить. Есть ли более грамотное решение этого? 15 ноя 2021 в 14:53
  • примерно так и надо, только сигнал будет progress = pyqtSignal(dict) и эмит соответственно словарем например self.progress.emit({'type': 'Проверяем', 'value': proxy}). В слоте уже и будет нужное вам n['type'] 15 ноя 2021 в 15:09
  • хорошо, спасибо, это я сразу сделала (progress = pyqtSignal(dict)), просто хотела уточнить, является ли такой подход корректным. 16 ноя 2021 в 18:58
0

Вариант 1. Можно воспользоваться стандартным threading из python

import threading
 
def check_proxies(filename, mode, encoding):
    all_proxies = open(filename, mode, encoding).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})')
 
if __name__ == '__main__':
    # Создаем объект потока с нужной нам функцией
    thread_check_proxies = threading.Thread(target=check_proxies, args=('proxies.txt','r','utf-8'))
    # Запускаем поток
    thread_check_proxies.start()

Вариант 2. В Qt есть несколько способов работы с потоками, все зависит от того, что именно вам нужно. Например, можно создать класс проверки прокси ("рабочего" - Worker), затем переместить его в поток и запустить соответствующий метод "рабочего" внутри потока:

class CheckProxiesWorker(QtCore.QObject):
    # Сигнал с результатом проверки прокси
    signal_check_result = QtCore.pyqtSignal(object)
    
    def __init__(self, filename, mode='r', encoding='utf-8'):
        super().__init__(self)
        self.filename = filename
        self.mode = mode
        self.encoding = encoding

    # 
    def work(self):
      all_proxies = open(self.filename, self.mode, self.encoding).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:
            # Эмитируем сигнал с результатом проверки            
            self.signal_check_result.emit(res)

Где-то в приложении запускаем поток:

# Метод обработки результата проверки
def eval_result(self, res):
    # ...

def start_thread_check_proxies:
  # создаем поток
  self.thread = QtCore.QThread()
  # создаем объект рабочего
  self.worker = CheckProxiesWorker('proxies.txt', 'r', 'utf-8')

  # перенесём объект в поток
  self.worker.moveToThread(self.thread)

  # соединяем сигнал с методом вывода полученной информации
  self.worker.signal_check_result.connect(self.eval_result)

  # соединяем сигнал запуска потока с рабочим методом
  self.thread.started.connect(self.worker.work)
  # запускаем поток
  self.thread.start()

Кроме того, можно наследоваться от QThread, тогда все будет немного лаконичнее.

2
  • разработчик Qt Bradley T.Hughes утверждает, что наследование QThread только для выполнения кода класса в отдельном потоке – идея в корне неправильная 12 ноя 2021 в 15:07
  • @Sergey Tatarincev это мнение давно опровергнуто и, более того, в официальной документации Qt есть пример с наследованием. Так что, пользуйтесь смело. doc.qt.io/qt-5/qthread.html 12 ноя 2021 в 15:12

Ваш ответ

Нажимая на кнопку «Отправить ответ», вы соглашаетесь с нашими условиями использования и подтверждаете, что прочитали и поняли наши политику конфиденциальности и нормы поведения.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.