Integração de Proxy com Google Cloud Functions: Configuração para Web Scraping e Automação

Google Cloud Functions é uma plataforma serverless para executar código sem gerenciar servidores. Ao trabalhar com scraping, automação de requisições API ou coleta de dados, muitas vezes é necessário roteamento de tráfego através de proxies para contornar bloqueios, rotação de IP e segmentação geográfica. Neste guia, vamos explorar a configuração de proxies no Cloud Functions usando Python e Node.js com exemplos práticos.

Por que usar proxies no Cloud Functions

Google Cloud Functions opera em um ambiente isolado com endereços IP compartilhados dos data centers do Google. Ao fazer requisições frequentes a APIs externas ou sites, surgem problemas:

• Bloqueios por IP — muitos serviços (Google, Facebook, marketplaces) reconhecem o tráfego de data centers e aplicam rate limiting ou bloqueio total.
• Restrições geográficas — para acessar conteúdo disponível apenas em determinados países (por exemplo, scraping de preços regionais no Wildberries ou Ozon).
• Rate limiting — um endereço IP pode fazer um número limitado de requisições por minuto. Proxies permitem distribuir a carga.
• Anonimato — ocultar a verdadeira origem das requisições ao trabalhar com dados sensíveis ou espionagem competitiva.

Cenários típicos de uso de proxies no Cloud Functions:

• Scraping de marketplaces (Wildberries, Ozon, Amazon) para monitoramento de preços de concorrentes
• Coleta de dados de redes sociais (Instagram, TikTok) através de API ou web scraping
• Automação de verificação de anúncios em diferentes regiões
• Requisições em massa a motores de busca (Google, Yandex) para análise de SEO
• Teste de funcionalidades de geolocalização em aplicativos

Quais tipos de proxies são adequados para Cloud Functions

A escolha do tipo de proxy depende da tarefa, orçamento e requisitos de anonimato. Aqui está uma comparação das principais opções:

Recomendações para escolha:

• Para scraping de marketplaces (Wildberries, Ozon, Amazon) — proxies residenciais com rotação por requisição, para que cada requisição use um novo IP.
• Para requisições API (Google Maps API, OpenWeatherMap) — proxies de data center com alta velocidade, se não houver restrições rígidas por IP.
• Para redes sociais (Instagram, TikTok) — proxies móveis, pois eles têm IPs de operadoras móveis e raramente são bloqueados.
• Para scraping de SEO (Google, Yandex) — proxies residenciais com vinculação geográfica à região desejada.

Configuração de proxy em Python (requests, aiohttp)

Python é a linguagem mais popular para Cloud Functions ao trabalhar com scraping e automação. Vamos considerar a integração de proxies com as bibliotecas requests (requisições síncronas) e aiohttp (requisições assíncronas).

Exemplo com a biblioteca requests (proxy HTTP)

import requests
import os

def parse_with_proxy(request):
    # Obtendo dados do proxy das variáveis de ambiente
    proxy_host = os.environ.get('PROXY_HOST', 'proxy.example.com')
    proxy_port = os.environ.get('PROXY_PORT', '8080')
    proxy_user = os.environ.get('PROXY_USER', 'username')
    proxy_pass = os.environ.get('PROXY_PASS', 'password')
    
    # Formando a URL do proxy com autenticação
    proxy_url = f"http://{proxy_user}:{proxy_pass}@{proxy_host}:{proxy_port}"
    
    proxies = {
        'http': proxy_url,
        'https': proxy_url
    }
    
    try:
        # Fazendo a requisição através do proxy com timeout
        response = requests.get(
            'https://api.example.com/data',
            proxies=proxies,
            timeout=10,
            headers={'User-Agent': 'Mozilla/5.0'}
        )
        
        # Verificando o status da resposta
        response.raise_for_status()
        
        return {
            'statusCode': 200,
            'body': response.json(),
            'ip_used': response.headers.get('X-Forwarded-For', 'unknown')
        }
        
    except requests.exceptions.ProxyError as e:
        return {'statusCode': 502, 'error': f'Erro de proxy: {str(e)}'}
    except requests.exceptions.Timeout:
        return {'statusCode': 504, 'error': 'Timeout da requisição'}
    except requests.exceptions.RequestException as e:
        return {'statusCode': 500, 'error': f'Requisição falhou: {str(e)}'}

