跳转至

代理轮换

ProxyRotator 可以在任何 Fetcher 和会话中切换代理，也能安全地供多个线程使用。

可以直接导入这个类：

from scrapling.fetchers import ProxyRotator

scrapling.engines.toolbelt.proxy_rotation.ProxyRotator 

ProxyRotator(proxies, strategy=cyclic_rotation)

线程安全的代理轮换器，支持替换轮换策略。

支持默认的循环轮换、通过可调用对象实现的自定义策略，以及 URL 字符串或 Playwright 格式的代理字典。

初始化代理轮换器。

参数 说明
proxies

代理 URL 或 Playwright 格式的代理字典列表。字符串格式："http://proxy1:8080" 或 "http://user:pass@proxy:8080"。字典格式：{"server": "http://proxy:8080", "username": "user", "password": "pass"}。

类型： List[ProxyType]

strategy

轮换策略函数，接收 (proxies, current_index)，返回 (proxy, next_index)，默认为 cyclic_rotation。

类型： RotationStrategy 默认值： cyclic_rotation

源代码位于 scrapling/engines/toolbelt/proxy_rotation.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def __init__(
    self,
    proxies: List[ProxyType],
    strategy: RotationStrategy = cyclic_rotation,
):
    """
    Initialize the proxy rotator.

    :param proxies: List of proxy URLs or Playwright-style proxy dicts.
        - String format: "http://proxy1:8080" or "http://user:pass@proxy:8080"
        - Dict format: {"server": "http://proxy:8080", "username": "user", "password": "pass"}
    :param strategy: Rotation strategy function. Takes (proxies, current_index) and returns (proxy, next_index). Defaults to cyclic_rotation.
    """
    if not proxies:
        raise ValueError("At least one proxy must be provided")

    if not callable(strategy):
        raise TypeError(f"strategy must be callable, got {type(strategy).__name__}")

    self._strategy = strategy
    self._lock = Lock()

    # Validate and store proxies
    self._proxies: List[ProxyType] = []
    self._proxy_to_index: Dict[str, int] = {}  # O(1) lookup by unique key (server + username)
    for i, proxy in enumerate(proxies):
        if isinstance(proxy, (str, dict)):
            if isinstance(proxy, dict) and "server" not in proxy:
                raise ValueError("Proxy dict must have a 'server' key")

            self._proxy_to_index[_get_proxy_key(proxy)] = i
            self._proxies.append(proxy)
        else:
            raise TypeError(f"Invalid proxy type: {type(proxy)}. Expected str or dict.")

    self._current_index = 0

__slots__ class-attribute instance-attribute 

__slots__ = (
    "_proxies",
    "_proxy_to_index",
    "_strategy",
    "_current_index",
    "_lock",
)

proxies property 

proxies

获取所有已配置代理的副本。

get_proxy 

get_proxy()

根据轮换策略获取下一个代理。

源代码位于 scrapling/engines/toolbelt/proxy_rotation.py 
88
89
90
91
92
def get_proxy(self) -> ProxyType:
    """Get the next proxy according to the rotation strategy."""
    with self._lock:
        proxy, self._current_index = self._strategy(self._proxies, self._current_index)
        return proxy

__len__ 

__len__()

返回已配置代理的总数。

源代码位于 scrapling/engines/toolbelt/proxy_rotation.py 
 99
100
101
def __len__(self) -> int:
    """Return the total number of configured proxies."""
    return len(self._proxies)

__repr__ 

__repr__()
源代码位于 scrapling/engines/toolbelt/proxy_rotation.py 
103
104
def __repr__(self) -> str:
    return f"ProxyRotator(proxies={len(self._proxies)})"