Integrazione dei proxy con Google Cloud Functions: configurazione per scraping e automazione

Google Cloud Functions è una piattaforma serverless per eseguire codice senza gestire server. Quando si lavora con il parsing, l'automazione delle richieste API o la raccolta di dati, è spesso necessario instradare il traffico attraverso un proxy per aggirare i blocchi, ruotare gli IP e targetizzare geograficamente. In questa guida esamineremo la configurazione del proxy in Cloud Functions utilizzando Python e Node.js con esempi pratici.

Perché utilizzare un proxy in Cloud Functions

Google Cloud Functions operano in un ambiente isolato con indirizzi IP condivisi dei data center di Google. Quando si effettuano richieste frequenti a API esterne o siti web, si presentano problemi:

• Blocchi per IP — molti servizi (Google, Facebook, marketplace) riconoscono il traffico dai data center e applicano limitazioni di velocità o blocchi totali.
• Limitazioni geografiche — per accedere a contenuti disponibili solo in determinati paesi (ad esempio, parsing dei prezzi regionali su Wildberries o Ozon).
• Limitazione di velocità — un indirizzo IP può effettuare un numero limitato di richieste al minuto. I proxy consentono di distribuire il carico.
• Anonimato — nascondere la vera origine delle richieste quando si lavora con dati sensibili o intelligence competitiva.

Scenari tipici di utilizzo dei proxy in Cloud Functions:

• Parsing di marketplace (Wildberries, Ozon, Amazon) per monitorare i prezzi dei concorrenti
• Raccolta di dati dai social media (Instagram, TikTok) tramite API o web scraping
• Automazione della verifica degli annunci pubblicitari in diverse regioni
• Richieste massicce ai motori di ricerca (Google, Yandex) per analisi SEO
• Test delle funzionalità di geolocalizzazione delle applicazioni

Quali tipi di proxy sono adatti per Cloud Functions

La scelta del tipo di proxy dipende dall'attività, dal budget e dai requisiti di anonimato. Ecco un confronto delle principali opzioni:

Raccomandazioni per la scelta:

• Per il parsing di marketplace (Wildberries, Ozon, Amazon) — proxy residenziali con rotazione su richiesta, affinché ogni richiesta venga effettuata con un nuovo IP.
• Per richieste API (Google Maps API, OpenWeatherMap) — proxy datacenter con alta velocità, se non ci sono restrizioni severe sugli IP.
• Per social media (Instagram, TikTok) — proxy mobile, poiché hanno IP di operatori mobili e vengono raramente bloccati.
• Per SEO-parsing (Google, Yandex) — proxy residenziali con geolocalizzazione al necessario regione.

Configurazione del proxy in Python (requests, aiohttp)

Python è il linguaggio più popolare per Cloud Functions quando si lavora con il parsing e l'automazione. Esaminiamo l'integrazione del proxy con le librerie requests (richieste sincrone) e aiohttp (richieste asincrone).

Esempio con la libreria requests (proxy HTTP)

import requests
import os

def parse_with_proxy(request):
    # Otteniamo i dati del proxy dalle variabili d'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')
    
    # Formiamo l'URL del proxy con autenticazione
    proxy_url = f"http://{proxy_user}:{proxy_pass}@{proxy_host}:{proxy_port}"
    
    proxies = {
        'http': proxy_url,
        'https': proxy_url
    }
    
    try:
        # Effettuiamo una richiesta tramite il proxy con timeout
        response = requests.get(
            'https://api.example.com/data',
            proxies=proxies,
            timeout=10,
            headers={'User-Agent': 'Mozilla/5.0'}
        )
        
        # Controlliamo lo stato della risposta
        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'Errore proxy: {str(e)}'}
    except requests.exceptions.Timeout:
        return {'statusCode': 504, 'error': 'Timeout della richiesta'}
    except requests.exceptions.RequestException as e:
        return {'statusCode': 500, 'error': f'Richiesta fallita: {str(e)}'}

Punti importanti:

• Variabili d'ambiente — conserva i dati del proxy (host, porta, login, password) in Secret Manager o nelle variabili d'ambiente di Cloud Functions, non nel codice.
• Timeout — imposta sempre un timeout, per evitare che la funzione si blocchi in caso di problemi con il proxy.
• User-Agent — aggiungi l'intestazione User-Agent, affinché le richieste sembrino provenire da un vero browser.
• Gestione degli errori — gestisci separatamente ProxyError (problemi con il proxy) e Timeout (proxy lento).

Esempio con aiohttp (richieste asincrone)

Per compiti ad alta intensità (ad esempio, parsing di oltre 1000 pagine), utilizza richieste asincrone con 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': 'Connessione proxy fallita'}
        except asyncio.TimeoutError:
            return {'error': 'Timeout della richiesta'}

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'
    ]
    
    # Avviamo richieste asincrone in parallelo
    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}

