Come ruotare i proxy durante lo scraping di dati web

Per quanto questo approccio alla raccolta di dati possa sembrare valido, è disapprovato da molti siti web e ci sono conseguenze per chi procede con lo scraping, come il divieto di utilizzare il nostro IP.

Una nota positiva è che i servizi proxy aiutano a evitare queste conseguenze. Ci permettono di assumere un IP diverso mentre raccogliamo dati online e, per quanto possa sembrare sicuro, è meglio usare più proxy. L'uso di più proxy durante lo scraping fa sì che l'interazione con il sito web appaia casuale e aumenta la sicurezza.

Il sito web di destinazione (fonte) di questa guida è una libreria online. Imita un sito di e-commerce di libri. Su di esso sono presenti libri con nome, prezzo e disponibilità. Poiché questa guida non si concentra sull'organizzazione dei dati restituiti, ma sulla rotazione dei proxy, i dati restituiti saranno presentati solo nella console.

Preparazione dell'ambiente di lavoro e integrazione dei proxy

Installiamo e importiamo alcuni moduli Python nel nostro file prima di iniziare a codificare le funzioni che ci aiuteranno a ruotare i proxy e a fare lo scraping del sito web.

pip install requests beautifulSoup4 lxml

3 dei 5 moduli Python necessari per questo script di scraping possono essere installati utilizzando il comando precedente. Requests ci permette di inviare richieste HTTP al sito web, beautifulSoup4 ci permette di estrarre le informazioni dall'HTML della pagina fornita da request e LXML è un parser HTML.

Inoltre, abbiamo bisogno del modulo di threading integrato per consentire il test multiplo dei proxy per vedere se funzionano e di json per leggere da un file JSON.

import requests
import threading
from requests.auth import HTTPProxyAuth
import json
from bs4 import BeautifulSoup
import lxml
import time

url_to_scrape = "https://books.toscrape.com"
valid_proxies = []
book_names = []
book_price = []
book_availability = []
next_button_link = ""

Fase 1: Verifica del proxy da un elenco di proxy

Costruire uno script di scraping che ruota i proxy significa avere un elenco di proxy tra cui scegliere durante la rotazione. Alcuni proxy richiedono l'autenticazione, altri no. Dobbiamo creare un elenco di dizionari con i dettagli del proxy, compresi il nome utente e la password del proxy se è necessaria l'autenticazione.

L'approccio migliore è quello di inserire le informazioni sul proxy in un file JSON separato, organizzato come quello qui sotto:

[
  {
    "proxy_address": "XX.X.XX.X:XX",
    "proxy_username": "",
    "proxy_password": ""
  },

  {
    "proxy_address": "XX.X.XX.X:XX",
    "proxy_username": "",
    "proxy_password": ""
  },
  {
    "proxy_address": "XX.X.XX.X:XX",
    "proxy_username": "",
    "proxy_password": ""
  },
  {
    "proxy_address": "XX.X.XX.X:XX",
    "proxy_username": "",
    "proxy_password": ""
  }
]

Nel campo "proxy_address", inserire l'indirizzo IP e la porta, separati da due punti. Nei campi "proxy_username" e "proxy_password", indicare il nome utente e la password per l'autorizzazione.

Qui sopra è riportato il contenuto di un file JSON con 4 proxy tra cui lo script può scegliere. Il nome utente e la password possono essere vuoti, indicando un proxy che non richiede l'autenticazione.

def verify_proxies(proxy:dict):
    try:
        if proxy['proxy_username'] != "" and  proxy['proxy_password'] != "":
            proxy_auth = HTTPProxyAuth(proxy['proxy_username'], proxy['proxy_password'])
            res = requests.get(
                url_to_scrape,
                auth = proxy_auth,
                proxies={
                "http" : proxy['proxy_address']
                }
            )
        else:
            res = requests.get(url_to_scrape, proxies={
                "http" : proxy['proxy_address'],
            })
        
        if res.status_code == 200:
            valid_proxies.append(proxy)
            print(f"Proxy Validated: {proxy['proxy_address']}")
            
    except:
        print("Proxy Invalidated, Moving on")