Pontos importantes:

• Variáveis de ambiente — armazene os dados do proxy (host, porta, login, senha) no Secret Manager ou variáveis de ambiente do Cloud Functions, e não no código.
• Timeouts — sempre defina timeout, para que a função não fique travada em caso de problemas com o proxy.
• User-Agent — adicione o cabeçalho User-Agent, para que as requisições pareçam vir de um navegador real.
• Tratamento de erros — trate separadamente ProxyError (problemas com o proxy) e Timeout (proxy lento).

Exemplo com aiohttp (requisições assíncronas)

Para tarefas de alta carga (por exemplo, scraping de 1000+ páginas), use requisições assíncronas com aiohttp:

import aiohttp
import asyncio
import os

async def fetch_with_proxy(url, proxy_url):
    async with aiohttp.ClientSession() as session:
        try:
            async with session.get(
                url,
                proxy=proxy_url,
                timeout=aiohttp.ClientTimeout(total=10),
                headers={'User-Agent': 'Mozilla/5.0'}
            ) as response:
                return await response.text()
        except aiohttp.ClientProxyConnectionError:
            return {'error': 'Falha na conexão do proxy'}
        except asyncio.TimeoutError:
            return {'error': 'Timeout da requisição'}

def parse_multiple_urls(request):
    proxy_url = f"http://{os.environ['PROXY_USER']}:{os.environ['PROXY_PASS']}@{os.environ['PROXY_HOST']}:{os.environ['PROXY_PORT']}"
    
    urls = [
        'https://example.com/page1',
        'https://example.com/page2',
        'https://example.com/page3'
    ]
    
    # Executando requisições assíncronas em paralelo
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    
    tasks = [fetch_with_proxy(url, proxy_url) for url in urls]
    results = loop.run_until_complete(asyncio.gather(*tasks))
    
    return {'statusCode': 200, 'results': results}

A abordagem assíncrona permite fazer 10-100 requisições paralelas através do proxy, o que é crítico para scraping de grandes volumes de dados dentro do tempo de execução limitado do Cloud Functions (até 9 minutos).

Trabalhando com proxies SOCKS5

Alguns provedores de proxy oferecem proxies SOCKS5 para um funcionamento mais confiável com tráfego UDP ou para contornar bloqueios. Para trabalhar com SOCKS5 em Python, use a biblioteca requests[socks]:

# Adicione ao requirements.txt:
# requests[socks]

import requests

def use_socks5_proxy(request):
    proxy_url = f"socks5://{os.environ['PROXY_USER']}:{os.environ['PROXY_PASS']}@{os.environ['PROXY_HOST']}:{os.environ['PROXY_PORT']}"
    
    proxies = {
        'http': proxy_url,
        'https': proxy_url
    }
    
    response = requests.get(
        'https://api.ipify.org?format=json',
        proxies=proxies,
        timeout=10
    )
    
    return {'statusCode': 200, 'ip': response.json()}

Configuração de proxy em Node.js (axios, node-fetch)

Node.js é a segunda linguagem mais popular para Cloud Functions. Vamos considerar a integração de proxies com as bibliotecas axios e node-fetch.

Exemplo com axios

const axios = require('axios');
const HttpsProxyAgent = require('https-proxy-agent');

exports.parseWithProxy = async (req, res) => {
  const proxyUrl = `http://${process.env.PROXY_USER}:${process.env.PROXY_PASS}@${process.env.PROXY_HOST}:${process.env.PROXY_PORT}`;
  
  const agent = new HttpsProxyAgent(proxyUrl);
  
  try {
    const response = await axios.get('https://api.example.com/data', {
      httpsAgent: agent,
      timeout: 10000,
      headers: {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
      }
    });
    
    res.status(200).json({
      success: true,
      data: response.data,
      proxyUsed: proxyUrl.split('@')[1] // Retornando host:porta sem senha
    });
    
  } catch (error) {
    if (error.code === 'ECONNREFUSED') {
      res.status(502).json({ error: 'Conexão do proxy recusada' });
    } else if (error.code === 'ETIMEDOUT') {
      res.status(504).json({ error: 'Timeout do proxy' });
    } else {
      res.status(500).json({ error: error.message });
    }
  }
};

