Rotador de proxies seguro para varios hilos, con estrategias de rotación intercambiables.
Admite rotación cíclica, que es la predeterminada, estrategias propias mediante una función, y proxies como cadenas de URL o diccionarios con formato de Playwright.
Inicia el rotador de proxies.
| PARÁMETRO |
DESCRIPCIÓN |
proxies
|
Lista de URL de proxies o diccionarios con formato de Playwright. Formato de cadena: "http://proxy1:8080" o "http://user:pass@proxy:8080". Formato de diccionario: {"server": "http://proxy:8080", "username": "user", "password": "pass"}.
TIPO:
List[ProxyType]
|
strategy
|
Función que define la estrategia de rotación. Recibe (proxies, current_index) y devuelve (proxy, next_index). El valor predeterminado es cyclic_rotation.
TIPO:
RotationStrategy
PREDETERMINADO:
cyclic_rotation
|
Código fuente en scrapling/engines/toolbelt/proxy_rotation.py
| 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__ = (
"_proxies",
"_proxy_to_index",
"_strategy",
"_current_index",
"_lock",
)
Obtiene una copia de todos los proxies configurados.
Obtiene el siguiente proxy según la estrategia de rotación.
Código fuente en scrapling/engines/toolbelt/proxy_rotation.py
| 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
|
Devuelve el número total de proxies configurados.
Código fuente en scrapling/engines/toolbelt/proxy_rotation.py
| def __len__(self) -> int:
"""Return the total number of configured proxies."""
return len(self._proxies)
|
Código fuente en scrapling/engines/toolbelt/proxy_rotation.py
| def __repr__(self) -> str:
return f"ProxyRotator(proxies={len(self._proxies)})"
|