Come precauzione, questa funzione assicura che i proxy forniti siano attivi e funzionanti. Per ottenere questo risultato, è sufficiente scorrere ogni dizionario del file JSON, inviare una richiesta GET al sito web e, se viene restituito un codice di stato 200, aggiungere quel proxy all'elenco dei proxy_validi, una variabile creata in precedenza per contenere i proxy funzionanti dall'elenco del file. Se la chiamata non ha successo, l'esecuzione continua.

Fase 2: Invio della richiesta di scraping del web

Poiché beautifulSoup ha bisogno del codice HTML del sito web per estrarre i dati di cui abbiamo bisogno, abbiamo creato request_function(), che prende l'URL e il proxy scelto e restituisce il codice HTML come testo. La variabile proxy ci consente di instradare la richiesta attraverso diversi proxy, quindi di ruotare il proxy.

def request_function(url, proxy):
    try:
        if proxy['proxy_username'] != "" and  proxy['proxy_password'] != "":
            proxy_auth = HTTPProxyAuth(proxy['proxy_username'], proxy['proxy_password'])
            response = requests.get(
                url,
                auth = proxy_auth,
                proxies={
                "http" : proxy['proxy_address']
                }
            )
        else:
            response = requests.get(url, proxies={
                "http" : proxy['proxy_address']
            })
        
        if response.status_code == 200:
            return response.text

    except Exception as err:
        print(f"Switching Proxies, URL access was unsuccessful: {err}")
        return None

Fase 3: Estrazione dei dati dal sito web di destinazione

data_extract() estrae i dati necessari dal codice HTML fornito. Raccoglie l'elemento HTML che contiene le informazioni sul libro, come il nome del libro, il prezzo e la disponibilità. Estrae anche il link per la pagina successiva.

Questo è particolarmente complicato perché il link è dinamico, quindi dobbiamo tenerne conto. Infine, esamina i libri ed estrae il nome, il prezzo e la disponibilità, quindi restituisce il link al pulsante successivo, che utilizzeremo per recuperare il codice HTML della pagina successiva.

def data_extract(response):
    soup = BeautifulSoup(response, "lxml")
    books = soup.find_all("li", class_="col-xs-6 col-sm-4 col-md-3 col-lg-3")
    next_button_link = soup.find("li", class_="next").find('a').get('href')
    next_button_link=f"{url_to_scrape}/{next_button_link}" if "catalogue" in next_button_link else f"{url_to_scrape}/catalogue/{next_button_link}"

    for each in books:
        book_names.append(each.find("img").get("alt"))
        book_price.append(each.find("p", class_="price_color").text)
        book_availability.append(each.find("p", class_="instock availability").text.strip())

    return next_button_link

Fase 4: Collegare tutto insieme

Per collegare tutto insieme, dobbiamo:

1. Caricare i dettagli del proxy dal file JSON. Avviare un thread per ogni proxy, utilizzando threading.Thread(). Questo ci aiuterà a testare più proxy alla volta. I proxy validi vengono aggiunti a valid_proxies().
2. Caricare la homepage della sorgente utilizzando un proxy valido. Se un proxy non funziona, si usa quello successivo, per assicurarsi che la homepage venga caricata o non restituisca None prima di continuare l'esecuzione.
3. Poi scorriamo i proxy attivi, usiamo la funzione request_function() per creare una richiesta GET. E se abbiamo ricevuto una richiesta GET, raccogliamo i dati dal sito.
4. Finalmente, stampiamo i dati raccolti nella console.

with open("proxy-list.json") as json_file:
    proxies = json.load(json_file)
    for each in proxies:
        threading.Thread(target=verify_proxies, args=(each, )).start() 


time.sleep(4)

for i in range(len(valid_proxies)):
    response = request_function(url_to_scrape, valid_proxies[i])
    if response != None:
        next_button_link = data_extract(response)
        break
    else:
        continue

for proxy in valid_proxies:
   print(f"Using Proxy: {proxy['proxy_address']}")
   response = request_function(next_button_link, proxy)
   if response is not None:
       next_button_link = data_extract(response)
   else:
       continue