Dependências para package.json:

{
  "dependencies": {
    "axios": "^1.6.0",
    "https-proxy-agent": "^7.0.2"
  }
}

Exemplo com node-fetch e SOCKS5

const fetch = require('node-fetch');
const { SocksProxyAgent } = require('socks-proxy-agent');

exports.fetchWithSocks5 = async (req, res) => {
  const proxyUrl = `socks5://${process.env.PROXY_USER}:${process.env.PROXY_PASS}@${process.env.PROXY_HOST}:${process.env.PROXY_PORT}`;
  
  const agent = new SocksProxyAgent(proxyUrl);
  
  try {
    const response = await fetch('https://api.ipify.org?format=json', {
      agent,
      timeout: 10000
    });
    
    const data = await response.json();
    
    res.status(200).json({
      success: true,
      yourIP: data.ip
    });
    
  } catch (error) {
    res.status(500).json({ error: error.message });
  }
};

Dependências para SOCKS5:

{
  "dependencies": {
    "node-fetch": "^2.7.0",
    "socks-proxy-agent": "^8.0.2"
  }
}

Autenticação de proxy: login/senha e whitelist de IP

Existem dois métodos principais de autenticação ao trabalhar com proxies:

1. Autenticação por login e senha

O método mais comum é passar as credenciais na URL do proxy:

http://username:password@proxy.example.com:8080

Vantagens: Facilidade de configuração, não requer um IP fixo de origem.

Desvantagens: As credenciais são passadas em cada requisição, pequeno overhead.

2. Autenticação por whitelist de IP

Alguns provedores permitem adicionar endereços IP do Cloud Functions à whitelist. O problema: o Cloud Functions usa IPs dinâmicos de um pool do Google Cloud.

Solução: Use Cloud NAT para rotear o tráfego de saída através de um IP externo estático:

1. Crie uma rede VPC e subnet no Google Cloud
2. Configure o Cloud NAT com reserva de um IP estático
3. Conecte o Cloud Functions ao VPC Connector
4. Adicione o IP estático à whitelist do provedor de proxy

Após a configuração, o proxy não requer login e senha:

proxies = {
    'http': 'http://proxy.example.com:8080',
    'https': 'http://proxy.example.com:8080'
}

Recomendação: Para a maioria dos casos, use autenticação por login/senha — é mais simples e não requer custos adicionais com Cloud NAT (a partir de $0.044/hora + tráfego).

Rotação de IP e gerenciamento de pool de proxies

Ao fazer scraping de grandes volumes de dados, é crítico usar rotação de IP para evitar bloqueios. Existem várias abordagens:

1. Rotação do lado do provedor (Rotating Proxies)

Muitos provedores oferecem rotating proxies — um único endpoint que muda automaticamente o IP a cada requisição ou por um timer:

# Um endpoint, IP muda automaticamente
proxy_url = "http://username:password@rotating.proxy.com:8080"

# Cada requisição vai com um novo IP
for i in range(100):
    response = requests.get('https://api.ipify.org', proxies={'http': proxy_url})
    print(f"Requisição {i}: IP = {response.text}")

Vantagens: Não é necessário gerenciar o pool de proxies manualmente, integração simples.

Desvantagens: Sem controle sobre IPs específicos, pode ser mais caro.

2. Gerenciamento manual do pool de proxies

Se você tiver uma lista de proxies estáticos, implemente a rotação no nível do código:

import random
import requests

# Pool de proxies (pode ser carregado do Secret Manager)
PROXY_POOL = [
    "http://user:pass@proxy1.example.com:8080",
    "http://user:pass@proxy2.example.com:8080",
    "http://user:pass@proxy3.example.com:8080",
]

def get_random_proxy():
    return random.choice(PROXY_POOL)

def parse_with_rotation(urls):
    results = []
    
    for url in urls:
        proxy = get_random_proxy()
        
        try:
            response = requests.get(
                url,
                proxies={'http': proxy, 'https': proxy},
                timeout=10
            )
            results.append({
                'url': url,
                'status': response.status_code,
                'proxy': proxy.split('@')[1]
            })
        except Exception as e:
            # Se o proxy não funcionar, tente outro
            proxy = get_random_proxy()
            response = requests.get(url, proxies={'http': proxy, 'https': proxy})
            results.append({'url': url, 'status': response.status_code})
    
    return results

