In this Article
Seja web scraping ou testes de sites, a necessidade de proxies é inegável; no entanto, você precisa de vários endereços diferentes para garantir um acesso consistente. Ao mesmo tempo, você não alternará entre proxies manualmente. O rotador proxy é a resposta nesta situação. Neste artigo, estamos construindo um rotador de proxy personalizado usando Python – do zero.
Por que é essencial construir um bom rotador de proxy
Um rotador percorre automaticamente um pool de proxy. Ele atribui outro endereço IP a cada solicitação ou em um intervalo de tempo definido. O servidor de destino não vê que requests vem da mesma fonte e o tráfego parece originar-se de endereços e locais diferentes, para que você possa passar despercebido pelos sistemas anti-bot. É por isso que construir um rotador proxy está no centro da raspagem ou da multicontabilidade. O rotador é responsável por manter um fluxo constante de autenticação, alinhando-se às diretrizes de tráfego do site e acessando conteúdo específico da região, especialmente em projetos de grande escala com múltiplas ameaças de conexão.
Além disso, existem algumas vantagens não tão óbvias em usar um rotador proxy. Por exemplo, aumenta a sua privacidade e anonimato, pois com o seu IP em constante mudança, é mais difícil para sites e rastreadores traçarem o perfil de suas atividades. Os hackers estão um passo além de você, pois com seu endereço real oculto, é mais difícil para os golpistas que visam dispositivos específicos descobrirem você.
Além disso, muitos sites oferecem resultados diferentes com base na sua localização ou preferências. Ao usar vários IPs, você obtém dados mais diversificados e precisos. A rotação também impede a visualização de conteúdo personalizado, caso você não precise dele.
Rotador proxy confiável: seus componentes
Além de simplesmente usar um novo IP para cada solicitação, você deve incluir mais alguns recursos, para que um rotador de proxy realmente beneficie você e não se torne uma fonte de problemas:
- Lógica de repetição – caso um proxy não funcione, o rotador usa outro;
- Rotação dos cabeçalhos User-Agent, para que seu requests não pareça vir da mesma fonte;
- Lidar com códigos de status HTTP “ruins” como 403 (“Proibido”), 429 (“Muitos requests”), 500 (“Erro interno do servidor”);
- Desativando proxies, que continuam falhando;
- Adicionando pequenos atrasos entre requests para evitar o acionamento de sistemas anti-bot;
- Logging – para que, em caso de nova tentativa, você veja o motivo e qual proxy e User-Agent foram utilizados;
- Verificação de integridade antes da raspagem – para testar todos os proxies antes de começar e ter certeza de que funcionam;
- Disjuntor para desabilitar proxies mortos por algum tempo;
- Sessões fixas por host para que, dentro de uma sessão, o domínio veja o mesmo IP, caso você precise (para contas múltiplas ou algo assim, quando você precisar manter a mesma sessão).
Manteremos esses pontos em mente ao criar nosso rotador de proxy.
Construindo um rotador proxy usando Python
Criaremos um rotador proxy em um único arquivo. Em nosso exemplo, nós o nomeamos proxy_rotator.py. Abaixo, você encontrará todos os trechos de código junto com uma explicação do que eles fazem. Você pode copiá-los e colá-los. Construímos código sem codificar dados, como credenciais, para que você não precise substituir nada mais tarde.
Começando
- Instale Python de seu site oficial (ou atualize-o). Você precisa da versão 3.8 ou superior.
- Execute o seguinte comando para instalar o requests biblioteca. Precisaremos dele para fazer HTTP requests.
pip install requests
- Obtenha uma lista de proxies na página do plano necessário. Você pode copiá-los e colá-los manualmente em um arquivo .txt do seu Painel DataImpulse ou baixe-os. Nomeamos nosso arquivo proxies.txt. Deveria ser assim:
http://login:password.host:port
https://login:password.host:port
Você pode alterar o formato do proxy, se necessário.
Importações e configuração
Primeiro, devemos trazer ferramentas chamadas “módulos” que iremos utilizar.
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
solicitar envia web requests, aleatório escolhe proxies e cabeçalhos e threading é responsável por lidar com vários threads de conexão.
Configurações padrão
A próxima parte cumpre várias funções cruciais: alterna os Agentes do Usuário, muda para outro proxy no caso de um código de erro e verifica se um 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"
Você pode substituir ou adicionar agentes de usuário e usar outro site para uma verificação de integridade, se desejar. No nosso caso, usamos httpbin.org/ip. Outras opções decentes são https://httpbin.org/get (mostra cabeçalhos de solicitação, bem como IP), https://api.ipify.org (ou https://api.ipify.org?format=json para devolver seu IP em JSON), https://ifconfig.me ou https://ifconfig.me/all.json (retorna cabeçalhos + IP em JSON), ou http://ip-api.com/json/ (mostra país, cidade, ISP e IP).
Objeto proxy
Esta parte define como os proxies são armazenados.
@dataclass(frozen=True)
class Proxy:
url: str
label: str = ""
def to_requests_proxies(self) -> Dict[str, str]:
return {"http": self.url, "https": self.url}
Novamente, você não precisa substituir nada aqui.
Lendo um arquivo proxy
Esta parte lê proxies de um .TXT arquivo.
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
Atraso na nova tentativa
Este trecho de código projeta uma lógica de nova tentativa. Com ele, um rotador espera mais tempo cada vez que uma solicitação falha para evitar limitações e bloqueios de serviço.
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)
A classe do rotador de proxy
O coração do seu rotador. Ele escolhe um proxy, tenta requests novamente e rastreia falhas. Ele também decide sobre uma estratégia – usar proxies aleatoriamente ou percorrer a lista, definir tempos limite e cabeçalhos e usar o mesmo proxy para um site. Ele envia requests, marca os proxies como permitidos ou bloqueados e executa verificações de integridade.
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)
Opções de linha de comando
Você precisa desta peça para executar um rotador com um comando de linha única via 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.")
Ponto de entrada
É o que inicia o seu rotador. Ele carrega proxies, verifica-os (se necessário) e cria requests.
if __name__ == "__main__":
main_cli()
Executando um rotador de proxy
Você pode executar seu rotador a partir de uma linha de comando usando o seguinte comando:
python proxy_rotator.py --proxies ./proxies.txt --url http://api.ipify.org
Mudar ./proxies.txt ao nome real do seu arquivo com proxies e – – URL para o destino URL de um site que você deseja copiar ou testar.
Além disso, você pode adicionar mais alguns sinalizadores:
– – estratégia aleatória – para que um rotador selecione um proxy de um pool aleatoriamente
– – novas tentativas 5 – você pode definir outro número de tentativas, se necessário
– – verificação de saúde – para testar proxies de antemão
– – não pegajoso – para desativar proxies fixos. Use apenas se você não precisar manter a mesma sessão
– – depurar – você verá registros detalhados, que o ajudarão a solucionar problemas.
Observação: às vezes, se você tiver Python 3 instalado, pode ser necessário usar “python3 proxy_rotator.py –-proxies ./proxies.txt –-url http://api.ipify.org” como o comando para executar um rotador.
Conclusão
Construir um rotador proxy não é longo nem difícil; no entanto, ajuda você a aproveitar ao máximo um pool de proxy. Um bom rotador que apresenta lógica de repetição, verificação de integridade e pausas de tempo, emparelhado com proxies confiáveis, pode levar seu web scraping ou multicontas para o próximo nível. Quanto ao IPs confiável e de origem ética, o DataImpulse está aqui para ajudá-lo. Conosco, você pode obter mais de 90 milhões de IPs em 195 locais para garantir conectividade inter-regional ininterrupta. Contate-nos em [email protected] ou pressione o “Tente agora” botão para começar.