for each in range(len(book_names)):
    print(f"No {each+1}: Book Name: {book_names[each]} Book Price: {book_price[each]} and Availability {book_availability[each]}")

Codice completo

import requests
import threading
from requests.auth import HTTPProxyAuth
import json
from bs4 import BeautifulSoup
import time

url_to_scrape = "https://books.toscrape.com"
valid_proxies = []
book_names = []
book_price = []
book_availability = []
next_button_link = ""


def verify_proxies(proxy: dict):
   try:
       if proxy['proxy_username'] != "" and proxy['proxy_password'] != "":
           proxy_auth = HTTPProxyAuth(proxy['proxy_username'], proxy['proxy_password'])
           res = requests.get(
               url_to_scrape,
               auth=proxy_auth,
               proxies={
                   "http": proxy['proxy_address'],
               }
           )
       else:
           res = requests.get(url_to_scrape, proxies={
               "http": proxy['proxy_address'],
           })

       if res.status_code == 200:
           valid_proxies.append(proxy)
           print(f"Proxy Validated: {proxy['proxy_address']}")

   except:
       print("Proxy Invalidated, Moving on")


# Recupera l'elemento HTML di una pagina
def request_function(url, proxy):
   try:
       if proxy['proxy_username'] != "" and proxy['proxy_password'] != "":
           proxy_auth = HTTPProxyAuth(proxy['proxy_username'], proxy['proxy_password'])
           response = requests.get(
               url,
               auth=proxy_auth,
               proxies={
                   "http": proxy['proxy_address'],
               }
           )
       else:
           response = requests.get(url, proxies={
               "http": proxy['proxy_address'],
           })

       if response.status_code == 200:
           return response.text

   except Exception as err:
       print(f"Switching Proxies, URL access was unsuccessful: {err}")
       return None


# Raschiamento
def data_extract(response):
   soup = BeautifulSoup(response, "lxml")
   books = soup.find_all("li", class_="col-xs-6 col-sm-4 col-md-3 col-lg-3")
   next_button_link = soup.find("li", class_="next").find('a').get('href')
   next_button_link = f"{url_to_scrape}/{next_button_link}" if "catalogue" in next_button_link else f"{url_to_scrape}/catalogue/{next_button_link}"

   for each in books:
       book_names.append(each.find("img").get("alt"))
       book_price.append(each.find("p", class_="price_color").text)
       book_availability.append(each.find("p", class_="instock availability").text.strip())

   return next_button_link


# Ottenere il proxy da JSON
with open("proxy-list.json") as json_file:
   proxies = json.load(json_file)
   for each in proxies:
       threading.Thread(target=verify_proxies, args=(each,)).start()

time.sleep(4)

for i in range(len(valid_proxies)):
   response = request_function(url_to_scrape, valid_proxies[i])
   if response is not None:
       next_button_link = data_extract(response)
       break
   else:
       continue

for proxy in valid_proxies:
   print(f"Using Proxy: {proxy['proxy_address']}")
   response = request_function(next_button_link, proxy)
   if response is not None:
       next_button_link = data_extract(response)
   else:
       continue

for each in range(len(book_names)):
   print(
       f"No {each + 1}: Book Name: {book_names[each]} Book Price: {book_price[each]} and Availability {book_availability[each]}")

Risultato finale

Dopo un'esecuzione riuscita, i risultati appaiono come quelli riportati di seguito. Questo va a estrarre informazioni su oltre 100 libri utilizzando i 2 proxy forniti.

L'utilizzo di più proxy per lo scraping del Web consente di aumentare il numero di richieste alla risorsa di destinazione e di aggirare il blocco. Per mantenere la stabilità del processo di scraping, è consigliabile utilizzare indirizzi IP che offrano un'elevata velocità e un forte fattore di fiducia, come i proxy statici ISP e dinamici residenziali. Inoltre, la funzionalità dello script fornito può essere facilmente espansa per soddisfare le diverse esigenze di scraping dei dati.