Rotação de proxies

A rotação de proxies é o processo automatizado de percorrer uma lista de endereços IP, ou proxies, para mascarar a origem das requisições de rede, contornar rate limits e manter o anonimato. Essa técnica é fundamental para operações que exigem um alto volume de requisições, como web scraping, pesquisa de mercado e verificação de anúncios, nas quais o uso constante de um único endereço IP levaria a rate limiting, CAPTCHAs ou banimentos diretos.

Por que rotacionar proxies?

O objetivo principal da rotação de proxies é distribuir o tráfego de rede entre vários endereços IP. Sites e serviços online costumam usar mecanismos de detecção sofisticados para identificar e bloquear requisições automatizadas originadas de um único endereço IP ou de uma faixa pequena de IPs. Com a rotação, cada requisição, ou série de requisições, parece vir de um local e de um dispositivo diferentes, o que torna significativamente mais difícil para os sistemas-alvo identificar e bloquear o cliente.

Principais benefícios da rotação de proxies:
* Contornar rate limits: Muitos serviços limitam o número de requisições que um IP pode fazer em um intervalo de tempo específico. A rotação permite ultrapassar esses limites.
* Evitar banimentos de IP: Atividade contínua a partir de um único IP pode levar a um ban. Rotacionar IPs reduz esse risco.
* Manter o anonimato: Ofusca o IP real do cliente, aumentando a privacidade.
* Acessar conteúdo com restrição geográfica: Usando proxies de diferentes localizações geográficas, é possível acessar conteúdo restrito a regiões específicas.
* Distribuição de carga: Espalha a carga entre vários pontos de saída da rede.

Um "pool de proxies" é um conjunto de servidores proxy disponíveis a partir do qual um endereço IP é selecionado para cada requisição. Uma rotação eficaz depende da gestão desse pool e da implementação de um algoritmo de seleção adequado.

Algoritmos de rotação de proxies

Existem vários algoritmos de rotação de proxies, cada um com vantagens e desvantagens específicas conforme o caso de uso.

Rotação sequencial simples

Esse algoritmo percorre o pool de proxies em uma ordem fixa. Cada requisição usa o próximo proxy da lista, voltando ao início assim que chega ao fim.

Características:
* Previsível: A sequência de proxies é conhecida.
* Distribuição uniforme: Garante que todos os proxies sejam usados igualmente ao longo do tempo.
* Simplicidade: Fácil de implementar.

Limitações:
* Um proxy bloqueado pode travar a sequência até ser removido ou pulado manualmente.
* Menos eficaz contra detecção sofisticada, capaz de rastrear padrões sequenciais de uso de IP.

Exemplo de implementação em Python:

import itertools

class SequentialProxyRotator:
    def __init__(self, proxies):
        self.proxies = proxies
        self.proxy_cycle = itertools.cycle(self.proxies)

    def get_next_proxy(self):
        return next(self.proxy_cycle)

# Example Usage:
proxy_list = [
    "http://user:[email protected]:8000",
    "http://user:[email protected]:8000",
    "http://user:[email protected]:8000",
]
rotator = SequentialProxyRotator(proxy_list)

print(rotator.get_next_proxy()) # http://user:[email protected]:8000
print(rotator.get_next_proxy()) # http://user:[email protected]:8000
print(rotator.get_next_proxy()) # http://user:[email protected]:8000
print(rotator.get_next_proxy()) # http://user:[email protected]:8000 (cycles back)

Rotação aleatória

Nessa abordagem, um proxy é escolhido aleatoriamente do pool ativo para cada requisição.

Características:
* Imprevisível: Dificulta que os sistemas-alvo detectem um padrão.
* Simples: Fácil de implementar com funções da biblioteca padrão.

Limitações:
* Alguns proxies podem ser usados com mais frequência que outros, o que pode levar a uma detecção mais rápida ou ao esgotamento de IPs específicos.
* Um proxy ruim pode ser selecionado repetidamente se não for removido do pool.

Exemplo de implementação em Python:

import random

class RandomProxyRotator:
    def __init__(self, proxies):
        self.proxies = proxies

    def get_next_proxy(self):
        return random.choice(self.proxies)

# Example Usage:
proxy_list = [
    "http://user:[email protected]:8000",
    "http://user:[email protected]:8000",
    "http://user:[email protected]:8000",
]
rotator = RandomProxyRotator(proxy_list)

print(rotator.get_next_proxy()) # Randomly selected
print(rotator.get_next_proxy()) # Randomly selected

