Блог

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

В настоящее время я закончил тем, что сделал что-то вроде этого:

 import random
import threading
import time
from typing import Dict

from loguru import logger


class ProxyManager(object):
    def __init__(self, proxy_file_path=None):
        self.proxies = self.load_proxies(proxy_file_path) if proxy_file_path else {}

    @staticmethod
    def load_proxies(proxy_file_path):
        with open(proxy_file_path) as proxy_file:
            proxies = {'https': [], 'http': []}
            for proxy in proxy_file.readlines():
                if proxy.count(':') > 1:
                    ip, port, username, password = proxy.rstrip().split(':')
                    url = f'http://{ip}:{port}:{username}@{password}'
                else:
                    ip, port = proxy.rstrip().split(':')
                    url = f'http://{ip}:{port}'
                proxies['https'].append(url)

            return proxies

    def random_proxy_https(self):
        return random.choice(self.proxies['https'])

    def all_proxies_https(self):
        return [proxy for proxy in self.proxies['https']]


# -------------------------------------------------------------------------
# Proxies path
# -------------------------------------------------------------------------
proxies: Dict[str, ProxyManager] = {
    "rotating": ProxyManager("./proxies/rotating.txt"),
    "test": ProxyManager("./proxies/test.txt")
}

# -------------------------------------------------------------------------
# Proxies available
# -------------------------------------------------------------------------
all_proxies = proxies["rotating"].all_proxies_https()
proxy_dict = dict(zip(all_proxies, ['available'] * len(all_proxies)))

proxy_lock: threading = threading.Lock()


# -------------------------------------------------------------------------
# Initialize availability of proxies
# -------------------------------------------------------------------------
class AvailableProxies:
    def __enter__(self):
        proxy_lock.acquire()
        self.proxy = None

        while not self.proxy:
            available = [
                att for att, value in proxy_dict.items() if "available" in value
            ]
            if available:
                self.proxy = random.choice(available)
                proxy_dict[self.proxy] = "busy"
                break
            else:
                logger.info("Waiting ... no proxies available")
                time.sleep(.2)

        proxy_lock.release()
        return self.proxy

    def __exit__(self, exc_type, exc_val, exc_tb):
        proxy_dict[self.proxy] = "available"
 

Идея заключается в том, что у меня будет несколько сценариев, которым понадобится класс ProxyManager, но не AvailableProxies — Это означает, что мне не нужно устанавливать переменные

 # -------------------------------------------------------------------------
# Proxies available
# -------------------------------------------------------------------------
all_proxies = proxies["rotating"].all_proxies_https()
proxy_dict = dict(zip(all_proxies, ['available'] * len(all_proxies)))

proxy_lock: threading = threading.Lock()
 

если конкретный сценарий не требует класса. Интересно, возможно ли установить переменные внутри класса AvailableProxies и что переменные all_proxies, proxy_dict amp; proxy_lock должны быть установлены только один раз внутри класса, когда мы импортируем AvailableProxies их в скрипт?