Проверка прокси в питоне / селене

Question

Я пытаюсь использовать селен и питон для подключения к сайту. Затем я решил добавить прокси для всего процесса и использовать их для подключения к сайту. Прокси очищаются с помощью Proxy Broker (модуль Python). Так что я могу получить много прокси. Затем я сохраняю его в текстовом файле, и когда мне нужно подключиться, я выбираю его случайным образом из текстового файла. Однако здесь возникает проблема. При подключении к сайту прокси иногда не работают. В настоящее время это код, который я использую:

        import asyncio
    from proxybroker import Broker


    async def save(proxies, filename):
        """Save proxies to a file."""
        with open(filename, 'w') as f:
            while True:
                proxy = await proxies.get()
                if proxy is None:
                    break
                proto = 'https' if 'HTTPS' in proxy.types else 'http'
                row = '%s://%s:%d\n' % (proto, proxy.host, proxy.port)
                f.write(row)


    def main():
        proxies = asyncio.Queue()
        broker = Broker(proxies)
        tasks = asyncio.gather(broker.find(types=['HTTP', 'HTTPS'], limit=5),
                            save(proxies, filename='proxies.txt'))
        loop = asyncio.get_event_loop()
        loop.run_until_complete(tasks)


    if __name__ == '__main__':
        main()

    lines = open('proxies.txt').read().splitlines()
    rproxy =random.choice(lines)
    PROXY =  rproxy

Этот код является примером кода со страницы примера Proxy Broker (https://proxybroker.readthedocs.io/en/latest/examples.html)

Итак, я хочу иметь возможность сделать одну из двух вещей:

Вариант 1. Проверьте прокси сразу после их удаления, затем сохраните прокси, которые работают в текстовом файле, и вызовите их позже

Вариант 2. Проверьте прокси-серверы непосредственно перед его подключением к сайту. Таким образом, он проверяет, работает ли прокси, а затем, если он это делает, он использует. Если это не так, он пытается с другим.

У меня нет ни малейшего понятия, как это сделать. Одна вещь, которую мой друг предложил, состояла в том, чтобы использовать запросы и посмотреть, работает ли прокси, но у меня есть проблемы с этим, потому что я не могу отформатировать список прокси для автоматического использования с запросами.

Любая помощь / советы очень ценятся. Заранее спасибо !!!!

(Изменить) Я уже пробовал сообщения, подобные этим:

Проверка прокси в python

https://github.com/ApsOps/proxy-checker

https://www.calazan.com/how-to-use-proxies-with-an-http-session-using-the-python-requests-package/

https://codereview.stackexchange.com/questions/169246/python-proxy-checker-scanner

Никто из них не работал для меня :(

Answer 1

Что ж, я посмотрел документацию прокси-брокера и обнаружил, что лучшим решением является проверка встроенного атрибута proxy.is_working:

results = []
for proto in ngtrs:
    if proto == 'CONNECT:25':
        result = await self._check_conn_25(proxy, proto)
    else:
        result = await self._check(proxy, proto)
     results.append(result)

proxy.is_working = True if any(results) else False

Вы можете реализовать это в своем коде так:

import asyncio
from proxybroker import Broker


def get_random_proxy():
    """
    Get random proxy from 'proxies.txt'.
    """
    lines = open('proxies.txt').read().splitlines()
    rproxy =random.choice(lines)
    PROXY =  rproxy


async def save(proxies, filename):
    """
    Save proxies to a file.
    """
    with open(filename, 'w') as file:
        while True:
            proxy = await proxies.get()
            if proxy is None:
                break
            # Check accurately if the proxy is working.
            if proxy.is_working:
                protocol = 'https' if 'HTTPS' in proxy.types else 'http'
                line = '{protocol}://{proxy.host}:{proxy.port}\n'
                file.write(line)


def main():
    proxies = asyncio.Queue()
    broker = Broker(proxies)
    tasks = asyncio.gather(broker.find(types=['HTTP', 'HTTPS'], limit=5),
                           save(proxies, filename='proxies.txt'))
    loop = asyncio.get_event_loop()
    loop.run_until_complete(tasks)


if __name__ == '__main__':
    main()