Como rodar proxies durante a recolha de dados da Web

Por muito boa que esta abordagem à recolha de dados pareça, é mal vista por muitos sites, e há consequências para seguir com o scraping, como a proibição do nosso IP.

Numa nota positiva, os serviços de proxy ajudam a evitar esta consequência. Permitem-nos assumir um IP diferente enquanto recolhemos dados online e, por mais seguro que pareça, usar vários proxies é melhor. A utilização de vários proxies durante a recolha de dados faz com que a interação com o sítio Web pareça aleatória e aumenta a segurança.

O sítio Web alvo (fonte) para este guia é uma livraria em linha. Ele imita um site de comércio eletrónico de livros. Nele há livros com nome, preço e disponibilidade. Como este guia não se concentra na organização dos dados retornados, mas na rotação de proxies, os dados retornados serão apresentados apenas no console.

Preparação do ambiente de trabalho e integração de proxies

Instale e importe alguns módulos Python em nosso arquivo antes de começarmos a codificar as funções que ajudariam na rotação dos proxies e na raspagem do site.

pip install requests beautifulSoup4 lxml

3 dos 5 módulos Python necessários para este script de raspagem podem ser instalados usando o comando acima. Requests permite-nos enviar pedidos HTTP para o website, beautifulSoup4 permite-nos extrair a informação do HTML da página fornecida pelos pedidos, e LXML é um analisador de HTML.

Além disso, também precisamos do módulo de threading embutido para permitir vários testes dos proxies para ver se eles funcionam e json para ler de um arquivo 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 = ""

Passo 1: Verificar o proxy a partir de uma lista de proxies

Construir um script de raspagem que gira proxies significa que precisamos de uma lista de proxies para escolher durante a rotação. Alguns proxies exigem autenticação e outros não. Devemos criar uma lista de dicionários com detalhes do proxy, incluindo o nome de usuário e a senha do proxy, se a autenticação for necessária.

A melhor abordagem para isso é colocar nossas informações de proxy em um arquivo JSON separado, organizado como o que está abaixo:

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

No campo "proxy_address", introduza o endereço IP e a porta, separados por dois pontos. Nos campos "proxy_username" e "proxy_password", forneça o nome de utilizador e a palavra-passe para autorização.

Acima está o conteúdo de um arquivo JSON com 4 proxies para o script escolher. O nome de usuário e a senha podem estar vazios, indicando um proxy que não requer autenticação.

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 precaução, esta função garante que os proxies fornecidos estão activos e a funcionar. Podemos conseguir isso percorrendo cada dicionário no arquivo JSON, enviando uma solicitação GET para o site e, se um código de status 200 for retornado, adicione esse proxy à lista de valid_proxies - uma variável que criamos anteriormente para abrigar os proxies que funcionam na lista do arquivo. Se a chamada não for bem-sucedida, a execução continua.

Passo 2: Enviando solicitação de raspagem da Web

Como o beautifulSoup precisa do código HTML do site para extrair os dados de que precisamos, criamos request_function(), que recebe a URL e o proxy de escolha e retorna o código HTML como texto. A variável proxy permite-nos encaminhar o pedido através de diferentes proxies, rodando assim o 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

Passo 3: Extrair dados do sítio Web de destino

data_extract() extrai os dados de que precisamos do código HTML fornecido. Ele reúne o elemento HTML que contém as informações do livro, como o nome do livro, o preço e a disponibilidade. Também extrai o link para a página seguinte.

Isto é particularmente complicado porque a hiperligação é dinâmica, pelo que tivemos de ter em conta o dinamismo. Finalmente, ele examina os livros e extrai o nome, o preço e a disponibilidade e, em seguida, retorna o link do próximo botão que usaríamos para recuperar o código HTML da próxima página.

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

Passo 4: Ligar tudo

Para ligar tudo junto, temos que:

1. Carregar os detalhes do proxy a partir do arquivo JSON. Iniciar uma thread para cada proxy usando o threading.Thread(). Isso nos ajudará a testar vários proxies ao mesmo tempo. Os proxies válidos são adicionados a valid_proxies().
2. Carrega a página inicial da fonte usando um proxy válido. Se um proxy não funcionar, usamos o próximo, tudo para garantir que a página inicial carregue ou não retorne Nenhum antes que a execução continue.
3. Em seguida, percorremos os proxies ativos, usamos a função request_function() para criar uma solicitação GET. E se recebemos uma solicitação GET, coletamos dados do site.
4. Finalmente, imprimimos os dados recolhidos na 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 o elemento HTML de uma 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


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


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

Após uma execução bem-sucedida, os resultados são parecidos com os abaixo. Isso continua a extrair informações sobre mais de 100 livros usando os 2 proxies fornecidos.

A utilização de vários proxies para a recolha de dados da Web permite aumentar o número de pedidos ao recurso alvo e ajuda a contornar o bloqueio. Para manter a estabilidade do processo de recolha de dados, é aconselhável utilizar endereços IP que ofereçam alta velocidade e um forte fator de confiança, tais como ISP estático e proxies residenciais dinâmicos. Além disso, a funcionalidade do script fornecido pode ser facilmente expandida para acomodar vários requisitos de raspagem de dados.