3. Proxies baseados em sessão (sticky sessions)

Para tarefas que precisam manter um IP durante a sessão (por exemplo, autenticação em um site), use o ID da sessão na URL do proxy:

# Adicionando ID da sessão ao login
import uuid

session_id = str(uuid.uuid4())
proxy_url = f"http://username-session-{session_id}:password@proxy.example.com:8080"

# Todas as requisições com esse session_id irão através de um único IP
session = requests.Session()
session.proxies = {'http': proxy_url, 'https': proxy_url}

# Autenticação
session.post('https://example.com/login', data={'user': 'test', 'pass': '123'})

# Requisições subsequentes na mesma sessão
session.get('https://example.com/dashboard')

Tratamento de erros e timeouts

Ao trabalhar com proxies no Cloud Functions, é crítico tratar erros corretamente para não perder dados e não exceder os limites de tempo de execução.

Tipos de erros e formas de tratamento

Exemplo de tratamento abrangente de erros

import requests
import time
from requests.exceptions import ProxyError, Timeout, RequestException

def fetch_with_retry(url, proxy_pool, max_retries=3):
    """
    Requisição com retry automático e troca de proxy em caso de erros
    """
    for attempt in range(max_retries):
        proxy = random.choice(proxy_pool)
        
        try:
            response = requests.get(
                url,
                proxies={'http': proxy, 'https': proxy},
                timeout=10,
                headers={'User-Agent': 'Mozilla/5.0'}
            )
            
            # Verificando o código de status
            if response.status_code == 200:
                return {'success': True, 'data': response.text, 'proxy': proxy}
            elif response.status_code == 429:
                # Rate limiting — esperar e tentar novamente
                time.sleep(2 ** attempt)  # Exponential backoff
                continue
            elif response.status_code == 403:
                # IP bloqueado — mudar proxy
                continue
            else:
                return {'success': False, 'status': response.status_code}
                
        except ProxyError:
            # Proxy não funciona — tentar o próximo
            print(f"Proxy {proxy} falhou, tentando outro...")
            continue
        except Timeout:
            # Timeout — tentar com outro proxy
            print(f"Timeout com {proxy}, tentando novamente...")
            continue
        except RequestException as e:
            # Outros erros
            print(f"Requisição falhou: {e}")
            if attempt == max_retries - 1:
                return {'success': False, 'error': str(e)}
            continue
    
    return {'success': False, 'error': 'Número máximo de tentativas excedido'}

Configuração de timeouts no Cloud Functions

Cloud Functions têm um limite de tempo de execução (por padrão 60 segundos, máximo 540 segundos). Considere isso ao configurar os timeouts do proxy:

• Connection timeout — tempo para estabelecer a conexão com o proxy (recomendado 5 segundos)
• Read timeout — tempo para receber a resposta do servidor alvo através do proxy (recomendado 10-15 segundos)
• Total timeout — tempo total para toda a requisição (deve ser menor que o timeout da função)

# Python: timeouts separados
response = requests.get(
    url,
    proxies=proxies,
    timeout=(5, 15)  # (connect timeout, read timeout)
)

# Node.js com axios
const response = await axios.get(url, {
  httpsAgent: agent,
  timeout: 10000  // total timeout em milissegundos
});

Melhores práticas e otimização de desempenho

Recomendações para trabalhar de forma eficaz com proxies no Cloud Functions:

1. Use variáveis de ambiente para credenciais

Nunca armazene logins e senhas de proxies no código. Use o Secret Manager ou variáveis de ambiente:

# Criando um segredo no Google Cloud
gcloud secrets create proxy-credentials \
    --data-file=proxy-config.json

# Fornecendo acesso ao Cloud Functions
gcloud secrets add-iam-policy-binding proxy-credentials \
    --member=serviceAccount:PROJECT_ID@appspot.gserviceaccount.com \
    --role=roles/secretmanager.secretAccessor

# Lendo o segredo no código
from google.cloud import secretmanager
import json

