웹 데이터를 스크랩하는 동안 프록시를 회전하는 방법

05.09.2024

댓글: 0

기사 내용:

작업 환경 준비 및 프록시 통합

1단계: 프록시 목록에서 프록시 확인
2단계: 웹 스크래핑 요청 보내기
3단계: 대상 웹사이트에서 데이터 추출
4단계: 모든 것을 하나로 연결하기

전체 코드
최종 결과

이러한 데이터 수집 방식은 좋은 것처럼 보이지만 많은 웹사이트에서는 스크래핑에 대해 거부감을 가지고 있으며, 스크래핑을 계속할 경우 IP 사용 금지와 같은 불이익을 받을 수 있습니다.

긍정적인 측면에서 프록시 서비스는 이러한 결과를 피하는 데 도움이 됩니다. 프록시 서비스를 사용하면 온라인에서 데이터를 수집하는 동안 다른 IP를 사용할 수 있으며, 여러 개의 프록시를 사용하는 것이 더 안전합니다. 스크래핑하는 동안 여러 프록시를 사용하면 웹사이트와의 상호 작용이 무작위로 나타나고 보안이 강화됩니다.

이 가이드의 대상 웹사이트(소스)는 온라인 서점입니다. 이 웹사이트는 책을 판매하는 이커머스 웹사이트를 모방합니다. 여기에는 책 이름, 가격 및 구매 가능 여부가 표시됩니다. 이 가이드에서는 반환된 데이터를 구성하는 것이 아니라 프록시를 순환하는 데 중점을 두므로 반환된 데이터는 콘솔에만 표시됩니다.

작업 환경 준비 및 프록시 통합

프록시 로테이션과 웹사이트 스크래핑에 도움이 되는 함수 코딩을 시작하기 전에 몇 가지 Python 모듈을 설치하고 파일로 가져옵니다.

pip install requests beautifulSoup4 lxml

이 스크래핑 스크립트에 필요한 파이썬 모듈 5개 중 3개는 위의 명령어를 사용하여 설치할 수 있습니다. Requests는 웹사이트로 HTTP 요청을 보낼 수 있게 해주고, beautifulSoup4는 요청이 제공한 페이지의 HTML에서 정보를 추출할 수 있게 해주며, LXML은 HTML 파서입니다.

또한 프록시가 작동하는지 확인하기 위해 프록시를 여러 번 테스트하고 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 = ""

1단계: 프록시 목록에서 프록시 확인

프록시를 로테이션하는 스크래핑 스크립트를 작성하려면 로테이션 중에 선택할 수 있는 프록시 목록이 필요합니다. 일부 프록시는 인증이 필요하고 다른 프록시는 필요하지 않습니다. 인증이 필요한 경우 프록시 사용자 이름과 비밀번호를 포함한 프록시 세부 정보가 포함된 사전 목록을 만들어야 합니다.

이를 위한 가장 좋은 방법은 프록시 정보를 아래와 같이 정리된 별도의 JSON 파일에 넣는 것입니다:

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

"proxy_address" 필드에 콜론으로 구분하여 IP 주소와 포트를 입력합니다. "proxy_username" 및 "proxy_password" 필드에 인증을 위한 사용자 이름과 비밀번호를 입력합니다.

위는 스크립트에서 선택할 수 있는 4개의 프록시가 포함된 JSON 파일의 내용입니다. 사용자 이름과 비밀번호는 비워둘 수 있으며, 인증이 필요 없는 프록시를 나타냅니다.

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

예방 조치로 이 함수는 제공된 프록시가 활성화되어 작동하는지 확인합니다. JSON 파일의 각 사전을 반복하여 웹사이트에 GET 요청을 보낸 다음 상태 코드 200이 반환되면 해당 프록시를 파일의 목록에서 작동하는 프록시를 저장하기 위해 앞서 만든 변수인 valid_proxies 목록에 추가하면 됩니다. 호출에 성공하지 못하면 실행이 계속됩니다.

2단계: 웹 스크래핑 요청 보내기

beautifulSoup에서 필요한 데이터를 추출하려면 웹사이트의 HTML 코드가 필요하므로 선택한 URL과 프록시를 받아 HTML 코드를 텍스트로 반환하는 request_function()을 만들었습니다. 프록시 변수를 사용하면 요청을 여러 프록시를 통해 라우팅할 수 있으므로 프록시를 순환할 수 있습니다.

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

3단계: 대상 웹사이트에서 데이터 추출

data_extract()는 제공된 HTML 코드에서 필요한 데이터를 추출합니다. 책 이름, 가격, 구매 가능 여부와 같은 책 정보를 담고 있는 HTML 요소를 수집합니다. 또한 다음 페이지로 연결되는 링크도 추출합니다.

링크가 동적이기 때문에 동적 요소를 고려해야 했기 때문에 특히 까다로운 작업입니다. 마지막으로 책을 살펴보고 이름, 가격 및 재고 여부를 추출한 다음 다음 페이지의 HTML 코드를 검색하는 데 사용할 다음 버튼 링크를 반환합니다.

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

4단계: 모든 것을 하나로 연결하기

모든 것을 서로 연결해야 합니다:

JSON 파일에서 프록시 세부 정보를 로드합니다. threading.Thread()를 사용하여 각 프록시에 대한 스레드를 시작합니다. 이렇게 하면 한 번에 여러 프록시를 테스트하는 데 도움이 됩니다. 유효한 프록시는 valid_proxies()에 추가됩니다.
유효한 프록시를 사용하여 소스의 홈페이지를 로드합니다. 프록시가 작동하지 않으면 다음 프록시를 사용하여 홈페이지가 로드되는지 확인하거나 실행을 계속하기 전에 None을 반환하지 않습니다.
그런 다음 활성 프록시를 순환하면서 request_function() 함수를 사용하여 GET 요청을 생성합니다. 그리고 GET 요청을 받으면 사이트에서 데이터를 수집합니다.
마지막으로 수집된 데이터를 콘솔에 출력합니다.

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


# 페이지의 HTML 요소를 검색합니다.
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


# 스크래핑
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


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

최종 결과

실행이 성공적으로 완료되면 결과는 아래와 같습니다. 제공된 2개의 프록시를 사용하여 100개 이상의 책에 대한 정보를 추출합니다.

웹 스크래핑에 여러 프록시를 사용하면 대상 리소스에 대한 요청 횟수가 증가하고 차단을 우회하는 데 도움이 됩니다. 스크래핑 프로세스의 안정성을 유지하려면 정적 ISP 및 동적 주거 프록시와 같이 빠른 속도와 강력한 신뢰 요소를 제공하는 IP 주소를 사용하는 것이 좋습니다. 또한 제공된 스크립트의 기능을 쉽게 확장하여 다양한 데이터 스크래핑 요구 사항을 수용할 수 있습니다.

0 댓글

이전 기사