L'approccio asincrono consente di effettuare 10-100 richieste parallele tramite il proxy, il che è critico per il parsing di grandi volumi di dati nel tempo di esecuzione limitato di Cloud Functions (fino a 9 minuti).

Lavorare con proxy SOCKS5

Alcuni fornitori di proxy offrono proxy SOCKS5 per un funzionamento più affidabile con il traffico UDP o per aggirare i blocchi. Per lavorare con SOCKS5 in Python, utilizza la libreria requests[socks]:

# Aggiungi a 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()}

Configurazione del proxy in Node.js (axios, node-fetch)

Node.js è il secondo linguaggio più popolare per Cloud Functions. Esaminiamo l'integrazione del proxy con le librerie axios e node-fetch.

Esempio con 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] // Restituiamo host:port senza password
    });
    
  } catch (error) {
    if (error.code === 'ECONNREFUSED') {
      res.status(502).json({ error: 'Connessione proxy rifiutata' });
    } else if (error.code === 'ETIMEDOUT') {
      res.status(504).json({ error: 'Timeout proxy' });
    } else {
      res.status(500).json({ error: error.message });
    }
  }
};

Dipendenze per package.json:

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

Esempio con 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 });
  }
};

Dipendenze per SOCKS5:

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

Autenticazione del proxy: login/password e whitelist IP

Ci sono due metodi principali di autenticazione quando si lavora con i proxy:

1. Autenticazione tramite login e password

Il metodo più comune è la trasmissione delle credenziali nell'URL del proxy:

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

Vantaggi: Facilità di configurazione, non richiede un IP sorgente fisso.

Svantaggi: Le credenziali vengono trasmesse in ogni richiesta, leggero overhead.

2. Autenticazione tramite whitelist IP

Alcuni fornitori consentono di aggiungere gli indirizzi IP di Cloud Functions nella whitelist. Problema: Cloud Functions utilizzano IP dinamici dal pool di Google Cloud.

Soluzione: Utilizza Cloud NAT per instradare il traffico in uscita tramite un IP esterno statico:

1. Crea una rete VPC e un subnet in Google Cloud
2. Configura Cloud NAT con riserva di un IP statico
3. Collega Cloud Functions al VPC Connector
4. Aggiungi l'IP statico nella whitelist del fornitore di proxy

Dopo la configurazione, il proxy non richiede login e password:

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

Raccomandazione: Per la maggior parte dei casi, utilizza l'autenticazione tramite login/password — è più semplice e non richiede costi aggiuntivi per Cloud NAT (da $0.044/ora + traffico).

Rotazione degli IP e gestione del pool di proxy

Quando si fa parsing di grandi volumi di dati, è critico utilizzare la rotazione degli IP per evitare blocchi. Ci sono diversi approcci:

1. Rotazione lato fornitore (Rotating Proxies)

Molti fornitori offrono rotating proxies — un singolo endpoint che cambia automaticamente IP ad ogni richiesta o secondo un timer:

# Un endpoint, l'IP cambia automaticamente
proxy_url = "http://username:password@rotating.proxy.com:8080"

# Ogni richiesta viene effettuata con un nuovo IP
for i in range(100):
    response = requests.get('https://api.ipify.org', proxies={'http': proxy_url})
    print(f"Richiesta {i}: IP = {response.text}")

Vantaggi: Non è necessario gestire manualmente il pool di proxy, integrazione semplice.

Svantaggi: Nessun controllo su IP specifici, può essere più costoso.

2. Gestione manuale del pool di proxy

Se hai un elenco di proxy statici, implementa la rotazione a livello di codice:

import random
import requests

# Pool di proxy (può essere caricato da 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 il proxy non funziona, proviamo un altro
            proxy = get_random_proxy()
            response = requests.get(url, proxies={'http': proxy, 'https': proxy})
            results.append({'url': url, 'status': response.status_code})
    
    return results

3. Proxy basati su sessione (sticky sessions)

Per compiti in cui è necessario mantenere un IP durante una sessione (ad esempio, autenticazione su un sito), utilizza l'ID di sessione nell'URL del proxy:

# Aggiungiamo l'ID di sessione nel login
import uuid

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

# Tutte le richieste con questo session_id passeranno attraverso un unico IP
session = requests.Session()
session.proxies = {'http': proxy_url, 'https': proxy_url}

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

# Richieste successive nella stessa sessione
session.get('https://example.com/dashboard')

Gestione degli errori e timeout

Quando si lavora con i proxy in Cloud Functions, è critico gestire correttamente gli errori per non perdere dati e non superare i limiti di tempo di esecuzione.

Tipi di errori e modi di gestione

Esempio di gestione complessiva degli errori

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