def get_proxy_config():
    client = secretmanager.SecretManagerServiceClient()
    name = f"projects/{PROJECT_ID}/secrets/proxy-credentials/versions/latest"
    response = client.access_secret_version(request={"name": name})
    return json.loads(response.payload.data.decode('UTF-8'))

2. Cache os resultados do scraping

Use Cloud Storage ou Firestore para armazenar dados em cache, para não fazer requisições repetidas através do proxy:

import hashlib
from google.cloud import storage

def fetch_with_cache(url, proxy):
    # Gerando a chave do cache com base na URL
    cache_key = hashlib.md5(url.encode()).hexdigest()
    
    # Verificando o cache no Cloud Storage
    bucket = storage.Client().bucket('my-cache-bucket')
    blob = bucket.blob(f"cache/{cache_key}.json")
    
    if blob.exists():
        # Retornando dados em cache
        return json.loads(blob.download_as_text())
    
    # Fazendo a requisição através do proxy
    response = requests.get(url, proxies={'http': proxy})
    data = response.json()
    
    # Salvando no cache
    blob.upload_from_string(json.dumps(data))
    
    return data

3. Monitoramento e logging

Monitore o desempenho do proxy e a frequência de erros através do Cloud Logging:

import logging
import time

def fetch_with_logging(url, proxy):
    start_time = time.time()
    
    try:
        response = requests.get(url, proxies={'http': proxy}, timeout=10)
        duration = time.time() - start_time
        
        logging.info({
            'url': url,
            'proxy': proxy.split('@')[1],
            'status': response.status_code,
            'duration': duration,
            'success': True
        })
        
        return response
        
    except Exception as e:
        duration = time.time() - start_time
        
        logging.error({
            'url': url,
            'proxy': proxy.split('@')[1],
            'error': str(e),
            'duration': duration,
            'success': False
        })
        
        raise

4. Otimização do cold start

Cloud Functions têm um atraso de cold start. Minimize as dependências e use versões mínimas das bibliotecas:

# requirements.txt — apenas bibliotecas necessárias
requests==2.31.0
# Evite bibliotecas pesadas como pandas, a menos que sejam críticas

Use variáveis globais para reutilizar conexões:

# Criando uma sessão uma vez no cold start
session = requests.Session()
session.proxies = {'http': PROXY_URL, 'https': PROXY_URL}

def parse_data(request):
    # Reutilizando a sessão entre chamadas
    response = session.get('https://api.example.com/data')
    return response.json()

5. Vinculação geográfica de proxies

Para tarefas com segmentação geográfica (por exemplo, scraping de preços regionais), use proxies vinculados a um país ou cidade específica:

# Exemplo com proxies residenciais, onde é possível especificar o país no login
proxy_url = f"http://username-country-ru:password@proxy.example.com:8080"

# Ou usar diferentes endpoints para diferentes países
PROXIES_BY_COUNTRY = {
    'RU': 'http://user:pass@ru.proxy.example.com:8080',
    'US': 'http://user:pass@us.proxy.example.com:8080',
    'DE': 'http://user:pass@de.proxy.example.com:8080'
}

def parse_by_country(country_code):
    proxy = PROXIES_BY_COUNTRY.get(country_code)
    response = requests.get('https://example.com', proxies={'http': proxy})
    return response.text

Conclusão

A integração de proxies com Google Cloud Functions abre amplas possibilidades para scraping, automação e trabalho com APIs sem restrições de IP. Os principais pontos a serem considerados: tratamento correto de erros com lógica de retry, uso de timeouts para evitar travamentos, rotação de IP para evitar bloqueios e armazenamento seguro de credenciais no Secret Manager.

Para a maioria das tarefas de scraping e automação, a escolha ideal são proxies residenciais — que oferecem alta anonimidade e baixo percentual de bloqueios devido ao uso de IPs de usuários reais. Para trabalhar com redes sociais e aplicativos móveis, recomendamos proxies móveis, que têm IPs de operadoras móveis e praticamente não são bloqueados por plataformas como Instagram e TikTok.

Com a configuração correta do Cloud Functions com proxies, você obtém uma solução escalável e econômica para processar grandes volumes de dados sem a necessidade de gerenciar a infraestrutura.