Jak obracać serwery proxy podczas skrobania danych internetowych?

Komentarze: 0

Jakkolwiek dobre wydaje się takie podejście do gromadzenia danych, jest ono źle widziane przez wiele witryn, a za kontynuowanie skrobania grożą konsekwencje, takie jak blokada naszego adresu IP.

Z drugiej strony, usługi proxy pomagają uniknąć tych konsekwencji. Pozwalają nam one przyjąć inny adres IP podczas gromadzenia danych online i choć wydaje się to bezpieczne, korzystanie z wielu serwerów proxy jest lepsze. Korzystanie z wielu serwerów proxy podczas skrobania sprawia, że interakcja z witryną wydaje się losowa i zwiększa bezpieczeństwo.

Stroną docelową (źródłem) dla tego przewodnika jest księgarnia internetowa. Imituje ona witrynę e-commerce dla książek. Znajdują się na niej książki z nazwą, ceną i dostępnością. Ponieważ ten przewodnik nie skupia się na organizowaniu zwracanych danych, ale na obracaniu proxy, zwracane dane będą prezentowane tylko w konsoli.

Przygotowanie środowiska pracy i integracja serwerów proxy

Zainstaluj i zaimportuj kilka modułów Pythona do naszego pliku, zanim zaczniemy kodować funkcje, które pomogą w rotacji serwerów proxy i skrobaniu witryny.

pip install requests beautifulSoup4 lxml

3 z 5 modułów Pythona potrzebnych do tego skryptu skrobiącego można zainstalować za pomocą powyższego polecenia. Requests pozwala nam wysłać żądanie HTTP do strony internetowej, beautifulSoup4 pozwala nam wyodrębnić informacje z HTML strony dostarczonej przez żądania, a LXML jest parserem HTML.

Ponadto potrzebujemy również wbudowanego modułu wątkowania, aby umożliwić wielokrotne testowanie serwerów proxy w celu sprawdzenia, czy działają, oraz json do odczytu z pliku 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 = ""

Krok 1: Weryfikacja serwera proxy z listy serwerów proxy

Zbudowanie skryptu skrobiącego, który rotuje serwery proxy oznacza, że potrzebujemy listy serwerów proxy do wyboru podczas rotacji. Niektóre proxy wymagają uwierzytelnienia, a inne nie. Musimy utworzyć listę słowników ze szczegółami serwera proxy, w tym nazwą użytkownika i hasłem serwera proxy, jeśli wymagane jest uwierzytelnienie.

Najlepszym podejściem do tego jest umieszczenie naszych informacji o proxy w osobnym pliku JSON zorganizowanym jak ten poniżej:

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

W polu "proxy_address" wprowadź adres IP i port, oddzielając je dwukropkiem. W polach "proxy_username" i "proxy_password" należy podać nazwę użytkownika i hasło do autoryzacji.

Powyżej znajduje się zawartość pliku JSON z 4 serwerami proxy do wyboru przez skrypt. Nazwa użytkownika i hasło mogą być puste, wskazując na serwer proxy, który nie wymaga uwierzytelniania.

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

Jako środek ostrożności, funkcja ta zapewnia, że dostarczone proxy są aktywne i działają. Możemy to osiągnąć, przechodząc w pętli przez każdy słownik w pliku JSON, wysyłając żądanie GET do strony internetowej, a jeśli zwrócony zostanie kod statusu 200, dodajemy ten serwer proxy do listy valid_proxies - zmiennej, którą utworzyliśmy wcześniej, aby przechowywać działające serwery proxy z listy w pliku. Jeśli wywołanie nie powiedzie się, wykonywanie jest kontynuowane.

Krok 2: Wysyłanie żądania web scrapingu

Ponieważ beautifulSoup potrzebuje kodu HTML strony internetowej, aby wyodrębnić potrzebne nam dane, stworzyliśmy funkcję request_function(), która pobiera adres URL i wybrany serwer proxy i zwraca kod HTML jako tekst. Zmienna proxy umożliwia nam przekierowanie żądania przez różne proxy, a tym samym obracanie 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

Krok 3: Wyodrębnianie danych z witryny docelowej

data_extract() wyodrębnia potrzebne nam dane z dostarczonego kodu HTML. Gromadzi element HTML zawierający informacje o książce, takie jak nazwa książki, cena i dostępność. Wyodrębnia również link do następnej strony.

Jest to szczególnie trudne, ponieważ link jest dynamiczny, więc musieliśmy uwzględnić jego dynamikę. Na koniec przegląda książki i wyodrębnia nazwę, cenę i dostępność, a następnie zwraca link do następnego przycisku, którego użyjemy do pobrania kodu HTML następnej strony.

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

Krok 4: Łączenie wszystkiego razem

Aby połączyć wszystko razem, musimy:

  1. Wczytać szczegóły proxy z pliku JSON. Uruchomić wątek dla każdego proxy za pomocą threading.Thread(). Pomoże nam to przetestować wiele proxy jednocześnie. Prawidłowe proxy są dodawane do valid_proxies().
  2. Wczytaj stronę główną źródła używając poprawnego proxy. Jeśli proxy nie działa, używamy następnego, aby upewnić się, że strona główna załaduje się lub nie zwróci None przed kontynuowaniem wykonywania.
  3. Następnie przechodzimy przez aktywne proxy, używamy funkcji request_function(), aby utworzyć żądanie GET. Jeśli otrzymaliśmy żądanie GET, zbieramy dane z witryny.
  4. Na koniec wypisujemy zebrane dane do konsoli.
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]}")

Pełny kod

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


# Pobiera element HTML strony
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


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


# Pobierz proxy z 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]}")

Wynik końcowy

Po pomyślnym wykonaniu, wyniki wyglądają jak poniżej. Wyodrębniono informacje o ponad 100 książkach przy użyciu 2 dostarczonych serwerów proxy.

1.png

2.png

3.png

4.png

Korzystanie z wielu serwerów proxy do skrobania stron internetowych umożliwia zwiększenie liczby żądań do zasobu docelowego i pomaga ominąć blokady. Aby zachować stabilność procesu skrobania, zaleca się korzystanie z adresów IP, które oferują dużą szybkość i silny współczynnik zaufania, takich jak statyczne serwery proxy dostawców usług internetowych i dynamiczne serwery proxy. Dodatkowo, funkcjonalność dostarczonego skryptu można łatwo rozszerzyć, aby spełnić różne wymagania dotyczące skrobania danych.

Komentarze:

0 komentarze