Rotação por tempo (variante LRU — Least Recently Used)

Esse algoritmo usa um proxy por um período definido ou por um número fixo de requisições antes de trocar para o próximo. Uma variante registra o horário do último uso de cada proxy e prioriza aqueles que não foram usados recentemente.

Características:
* Uso controlado: Garante que um proxy não seja usado em excesso em um intervalo curto.
* Menor rastro: Distribui a atividade de cada IP ao longo do tempo.

Limitações:
* Exige gerenciamento de estado para cada proxy (horário do último uso, contagem de requisições).
* A complexidade aumenta com a necessidade de gerenciar proxies ativos e inativos.

Exemplo de implementação em Python (LRU conceitual):

import time
from collections import deque

class TimedProxyRotator:
    def __init__(self, proxies, rotation_interval_seconds=60):
        self.proxies = deque(proxies)
        self.rotation_interval = rotation_interval_seconds
        self.current_proxy = None
        self.last_switch_time = 0

    def get_next_proxy(self):
        if self.current_proxy is None or (time.time() - self.last_switch_time) > self.rotation_interval:
            # Rotate proxy
            if self.current_proxy:
                self.proxies.append(self.current_proxy) # Put current back to end
            self.current_proxy = self.proxies.popleft()
            self.last_switch_time = time.time()
        return self.current_proxy

# Example Usage:
proxy_list = [
    "http://user:[email protected]:8000",
    "http://user:[email protected]:8000",
    "http://user:[email protected]:8000",
]
rotator = TimedProxyRotator(proxy_list, rotation_interval_seconds=10)

print(f"Initial: {rotator.get_next_proxy()}")
time.sleep(2)
print(f"Still same: {rotator.get_next_proxy()}")
# Simulate waiting for rotation_interval
# time.sleep(10)
# print(f"Rotated: {rotator.get_next_proxy()}")

Rotação adaptativa com monitoramento de saúde

Esse algoritmo avançado monitora o desempenho e a confiabilidade de cada proxy. Proxies que falham de forma consistente, estão lentos ou sofrem banimentos são removidos temporária ou permanentemente do pool ativo. Proxies novos ou recuperados são adicionados de volta.

Características:
* Alta confiabilidade: Prioriza proxies funcionais, minimizando falhas de requisição.
* Pool dinâmico: Adapta-se às mudanças na saúde dos proxies.
* Desempenho ótimo: Usa proxies mais rápidos e mais confiáveis.

Limitações:
* Bem mais complexo de implementar.
* Exige monitoramento contínuo e um mecanismo robusto de verificação de saúde.
* O overhead das verificações de saúde pode ser considerável com um pool grande.

Exemplo de implementação em Python (framework conceitual):

import time
import requests

class Proxy:
    def __init__(self, address):
        self.address = address
        self.is_healthy = True
        self.failure_count = 0
        self.last_used = 0
        self.response_times = []

    def mark_unhealthy(self):
        self.is_healthy = False
        self.failure_count += 1
        # Implement logic to temporarily disable or remove after N failures

    def mark_healthy(self):
        self.is_healthy = True
        self.failure_count = 0 # Reset on success

    def record_usage(self):
        self.last_used = time.time()

    def add_response_time(self, r_time):
        self.response_times.append(r_time)
        if len(self.response_times) > 10: # Keep last 10
            self.response_times.pop(0)

    @property
    def avg_response_time(self):
        return sum(self.response_times) / len(self.response_times) if self.response_times else float('inf')

class HealthAwareProxyRotator:
    def __init__(self, proxy_addresses, max_failures=3):
        self.proxies = {addr: Proxy(addr) for addr in proxy_addresses}
        self.max_failures = max_failures
        self.active_proxies = deque([p for p in self.proxies.values() if p.is_healthy])
        self.inactive_proxies = []

    def get_next_proxy(self):
        if not self.active_proxies:
            self.attempt_reactivate_proxies()
            if not self.active_proxies:
                raise Exception("No healthy proxies available.")

        # Simple sequential or random from active, for demonstration
        proxy_obj = self.active_proxies.popleft()
        self.active_proxies.append(proxy_obj) # Put back for sequential-like rotation
        proxy_obj.record_usage()
        return proxy_obj.address

    def report_status(self, proxy_address, success, response_time=None):
        proxy_obj = self.proxies.get(proxy_address)
        if not proxy_obj:
            return

        if success:
            proxy_obj.mark_healthy()
            if response_time is not None:
                proxy_obj.add_response_time(response_time)
        else:
            proxy_obj.mark_unhealthy()
            if proxy_obj.failure_count >= self.max_failures and proxy_obj in self.active_proxies:
                self.active_proxies.remove(proxy_obj)
                self.inactive_proxies.append(proxy_obj)
                print(f"Proxy {proxy_address} moved to inactive due to {proxy_obj.failure_count} failures.")

    def attempt_reactivate_proxies(self):
        # Implement periodic health checks for inactive proxies
        # For simplicity, just move all inactive proxies back to active if they exist
        if self.inactive_proxies:
            print("Attempting to reactivate inactive proxies...")
            for proxy_obj in list(self.inactive_proxies): # Iterate copy to allow modification
                # In a real system, you'd perform a health check here
                # For this example, assume they become healthy after some time
                proxy_obj.mark_healthy()
                self.active_proxies.append(proxy_obj)
                self.inactive_proxies.remove(proxy_obj)
            print(f"Reactivated {len(self.active_proxies)} proxies.")

