In this Article
Ya sea web scraping o pruebas de sitios web, la necesidad de servidores proxy es innegable; sin embargo, necesita muchas direcciones diferentes para garantizar un acceso coherente. Al mismo tiempo, no cambiará entre servidores proxy manualmente. El rotador proxy es la respuesta en esta situación. En este artículo, estamos construyendo un rotador de proxy personalizado usando Python, desde cero.
¿Por qué es esencial construir un buen rotador de proxy?
Un rotador recorre automáticamente un grupo de proxy. Asigna otra dirección IP a cada solicitud o en un intervalo de tiempo establecido. El servidor de destino no ve que requests proviene de la misma fuente y el tráfico parece provenir de diferentes direcciones y ubicaciones, por lo que puede pasar desapercibido para los sistemas anti-bot. Es por eso que construir un rotador proxy es el núcleo del scraping o multicuenta. El rotador es responsable de mantener un flujo de autenticación constante, alinearse con las pautas de tráfico del sitio web y acceder a contenido específico de la región, especialmente en proyectos a gran escala con múltiples amenazas de conexión.
Además, existen algunas ventajas no tan obvias al utilizar un rotador proxy. Por ejemplo, mejora su privacidad y anonimato, ya que con su IP en constante cambio, es más difícil para los sitios web y rastreadores perfilar sus actividades. Los piratas informáticos están un paso más lejos de usted, ya que con su dirección real oculta, es más difícil para los estafadores que apuntan a dispositivos específicos descubrirlo.
Además, muchos sitios web ofrecen resultados diferentes según su ubicación o preferencias. Al utilizar varios IPs, obtiene datos más diversos y precisos. La rotación también impide ver contenido personalizado si no lo necesita.
Rotador proxy confiable: sus componentes
Además de simplemente usar un nuevo IP para cada solicitud, debes incluir algunas funciones más, para que un rotador de proxy realmente te beneficie y no se convierta en una fuente de problemas:
- Lógica de reintento: en caso de que un proxy no funcione, el rotador utiliza otro;
- Girar los encabezados User-Agent, para que su requests no parezca provenir de la misma fuente;
- Manejo de códigos de estado HTTP “malos” como 403 (“Prohibido”), 429 (“Demasiados requests”), 500 (“Error interno del servidor”);
- Deshabilitar los servidores proxy, que siguen fallando;
- Agregar pequeños retrasos entre requests para evitar activar sistemas anti-bot;
- Registro: para que, en caso de un reintento, pueda ver el motivo y qué proxy y User-Agent se utilizaron;
- Comprobación de estado antes del scraping: para probar todos los proxies antes de comenzar y asegurarse de que funcionen;
- Disyuntor para desactivar los servidores proxy inactivos durante algún tiempo;
- Sesiones fijas por host para que dentro de una sesión, el dominio vea el mismo IP, en caso de que lo necesite (para cuentas múltiples o algo así, cuando necesita permanecer en la misma sesión).
Tendremos esos puntos en cuenta al crear nuestro rotador de proxy.
Construyendo un rotador proxy usando Python
Crearemos un rotador proxy en un solo archivo. En nuestro ejemplo, lo llamamos proxy_rotator.py. A continuación, encontrará todos los fragmentos de código junto con una explicación de lo que hacen. Puedes copiarlos y pegarlos. Creamos código sin codificar datos como credenciales, por lo que no necesitará reemplazar nada más adelante.
Empezando
- Instale Python desde su sitio web oficial (o actualizarlo). Necesitas la versión 3.8 o superior.
- Ejecute el siguiente comando para instalar el requests biblioteca. Lo necesitaremos para hacer HTTP requests.
pip install requests
- Obtenga una lista de proxies en la página del plan necesario. Puede copiarlos y pegarlos manualmente en un archivo .txt desde su Panel de control DataImpulse o descargarlos. Nombramos nuestro archivo proxies.txt. Debería verse así:
http://login:password.host:port
https://login:password.host:port
Puede cambiar el formato del proxy si es necesario.
Importaciones y configuración
Primero, debemos incorporar herramientas llamadas “módulos” que usaremos.
import argparse
import logging
import os
import random
import threading
import time
from dataclasses import dataclass
from typing import Dict, Iterable, List, Optional, Tuple
import requests
from requests import Response
pedido envía web requests, aleatorio elige proxies y encabezados, y threading es responsable de manejar múltiples subprocesos de conexión.
Configuración predeterminada
La siguiente parte cumple varias funciones cruciales: rota los agentes de usuario, cambia a otro proxy en caso de un código de error y comprueba si un proxy funciona.
DEFAULT_USER_AGENTS: List[str] = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:125.0) Gecko/20100101 Firefox/125.0",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36 Edg/124.0.0.0",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 13_4) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.5 Safari/605.1.15",
"Mozilla/5.0 (iPhone; CPU iPhone OS 16_5 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.5 Mobile/15E148 Safari/604.1",
"Mozilla/5.0 (Linux; Android 13; Pixel 7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Mobile Safari/537.36",
]
DEFAULT_ROTATE_ON_STATUS = {403, 407, 409, 418, 421, 425, 429, 500, 502, 503, 504}
DEFAULT_HEALTHCHECK_URL = "http://api.ipify.org"
Puede reemplazar o agregar agentes de usuario y utilizar otro sitio web para un control de estado si lo desea. En nuestro caso utilizamos httpbin.org/ip. Otras opciones decentes son https://httpbin.org/get (muestra encabezados de solicitud así como IP), https://api.ipify.org (o https://api.ipify.org?format=json para devolver su IP en JSON), https://ifconfig.me o https://ifconfig.me/all.json (devuelve encabezados + IP en JSON), o http://ip-api.com/json/ (muestra país, ciudad, ISP y IP).
Objeto proxy
Esta parte define cómo se almacenan los servidores proxy.
@dataclass(frozen=True)
class Proxy:
url: str
label: str = ""
def to_requests_proxies(self) -> Dict[str, str]:
return {"http": self.url, "https": self.url}
Nuevamente, no es necesario reemplazar nada aquí.
Leer un archivo proxy
Esta parte lee proxies de un .TXT archivo.
def parse_proxy_file(path: str) -> List[Proxy]:
proxies: List[Proxy] = []
with open(path, "r", encoding="utf-8") as f:
for line in f:
raw = line.strip()
if not raw or raw.startswith("#"):
continue
raw = raw.split(" #", 1)[0].split("\t#", 1)[0].strip()
proxies.append(Proxy(url=raw))
if not proxies:
raise ValueError(f"No proxies found in file: {path}")
return proxies
Retraso de reintento
Este fragmento de código diseña una lógica de reintento. Con él, un rotador espera más cada vez que falla una solicitud para evitar limitaciones y bloqueos de servicio.
def _exp_backoff_sleep(base: float, attempt: int, max_sleep: float) -> None:
sleep_s = min(base * (2 ** attempt) + random.random() * base, max_sleep)
time.sleep(sleep_s)
La clase de rotador proxy
El corazón de tu rotador. Elige un proxy, intenta requests nuevamente y realiza un seguimiento de los errores. También decide una estrategia: si usar proxies aleatoriamente o recorrer la lista, establece tiempos de espera y encabezados, y usa el mismo proxy para un sitio web. Envía requests, marca los servidores proxy como incluidos en la lista blanca o bloqueados y ejecuta comprobaciones de estado.
class ProxyRotator:
def __init__(
self,
proxies: Iterable[Proxy],
strategy: str = "round_robin",
max_retries: int = 3,
timeout: Tuple[float, float] = (10.0, 30.0),
rotate_on_status: Optional[Iterable[int]] = None,
user_agents: Optional[List[str]] = None,
sticky_per_host: bool = True,
circuit_threshold: int = 2,
circuit_cooldown: float = 60.0,
backoff_base: float = 0.5,
backoff_max: float = 5.0,
healthcheck_url: Optional[str] = None,
logger: Optional[logging.Logger] = None,
) -> None:
self.proxies: List[Proxy] = list(proxies)
if not self.proxies:
raise ValueError("ProxyRotator requires at least one proxy")
if strategy not in {"round_robin", "random"}:
raise ValueError("strategy must be 'round_robin' or 'random'")
self.strategy = strategy
self.max_retries = max_retries
self.timeout = timeout
self.rotate_on_status = set(rotate_on_status or DEFAULT_ROTATE_ON_STATUS)
self.user_agents = user_agents or DEFAULT_USER_AGENTS
self.sticky_per_host = sticky_per_host
self.circuit_threshold = circuit_threshold
self.circuit_cooldown = circuit_cooldown
self.failure_counts: Dict[Proxy, int] = {p: 0 for p in self.proxies}
self.disabled_until: Dict[Proxy, float] = {p: 0.0 for p in self.proxies}
self.backoff_base = backoff_base
self.backoff_max = backoff_max
self.healthcheck_url = healthcheck_url
self._local = threading.local()
self._sticky_map: Dict[str, Proxy] = {}
self._idx = 0
self._lock = threading.Lock()
self.log = logger or logging.getLogger("proxy_rotator")
def get(self, url: str, **kwargs) -> Response:
return self.request("GET", url, **kwargs)
def post(self, url: str, **kwargs) -> Response:
return self.request("POST", url, **kwargs)
def request(self, method: str, url: str, **kwargs) -> Response:
session = getattr(self._local, "session", None)
if session is None:
session = requests.Session()
self._local.session = session
headers = kwargs.pop("headers", {}) or {}
headers.setdefault("User-Agent", random.choice(self.user_agents))
timeout = kwargs.pop("timeout", self.timeout)
last_exc: Optional[Exception] = None
last_response: Optional[Response] = None
for attempt in range(self.max_retries + 1):
proxy = self._choose_proxy(url)
if proxy is None:
raise RuntimeError("No healthy proxies currently available")
try:
self.log.debug("Using proxy %s for %s %s", proxy.url, method, url)
response = session.request(
method,
url,
headers=headers,
proxies=proxy.to_requests_proxies(),
timeout=timeout,
**kwargs,
)
last_response = response
if response.status_code in self.rotate_on_status:
self._mark_failure(proxy, f"HTTP {response.status_code}")
self._maybe_unstick_host(url, proxy)
if attempt < self.max_retries:
_exp_backoff_sleep(self.backoff_base, attempt, self.backoff_max)
continue
else:
self._mark_success(proxy)
return response
except requests.RequestException as exc:
last_exc = exc
self._mark_failure(proxy, repr(exc))
self._maybe_unstick_host(url, proxy)
if attempt < self.max_retries:
_exp_backoff_sleep(self.backoff_base, attempt, self.backoff_max)
continue
break
if last_exc:
raise last_exc
assert last_response is not None
return last_response
def health_check(self, url: Optional[str] = None, sample: Optional[int] = None) -> Dict[str, bool]:
check_url = url or self.healthcheck_url or DEFAULT_HEALTHCHECK_URL
to_test = list(self.proxies)
if sample is not None:
to_test = random.sample(to_test, k=min(sample, len(to_test)))
results: Dict[str, bool] = {}
for p in to_test:
try:
r = requests.get(check_url, proxies=p.to_requests_proxies(), timeout=(5, 10))
healthy = r.ok
except Exception:
healthy = False
results[p.url] = healthy
if healthy:
self._mark_success(p)
else:
self._mark_failure(p, "healthcheck")
return results
def _choose_proxy(self, url: str) -> Optional[Proxy]:
host = requests.utils.urlparse(url).hostname or ""
now = time.time()
with self._lock:
if self.sticky_per_host:
sticky = self._sticky_map.get(host)
if sticky and self.disabled_until.get(sticky, 0.0) <= now:
return sticky
enabled = [p for p in self.proxies if self.disabled_until.get(p, 0.0) <= now]
if not enabled:
return None
if self.strategy == "round_robin":
proxy = enabled[self._idx % len(enabled)]
self._idx = (self._idx + 1) % len(enabled)
else:
proxy = random.choice(enabled)
if self.sticky_per_host:
self._sticky_map[host] = proxy
return proxy
def _mark_failure(self, proxy: Proxy, reason: str) -> None:
with self._lock:
cnt = self.failure_counts.get(proxy, 0) + 1
self.failure_counts[proxy] = cnt
if cnt >= self.circuit_threshold:
self.disabled_until[proxy] = time.time() + self.circuit_cooldown
self.log.warning("Disabling proxy %s for %.1fs (reason: %s)", proxy.url, self.circuit_cooldown, reason)
def _mark_success(self, proxy: Proxy) -> None:
with self._lock:
self.failure_counts[proxy] = 0
self.disabled_until[proxy] = 0.0
def _maybe_unstick_host(self, url: str, proxy: Proxy) -> None:
if not self.sticky_per_host:
return
host = requests.utils.urlparse(url).hostname or ""
with self._lock:
if self._sticky_map.get(host) == proxy:
self._sticky_map.pop(host, None)
Opciones de línea de comandos
Necesita esta pieza para ejecutar un rotador con un comando de una sola línea a través de la terminal.
def build_arg_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(description="HTTP proxy rotator")
p.add_argument("--url", required=False, help="Target URL to request (for quick testing)")
p.add_argument("--method", default="GET", help="HTTP method to use (default: GET)")
p.add_argument(
"--proxies",
help="Path to proxies.txt file (one proxy URL per line). If omitted, reads PROXY_LIST env (comma-separated)",
)
p.add_argument("--strategy", choices=["round_robin", "random"], default="round_robin")
p.add_argument("--retries", type=int, default=3, help="Max retry attempts across proxies")
p.add_argument(
"--timeout",
type=float,
default=15.0,
help="Read timeout in seconds (connect timeout is fixed at 10s in this simple CLI)",
)
p.add_argument(
"--rotate-on",
default=",".join(str(s) for s in sorted(DEFAULT_ROTATE_ON_STATUS)),
help="Comma-separated status codes that trigger rotation",
)
p.add_argument(
"--no-sticky",
action="store_true",
help="Disable sticky-per-host behavior",
)
p.add_argument(
"--healthcheck",
action="store_true",
help="Run a healthcheck across proxies and print a summary before the request",
)
p.add_argument(
"--debug",
action="store_true",
help="Enable verbose logging",
)
return p
def _load_proxies_from_args(args: argparse.Namespace) -> List[Proxy]:
if args.proxies:
return parse_proxy_file(args.proxies)
env = os.getenv("PROXY_LIST", "").strip()
if env:
return [Proxy(url=p.strip()) for p in env.split(",") if p.strip()]
raise SystemExit("No proxies provided. Use --proxies or set PROXY_LIST env.")
def main_cli() -> None:
parser = build_arg_parser()
args = parser.parse_args()
logging.basicConfig(
level=logging.DEBUG if args.debug else logging.INFO,
format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
log = logging.getLogger("proxy_rotator")
proxies = _load_proxies_from_args(args)
rotate_on_set = {int(x.strip()) for x in args.rotate_on.split(",") if x.strip()}
rotator = ProxyRotator(
proxies=proxies,
strategy=args.strategy,
max_retries=args.retries,
timeout=(10.0, args.timeout),
rotate_on_status=rotate_on_set,
user_agents=DEFAULT_USER_AGENTS,
sticky_per_host=not args.no_sticky,
circuit_threshold=2,
circuit_cooldown=60.0,
backoff_base=0.5,
backoff_max=5.0,
healthcheck_url=DEFAULT_HEALTHCHECK_URL,
logger=log,
)
if args.healthcheck:
results = rotator.health_check()
total = len(results)
healthy = sum(1 for ok in results.values() if ok)
log.info("Healthcheck: %s/%s proxies OK", healthy, total)
for url, ok in results.items():
log.info(" %-40s %s", url, "OK" if ok else "BAD")
if args.url:
try:
resp = rotator.request(args.method.upper(), args.url)
print(f"Status: {resp.status_code}")
text = resp.text
preview = text[:500].replace("\n", " ")
print(f"Body preview (first 500 chars):\n{preview}")
except Exception as e:
log.error("Request failed: %s", e)
raise
else:
log.info("No --url provided; nothing else to do. Add --url to test a request.")
Punto de entrada
Es lo que pone en marcha tu rotador. Carga proxies, los verifica (si lo requiere) y genera requests.
if __name__ == "__main__":
main_cli()
Ejecutar un rotador de proxy
Puede ejecutar su rotador desde una línea de comando usando el siguiente comando:
python proxy_rotator.py --proxies ./proxies.txt --url http://api.ipify.org
Cambiar ./proxies.txt al nombre real de su archivo con proxies y – – URL al objetivo URL de un sitio web que desea extraer o probar.
Además, puedes agregar algunas banderas más:
– – estrategia aleatoria – para que un rotador seleccione un proxy de un grupo al azar
– – reintentos 5 – puedes establecer otro número de reintentos si es necesario
– – control de salud – para probar los servidores proxy de antemano
– – no pegajoso – para deshabilitar los servidores proxy fijos. Úselo solo si no necesita mantener la misma sesión
– – depurar – Verá registros detallados que le ayudarán a solucionar problemas.
Nota: a veces, si tiene instalado Python 3, es posible que necesite usar “python3 proxy_rotator.py –-proxies ./proxies.txt –-url http://api.ipify.org” como comando para ejecutar un rotador.
Conclusión
Construir un rotador proxy no es largo ni difícil; sin embargo, le ayuda a aprovechar al máximo un grupo de proxy. Un buen rotador que incluya lógica de reintento, verificación de estado y pausas de tiempo, junto con proxies confiables, puede llevar su web scraping o multicuenta al siguiente nivel. En cuanto al IPs confiable y de origen ético, DataImpulse está aquí para ayudarlo. Con nosotros, puede obtener más de 90 millones de IPs desde 195 ubicaciones para garantizar una conectividad interregional ininterrumpida. Contáctenos en [email protected] o presione el “Pruébalo ahora” botón para comenzar.