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.
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 = ""
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.
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
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
Aby połączyć wszystko razem, musimy:
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]}")
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]}")
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.
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