# Example Usage:
proxy_list = [
    "http://user:[email protected]:8000",
    "http://user:[email protected]:8000", # Assume this one fails
    "http://user:[email protected]:8000",
]
rotator = HealthAwareProxyRotator(proxy_list)

# Simulate requests
p1 = rotator.get_next_proxy()
print(f"Using {p1}")
rotator.report_status(p1, True, 0.5)

p2 = rotator.get_next_proxy()
print(f"Using {p2}")
rotator.report_status(p2, False) # Fail
rotator.report_status(p2, False) # Fail
rotator.report_status(p2, False) # Fail - Should be moved to inactive

p3 = rotator.get_next_proxy()
print(f"Using {p3}")
rotator.report_status(p3, True, 0.7)

p_next = rotator.get_next_proxy() # Should now skip p2
print(f"Next active: {p_next}")

# If all active proxies fail, it would attempt reactivation
# rotator.report_status(p1, False, 0.5)
# rotator.report_status(p1, False, 0.5)
# rotator.report_status(p1, False, 0.5)
# rotator.report_status(p3, False, 0.5)
# rotator.report_status(p3, False, 0.5)
# rotator.report_status(p3, False, 0.5)
# print(rotator.get_next_proxy()) # This would trigger reactivation attempt

Rotação aleatória ponderada

Esse algoritmo atribui um peso a cada proxy, influenciando sua probabilidade de ser selecionado. Os pesos podem se basear em fatores como taxa histórica de sucesso, tempo de resposta ou localização geográfica.

Características:
* Uso priorizado: Favorece proxies de alto desempenho ou proxies específicos.
* Flexível: Os pesos podem ser ajustados dinamicamente.

Limitações:
* Exige um mecanismo para determinar e atualizar os pesos.
* Menos eficaz se os pesos não forem mantidos com precisão.

Exemplo de implementação em Python:

import random

class WeightedRandomProxyRotator:
    def __init__(self, proxies_with_weights):
        # proxies_with_weights is a list of tuples: [("proxy_addr", weight), ...]
        self.proxy_addresses = [pw[0] for pw in proxies_with_weights]
        self.weights = [pw[1] for pw in proxies_with_weights]

    def get_next_proxy(self):
        return random.choices(self.proxy_addresses, weights=self.weights, k=1)[0]

# Example Usage:
weighted_proxy_list = [
    ("http://user:[email protected]:8000", 5), # High weight, used more often
    ("http://user:[email protected]:8000", 1), # Low weight
    ("http://user:[email protected]:8000", 3),
]
rotator = WeightedRandomProxyRotator(weighted_proxy_list)

# print(rotator.get_next_proxy()) # Will likely print 1.1.1.1 more often

Comparação dos algoritmos

Detalhes práticos de implementação

Gestão do pool de proxies

Um sistema robusto de rotação de proxies exige uma gestão eficaz do pool.
* Estrutura de dados: Um deque do módulo collections é adequado para rotação sequencial ou do tipo LRU, graças às operações eficientes de append e popleft. Para sistemas com monitoramento de saúde, um dicionário que mapeia endereços de proxy para objetos Proxy customizados (como no exemplo) é eficaz para armazenar metadados.
* Adicionar/remover proxies: O sistema deve suportar atualizações dinâmicas da lista de proxies sem interrupção.
* Validação inicial: Antes de adicionar proxies ao pool ativo, uma verificação inicial de saúde confirma seu funcionamento e desempenho.

Integração com bibliotecas de requisições

A maioria das bibliotecas de requisições HTTP suporta proxies. Em Python, a biblioteca requests é a mais usada.