def fetch_with_retry(url, proxy_pool, max_retries=3):
    """
    Richiesta con retry automatico e cambio di proxy in caso di errori
    """
    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'}
            )
            
            # Controlliamo il codice di stato
            if response.status_code == 200:
                return {'success': True, 'data': response.text, 'proxy': proxy}
            elif response.status_code == 429:
                # Limitazione di velocità — aspettiamo e riproviamo
                time.sleep(2 ** attempt)  # Backoff esponenziale
                continue
            elif response.status_code == 403:
                # IP bloccato — cambiamo proxy
                continue
            else:
                return {'success': False, 'status': response.status_code}
                
        except ProxyError:
            # Proxy non funziona — proviamo il successivo
            print(f"Proxy {proxy} fallito, provando un altro...")
            continue
        except Timeout:
            # Timeout — proviamo con un altro proxy
            print(f"Timeout con {proxy}, riprovando...")
            continue
        except RequestException as e:
            # Altri errori
            print(f"Richiesta fallita: {e}")
            if attempt == max_retries - 1:
                return {'success': False, 'error': str(e)}
            continue
    
    return {'success': False, 'error': 'Massimo numero di tentativi superato'}

Impostazione dei timeout in Cloud Functions

Le Cloud Functions hanno un limite di tempo di esecuzione (di default 60 secondi, massimo 540 secondi). Tieni conto di questo quando imposti i timeout del proxy:

• Connection timeout — tempo per stabilire la connessione con il proxy (consigliato 5 secondi)
• Read timeout — tempo per ricevere la risposta dal server target tramite il proxy (consigliato 10-15 secondi)
• Total timeout — tempo totale per l'intera richiesta (deve essere inferiore al timeout della funzione)

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

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

Best practices e ottimizzazione delle prestazioni

Raccomandazioni per un funzionamento efficace con i proxy in Cloud Functions:

1. Utilizza variabili d'ambiente per le credenziali

Non memorizzare mai login e password dei proxy nel codice. Utilizza Secret Manager o variabili d'ambiente:

# Creazione di un segreto in Google Cloud
gcloud secrets create proxy-credentials \
    --data-file=proxy-config.json

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

# Lettura del segreto nel codice
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 i risultati del parsing

Utilizza Cloud Storage o Firestore per memorizzare i dati in cache, in modo da non effettuare richieste ripetute tramite il proxy:

import hashlib
from google.cloud import storage

def fetch_with_cache(url, proxy):
    # Generiamo una chiave di cache basata sull'URL
    cache_key = hashlib.md5(url.encode()).hexdigest()
    
    # Controlliamo la cache in Cloud Storage
    bucket = storage.Client().bucket('my-cache-bucket')
    blob = bucket.blob(f"cache/{cache_key}.json")
    
    if blob.exists():
        # Restituiamo i dati memorizzati in cache
        return json.loads(blob.download_as_text())
    
    # Effettuiamo una richiesta tramite il proxy
    response = requests.get(url, proxies={'http': proxy})
    data = response.json()
    
    # Salviamo nella cache
    blob.upload_from_string(json.dumps(data))
    
    return data

3. Monitoraggio e logging

Monitora le prestazioni del proxy e la frequenza degli errori tramite 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. Ottimizzazione del cold start

Le Cloud Functions hanno un ritardo di cold start. Minimizza le dipendenze e utilizza versioni minime delle librerie:

# requirements.txt — solo librerie necessarie
requests==2.31.0
# Evita librerie pesanti come pandas, se non critiche

Utilizza variabili globali per riutilizzare le connessioni:

# Creiamo la sessione una volta al cold start
session = requests.Session()
session.proxies = {'http': PROXY_URL, 'https': PROXY_URL}

def parse_data(request):
    # Riutilizziamo la sessione tra le chiamate
    response = session.get('https://api.example.com/data')
    return response.json()

5. Geolocalizzazione del proxy

Per compiti con targetizzazione geografica (ad esempio, parsing dei prezzi regionali), utilizza proxy con legame a un paese o città specifici:

# Esempio con proxy residenziali, dove è possibile specificare il paese nel login
proxy_url = f"http://username-country-ru:password@proxy.example.com:8080"

# Oppure utilizzare diversi endpoint per diversi paesi
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

Conclusione

L'integrazione del proxy con Google Cloud Functions apre ampie possibilità per il parsing, l'automazione e il lavoro con API senza limitazioni sugli IP. I punti principali da considerare: corretta gestione degli errori con logica di retry, utilizzo di timeout per prevenire blocchi, rotazione degli IP per evitare blocchi e memorizzazione sicura delle credenziali in Secret Manager.

Per la maggior parte delle attività di parsing e automazione, la scelta ottimale saranno proxy residenziali — forniscono un'elevata anonimato e una bassa percentuale di blocchi grazie all'utilizzo di IP di utenti reali. Per lavorare con social media e applicazioni mobili, raccomandiamo proxy mobili, che hanno IP di operatori mobili e vengono praticamente mai bloccati da piattaforme come Instagram e TikTok.

Con una corretta configurazione delle Cloud Functions con proxy, ottieni una soluzione scalabile ed economica per elaborare grandi volumi di dati senza la necessità di gestire l'infrastruttura.