Cómo rotar los proxies durante el scraping de datos web

Por muy bueno que parezca este enfoque de recopilación de datos, está mal visto por muchos sitios web, y hay consecuencias por seguir con el scraping, como el baneo de nuestra IP.

Como nota positiva, los servicios proxy ayudan a evitar esta consecuencia. Nos permiten adoptar una IP diferente mientras recopilamos datos online, y por muy seguro que parezca, usar varios proxies es mejor. Usar varios proxies mientras se hace scraping hace que la interacción con el sitio web parezca aleatoria y mejora la seguridad.

El sitio web objetivo (fuente) de esta guía es una librería online. Imita un sitio web de comercio electrónico de libros. En ella aparecen libros con nombre, precio y disponibilidad. Como esta guía no se centra en organizar los datos devueltos sino en rotar proxies, los datos devueltos sólo se presentarán en la consola.

Preparando el entorno de trabajo e integrando proxies

Instalar e importar algunos módulos de Python en nuestro fichero antes de poder empezar a codificar las funciones que ayudarían a rotar los proxies y a scrapear la web.

pip install requests beautifulSoup4 lxml

3 de los 5 módulos Python necesarios para este script de scraping se pueden instalar usando el comando anterior. Requests nos permite enviar peticiones HTTP a la página web, beautifulSoup4 nos permite extraer la información del HTML de la página proporcionada por requests, y LXML es un parser de HTML.

Además, también necesitamos el módulo de threading integrado para poder hacer múltiples pruebas con los proxies para ver si funcionan y json para leer de un fichero 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 = ""

Paso 1: Verificar el proxy a partir de una lista de proxies

Construir un script de scraping que rote proxies significa que necesitamos una lista de proxies para elegir durante la rotación. Algunos proxies requieren autenticación y otros no. Debemos crear una lista de diccionarios con los detalles del proxy, incluyendo el nombre de usuario y la contraseña del proxy si se necesita autenticación.

El mejor enfoque para esto es poner nuestra información de proxy en un archivo JSON separado organizado como el que se muestra a continuación:

[
  {
    "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": ""
  }
]

En el campo "proxy_address", introduzca la dirección IP y el puerto, separados por dos puntos. En los campos "proxy_username" y "proxy_password", indique el nombre de usuario y la contraseña para la autorización.

Arriba está el contenido de un archivo JSON con 4 proxies para que el script elija. El nombre de usuario y la contraseña pueden estar vacíos, indicando un proxy que no requiere autenticación.

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")

Como precaución, esta función asegura que los proxies proporcionados están activos y funcionando. Podemos lograr esto recorriendo cada diccionario en el archivo JSON, enviando una solicitud GET al sitio web, y si se devuelve un código de estado 200, entonces agregar ese proxy a la lista de valid_proxies - una variable que creamos anteriormente para alojar los proxies que funcionan de la lista en el archivo. Si la llamada no tiene éxito, la ejecución continúa.

Paso 2: Envío de la solicitud de web scraping

Como beautifulSoup necesita el código HTML del sitio web para extraer los datos que necesitamos, hemos creado request_function(), que toma la URL y el proxy elegidos y devuelve el código HTML como texto. La variable proxy nos permite enrutar la petición a través de diferentes proxies, rotando así el 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

Paso 3: Extracción de datos del sitio web de destino

data_extract() extrae los datos que necesitamos del código HTML proporcionado. Recoge el elemento HTML que alberga la información del libro como el nombre del libro, el precio y la disponibilidad. También extrae el enlace a la página siguiente.

Esto es especialmente complicado porque el enlace es dinámico, así que tuvimos que tener en cuenta el dinamismo. Por último, busca entre los libros y extrae el nombre, el precio y la disponibilidad, luego devuelve el enlace del botón siguiente que utilizaríamos para recuperar el código HTML de la página siguiente.

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

Paso 4: enlazarlo todo

Para enlazar todo entre sí, tenemos que:

1. Cargar los detalles del proxy desde el archivo JSON. Iniciar un hilo para cada proxy utilizando el threading.Thread(). Esto nos ayudará a probar múltiples proxies a la vez. Los proxies válidos se añaden a valid_proxies().
2. Carga la página principal de la fuente usando un proxy válido. Si un proxy no funciona, usamos el siguiente, todo para asegurar que la página principal carga o no devuelve None antes de continuar la ejecución.
3. Luego recorremos los proxies activos, usamos la función request_function() para crear una petición GET. Y si recibimos una solicitud GET, recopilamos datos del sitio.
4. Por último, imprimimos los datos recopilados en la consola.

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]}")

Código 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 el elemento HTML de una página
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


# Raspado
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


# Obtener proxy a partir de 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]}")

Resultado final

Tras una ejecución satisfactoria, los resultados son como los que se muestran a continuación. A continuación, se extrae información sobre más de 100 libros utilizando los 2 proxies proporcionados.

El uso de varios proxies para el web scraping permite aumentar el número de peticiones al recurso de destino y ayuda a eludir el bloqueo. Para mantener la estabilidad del proceso de raspado, es aconsejable utilizar direcciones IP que ofrezcan alta velocidad y un fuerte factor de confianza, como proxies ISP estáticos y proxies residenciales dinámicos. Además, la funcionalidad del script proporcionado puede ampliarse fácilmente para adaptarse a diversos requisitos de raspado de datos.