import requests

def make_request_with_proxy(url, proxy_address):
    proxies = {
        "http": proxy_address,
        "https": proxy_address,
    }
    try:
        start_time = time.time()
        response = requests.get(url, proxies=proxies, timeout=10)
        end_time = time.time()
        response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx)
        return response.text, (end_time - start_time)
    except requests.exceptions.RequestException as e:
        print(f"Request failed with proxy {proxy_address}: {e}")
        return None, None

# Example using a rotator:
# rotator = HealthAwareProxyRotator(proxy_list)
# selected_proxy = rotator.get_next_proxy()
# content, r_time = make_request_with_proxy("http://example.com", selected_proxy)
# if content:
#     rotator.report_status(selected_proxy, True, r_time)
# else:
#     rotator.report_status(selected_proxy, False)

Tratamento de erros e retentativas

Quando um proxy falha, o sistema deve:
1. Marcar o proxy como não saudável: Atualizar seu status no pool.
2. Repetir com um novo proxy: Selecionar outro proxy do pool e tentar a requisição novamente.
3. Definir limites de retentativa: Evitar loops infinitos em falhas persistentes.

def robust_request(url, rotator, max_retries=3):
    for _ in range(max_retries):
        try:
            proxy_address = rotator.get_next_proxy()
            print(f"Attempting {url} with {proxy_address}")
            content, r_time = make_request_with_proxy(url, proxy_address)
            if content:
                rotator.report_status(proxy_address, True, r_time)
                return content
            else:
                rotator.report_status(proxy_address, False)
        except Exception as e:
            print(f"Error during request attempt: {e}")
            # The get_next_proxy itself might raise an exception if no healthy proxies
            pass
    raise Exception(f"Failed to fetch {url} after {max_retries} retries.")

# Example usage with the HealthAwareRotator and a dummy URL
# try:
#     final_content = robust_request("http://dummy-url-that-might-fail.com", rotator)
#     print("Request successful.")
# except Exception as e:
#     print(f"Final failure: {e}")

Considerações sobre concorrência

Em ambientes multithread ou assíncronos, o estado do rotador de proxies (o pool, o proxy atual, as métricas de saúde) precisa ser thread-safe.
* Locks: Use threading.Lock para proteger seções críticas ao atualizar as estruturas de dados compartilhadas do pool de proxies.
* Filas: queue.Queue pode gerenciar os proxies, permitindo que vários workers façam get e put de proxies com segurança.

import threading
import queue
import time

class ThreadSafeProxyRotator:
    def __init__(self, proxies):
        self.proxy_queue = queue.Queue()
        for p in proxies:
            self.proxy_queue.put(p)
        self.lock = threading.Lock()
        self.in_use = set() # Track proxies currently in use by a thread

    def get_proxy(self):
        with self.lock:
            if self.proxy_queue.empty() and not self.in_use:
                raise Exception("No proxies available in pool.")
            elif self.proxy_queue.empty(): # All proxies are currently in use
                # Implement waiting or re-adding used proxies here for a real system
                # For simplicity, just raise an error for now
                raise Exception("All proxies are currently in use.")

            proxy = self.proxy_queue.get()
            self.in_use.add(proxy)
            return proxy

    def return_proxy(self, proxy, is_healthy=True):
        with self.lock:
            if proxy in self.in_use:
                self.in_use.remove(proxy)
                if is_healthy:
                    self.proxy_queue.put(proxy) # Return to pool
                else:
                    print(f"Proxy {proxy} marked unhealthy and not returned to pool.")
            else:
                print(f"Attempted to return unknown or already returned proxy: {proxy}")

# Example of a worker thread
# def worker(rotator, thread_id):
#     try:
#         proxy = rotator.get_proxy()
#         print(f"Thread {thread_id} using {proxy}")
#         time.sleep(random.uniform(1, 3)) # Simulate work
#         success = random.choice([True, False]) # Simulate success/failure
#         rotator.return_proxy(proxy, success)
#         print(f"Thread {thread_id} finished with {proxy}, success: {success}")
#     except Exception as e:
#         print(f"Thread {thread_id} error: {e}")
#
# proxy_list = ["proxy1", "proxy2", "proxy3"]
# ts_rotator = ThreadSafeProxyRotator(proxy_list)
# threads = []
# for i in range(5):
#     t = threading.Thread(target=worker, args=(ts_rotator, i))
#     threads.append(t)
#     t.start()
#
# for t in threads:
#     t.join()