前言
在實際的爬蟲應用中，爬蟲程序經(jīng)常會通過代理服務器來進行網(wǎng)絡訪問，以避免訪問過于頻繁而受到網(wǎng)站服務器的限制。但是，代理服務器的IP地址也可能被目標網(wǎng)站限制，導致無法正常訪問。這時候，我們需要在代理IP被封后立刻換下一個IP繼續(xù)任務，以保證爬蟲的正常運行。
本文將介紹在Python中如何實現(xiàn)代理IP的動態(tài)切換，并給出相關的代碼案例。在講解具體實現(xiàn)方法之前，我們先了解一下代理服務器的基本原理。
一、 代理服務器的工作原理
代理服務器是一種在客戶端與服務器之間進行轉發(fā)的服務器。當客戶端向服務器發(fā)起網(wǎng)絡請求時，代理服務器會先接收這個請求，然后再將請求轉發(fā)給目標服務器，最后將目標服務器返回的響應結果再轉發(fā)給客戶端。代理服務器在這個過程中扮演了中間人的角色，可以對客戶端和服務器之間的通信進行攔截和修改。
代理服務器對爬蟲程序的作用主要體現(xiàn)在以下兩個方面：
- 隱藏客戶端的真實IP地址，保護客戶端的隱私和安全；
- 分散客戶端的網(wǎng)絡訪問，降低被目標服務器封禁的風險。
代理服務器有多種工作模式，其中最常用的模式是HTTP代理。HTTP代理是基于HTTP協(xié)議的代理模式，客戶端的HTTP請求會先被發(fā)送到代理服務器上，然后再由代理服務器轉發(fā)給目標服務器，例如以下代碼：
import requests
proxies = {
? ? 'http': 'http://127.0.0.1:8080', ?# HTTP代理服務器地址和端口
? ? 'https': 'http://127.0.0.1:8080' ?# HTTPS代理服務器地址和端口
}
response = requests.get("http://www.example.com", proxies=proxies)在這個例子中，我們使用了requests庫來發(fā)送HTTP請求，其中proxies參數(shù)指定了HTTP代理服務器的地址和端口。需要注意的是，這里使用的代理服務器是本機上的一個HTTP代理服務器，如果要使用其他代理服務器，需要替換IP地址和端口號。
為了實現(xiàn)代理IP的動態(tài)切換，我們需要了解如何使用Python來自動獲取可用的代理IP列表，并在IP被封后自動切換到下一個可用的IP。接下來，我們將詳細介紹這個過程。
二、獲取可用的代理IP列表
獲取可用的代理IP列表有多種方法，其中一種常用的方法是從代理IP網(wǎng)站上爬取代理IP信息。代理IP網(wǎng)站上通常會提供免費的代理IP列表，我們只需要對其進行爬取和驗證即可得到可用的代理IP列表。
以下是一個實現(xiàn)自動獲取代理IP列表的示例代碼：
import requests
from bs4 import BeautifulSoup
import time
def get_proxy_list():
? ? # 獲取代理IP列表的URL
? ? url = "http://www.example.com/proxy_list.html"
? ? # 發(fā)送請求獲取頁面內(nèi)容
? ? response = requests.get(url)
? ? soup = BeautifulSoup(response.text, 'html.parser')
? ? # 解析HTML頁面，獲取代理IP列表
? ? proxy_list = []
? ? for tr in soup.find_all('tr'):
? ? ? ? tds = tr.find_all('td')
? ? ? ? if len(tds) == 2:
? ? ? ? ? ? ip = tds[0].get_text()
? ? ? ? ? ? port = tds[1].get_text()
? ? ? ? ? ? proxy = '{}:{}'.format(ip, port)
? ? ? ? ? ? proxy_list.append(proxy)
? ? return proxy_list
def test_proxy(proxy):
? ? # 測試代理IP的可用性
? ? try:
? ? ? ? proxies = {
? ? ? ? ? ? 'http': 'http://{}'.format(proxy),
? ? ? ? ? ? 'https': 'https://{}'.format(proxy)
? ? ? ? }
? ? ? ? response = requests.get('http://www.baidu.com', proxies=proxies, timeout=5)
? ? ? ? if response.status_code == 200:
? ? ? ? ? ? return True
? ? except:
? ? ? ? return False
def get_available_proxies(proxy_list):
? ? # 獲取可用的代理IP列表
? ? available_proxies = []
? ? for proxy in proxy_list:
? ? ? ? if test_proxy(proxy):
? ? ? ? ? ? available_proxies.append(proxy)
? ? return available_proxies
if __name__ == '__main__':
? ? proxy_list = get_proxy_list()
? ? available_proxies = get_available_proxies(proxy_list)
? ? print('Available proxies: {}'.format(available_proxies))在這個示例代碼中，我們首先定義了一個get_proxy_list函數(shù)，用于從網(wǎng)站上獲取代理IP列表。該函數(shù)通過requests庫發(fā)送HTTP請求，然后使用BeautifulSoup庫解析HTML頁面，獲取代理IP列表。
接下來，我們定義了一個test_proxy函數(shù)，用于測試代理IP的可用性。該函數(shù)使用requests庫發(fā)送HTTP請求，如果請求成功返回了200狀態(tài)碼，則認為該代理IP可用。
最后，我們定義了一個get_available_proxies函數(shù)，用于獲取可用的代理IP列表。該函數(shù)遍歷原始代理IP列表，依次測試每個代理IP的可用性，將可用的代理IP添加到新的列表中。
注意，在測試代理IP的可用性時，我們需要設置一個較短的超時時間，以避免因為等待時間過長而浪費時間。此外，由于測試代理IP的過程很可能會失敗，因此我們還需要添加異常處理邏輯，確保程序不會因為一個代理IP的失效而停止運行。
三、實現(xiàn)代理IP的動態(tài)切換
在獲取可用的代理IP列表后，我們需要實現(xiàn)代理IP的動態(tài)切換。具體思路是，在向目標服務器發(fā)送HTTP請求前，先從代理IP列表中選取一個可用的代理IP，如果該代理IP不能正常工作，則切換到下一個可用的代理IP，直到找到能正常工作的代理IP為止。
以下是一個實現(xiàn)代理IP的動態(tài)切換的示例代碼：
import requests
from bs4 import BeautifulSoup
import random
import time
# 全局變量，代理IP列表
PROXY_LIST = []
def get_proxy_list():
? ? # 獲取代理IP列表的URL
? ? url = "http://www.example.com/proxy_list.html"
? ? # 發(fā)送請求獲取頁面內(nèi)容
? ? response = requests.get(url)
? ? soup = BeautifulSoup(response.text, 'html.parser')
? ? # 解析HTML頁面，獲取代理IP列表
? ? proxy_list = []
? ? for tr in soup.find_all('tr'):
? ? ? ? tds = tr.find_all('td')
? ? ? ? if len(tds) == 2:
? ? ? ? ? ? ip = tds[0].get_text()
? ? ? ? ? ? port = tds[1].get_text()
? ? ? ? ? ? proxy = '{}:{}'.format(ip, port)
? ? ? ? ? ? proxy_list.append(proxy)
? ? return proxy_list
def test_proxy(proxy):
? ? # 測試代理IP的可用性
? ? try:
? ? ? ? proxies = {
? ? ? ? ? ? 'http': 'http://{}'.format(proxy),
? ? ? ? ? ? 'https': 'https://{}'.format(proxy)
? ? ? ? }
? ? ? ? response = requests.get('http://www.baidu.com', proxies=proxies, timeout=5)
? ? ? ? if response.status_code == 200:
? ? ? ? ? ? return True
? ? except:
? ? ? ? return False
def get_available_proxies(proxy_list):
? ? # 獲取可用的代理IP列表
? ? available_proxies = []
? ? for proxy in proxy_list:
? ? ? ? if test_proxy(proxy):
? ? ? ? ? ? available_proxies.append(proxy)
? ? return available_proxies
def get_random_proxy():
? ? # 獲取隨機的代理IP
? ? global PROXY_LIST
? ? if not PROXY_LIST:
? ? ? ? # 第一次使用時，先獲取可用的代理IP列表
? ? ? ? proxy_list = get_proxy_list()
? ? ? ? PROXY_LIST = get_available_proxies(proxy_list)
? ? if not PROXY_LIST:
? ? ? ? # 如果沒有可用的代理IP，等待一段時間后重試
? ? ? ? time.sleep(60)
? ? ? ? proxy_list = get_proxy_list()
? ? ? ? PROXY_LIST = get_available_proxies(proxy_list)
? ? return random.choice(PROXY_LIST)
def make_request(url):
? ? # 發(fā)送HTTP請求
? ? while True:
? ? ? ? # 從代理IP列表中隨機選擇一個IP
? ? ? ? proxy =
get_random_proxy()
? ? ? ? proxies = {
? ? ? ? ? ? 'http': 'http://{}'.format(proxy),
? ? ? ? ? ? 'https': 'https://{}'.format(proxy)
? ? ? ? }
? ? ? ? try:
? ? ? ? ? ? # 發(fā)送HTTP請求
? ? ? ? ? ? response = requests.get(url, proxies=proxies, timeout=5)
? ? ? ? ? ? if response.status_code == 200:
? ? ? ? ? ? ? ? return response
? ? ? ? except:
? ? ? ? ? ? # 如果代理IP失效，從列表中移除該IP
? ? ? ? ? ? PROXY_LIST.remove(proxy)
if __name__ == '__main__':
? ? url = 'http://www.example.com'
? ? response = make_request(url)
? ? print(response.text)在這個示例代碼中，我們定義了一個全局變量PROXY_LIST，用于保存可用的代理IP列表。首先，我們定義了一個get_random_proxy函數(shù)，用于從代理IP列表中隨機選擇一個代理IP，并在需要時動態(tài)更新可用的代理IP列表。
接下來，我們定義了一個make_request函數(shù)，用于發(fā)送HTTP請求。該函數(shù)在調(diào)用get_random_proxy函數(shù)獲取代理IP后，使用requests庫發(fā)送HTTP請求，并在請求成功后返回響應結果。如果請求失敗，則說明代理IP失效，需要從可用的代理IP列表中移除該代理IP，并重新選擇一個代理IP進行請求。
最后，在程序的主函數(shù)中，我們定義了一個URL地址，并調(diào)用make_request函數(shù)發(fā)送HTTP請求。如果請求成功，則輸出響應內(nèi)容。
至此，我們已經(jīng)完成了代理IP的動態(tài)切換功能的實現(xiàn)。接下來，我們對上述代碼進行修改，加入一些必要的異常處理邏輯和日志記錄功能。
四、異常處理和日志記錄
在實際的爬蟲應用中，我們經(jīng)常會遇到各種意外情況，例如代理IP失效、網(wǎng)絡連接超時、目標網(wǎng)站返回錯誤響應等。為了保證程序的穩(wěn)定性和可靠性，我們需要對這些情況進行合理的異常處理和日志記錄。
以下是一個加入異常處理和日志記錄的示例代碼：
import requests
from requests.exceptions import ProxyError, Timeout, ConnectionError
from bs4 import BeautifulSoup
import random
import time
import logging
# 全局變量，代理IP列表
PROXY_LIST = []
def init_logging():
? ? # 初始化日志記錄器
? ? logger = logging.getLogger()
? ? logger.setLevel(logging.INFO)
? ? formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
? ? handler = logging.FileHandler('proxy.log')
? ? handler.setFormatter(formatter)
? ? logger.addHandler(handler)
? ? return logger
def get_proxy_list():
? ? # 獲取代理IP列表的URL
? ? url = "http://www.example.com/proxy_list.html"
? ? # 發(fā)送請求獲取頁面內(nèi)容
? ? response = requests.get(url)
? ? soup = BeautifulSoup(response.text, 'html.parser')
? ? # 解析HTML頁面，獲取代理IP列表
? ? proxy_list = []
? ? for tr in soup.find_all('tr'):
? ? ? ? tds = tr.find_all('td')
? ? ? ? if len(tds) == 2:
? ? ? ? ? ? ip = tds[0].get_text()
? ? ? ? ? ? port = tds[1].get_text()
? ? ? ? ? ? proxy = '{}:{}'.format(ip, port)
? ? ? ? ? ? proxy_list.append(proxy)
? ? return proxy_list
def test_proxy(proxy):
? ? # 測試代理IP的可用性
? ? try:
? ? ? ? proxies = {
? ? ? ? ? ? 'http': 'http://{}'.format(proxy),
? ? ? ? ? ? 'https': 'https://{}'.format(proxy)
? ? ? ? }
? ? ? ? response = requests.get('http://www.baidu.com', proxies=proxies, timeout=5)
? ? ? ? if response.status_code == 200:
? ? ? ? ? ? return True
? ? except:
? ? ? ? return False
def get_available_proxies(proxy_list):
? ? # 獲取可用的代理IP列表
? ? available_proxies = []
? ? for proxy in proxy_list:
? ? ? ? if test_proxy(proxy):
? ? ? ? ? ? available_proxies.append(proxy)
? ? return available_proxies
def get_random_proxy():
? ? # 獲取隨機的代理IP
? ? global PROXY_LIST
? ? if not PROXY_LIST:
? ? ? ? # 第一次使用時，先獲取可用的代理IP列表
? ? ? ? proxy_list = get_proxy_list()
? ? ? ? PROXY_LIST = get_available_proxies(proxy_list)
? ? if not PROXY_LIST:
? ? ? ? # 如果沒有可用的代理IP，等待一段時間后重試
? ? ? ? time.sleep(60)
? ? ? ? proxy_list = get_proxy_list()
? ? ? ? PROXY_LIST = get_available_proxies(proxy_list)
? ? return random.choice(PROXY_LIST)
def make_request(url):
? ? # 發(fā)送HTTP請求
? ? while True:
? ? ? ? # 從代理IP列表中隨機選擇一個IP
? ? ? ? proxy = get_random_proxy()
? ? ? ? proxies = {
? ? ? ? ? ? 'http': 'http://{}'.format(proxy),
? ? ? ? ? ? 'https': 'https://{}'.format(proxy)
? ? ? ? }
? ? ? ? try:
? ? ? ? ? ? # 發(fā)送HTTP請求
? ? ? ? ? ? response = requests.get(url, proxies=proxies, timeout=5)
? ? ? ? ? ? if response.status_code == 200:
? ? ? ? ? ? ? ? return response
? ? ? ? except ProxyError as e:
? ? ? ? ? ? # 代理服務器錯誤，從列表中移除該IP
? ? ? ? ? ? PROXY_LIST.remove(proxy)
? ? ? ? ? ? logging.warning('ProxyError: {}'.format(str(e)))
? ? ? ? except Timeout as e:
? ? ? ? ? ? # 超時錯誤，重試
? ? ? ? ? ? logging.warning('Timeout: {}'.format(str(e)))
? ? ? ? except ConnectionError as e:
? ? ? ? ? ? # 連接錯誤，重試
? ? ? ? ? ? logging.warning('ConnectionError: {}'.format(str(e)))
? ? ? ? except Exception as e:
? ? ? ? ? ? # 其他未知錯誤，重試
? ? ? ? ? ? logging.warning('Exception: {}'.format(str(e)))
if __name__ == '__main__':
? ? init_logging()
? ? url = 'http://www.example.com'
? ? response = make_request(url)
? ? print(response.text)在這個示例代碼中，我們首先引入了requests.exceptions模塊和logging模塊。requests.exceptions模塊提供了一些常見的網(wǎng)絡請求異常類型，我們可以通過捕獲這些異常類型來實現(xiàn)異常處理。logging模塊則提供了一個日志記錄器，我們可以使用它來記錄程序運行時的異常和錯誤信息。
接下來，在程序的主函數(shù)中，我們調(diào)用了一個init_logging函數(shù)，用于初始化日志記錄器。該函數(shù)設置了日志記錄器的級別、格式和輸出文件，并返回一個記錄器實例。
最后，在make_request函數(shù)中，我們通過try-except語句對網(wǎng)絡請求中可能出現(xiàn)的異常進行了捕獲和處理。例如，如果代理服務器返回了錯誤碼，我們將該代理IP從列表中移除，并記錄警告日志。如果發(fā)生超時錯誤、連接錯誤或其他未知錯誤，我們直接記錄警告日志，并在下一次循環(huán)中重試。
至此，我們已經(jīng)完成了對代理IP的動態(tài)切換功能的實現(xiàn)，并加入了必要的異常處理和日志記錄功能。文章來源：http://www.zghlxwxcb.cn/news/detail-739089.html
總結
為了實現(xiàn)在代理IP被封后立即切換到下一個IP，我們可以在爬蟲程序中加入一個代理IP池，定時從可用的代理IP列表中隨機選擇一個IP，并發(fā)送HTTP請求。如果請求失敗，我們可以將失敗的代理IP從列表中移除，并在下一次選擇IP時避開此IP。同時，我們需要加入必要的異常處理和日志記錄功能，以保證程序的穩(wěn)定性和可靠性。這樣，即使某個代理IP被封，我們也能夠及時切換到下一個可用的IP，繼續(xù)執(zhí)行爬蟲任務。文章來源地址http://www.zghlxwxcb.cn/news/detail-739089.html
到了這里，關于python爬蟲，如何在代理的IP被封后立刻換下一個IP繼續(xù)任務？的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關文章，希望大家以后多多支持TOY模板網(wǎng)！
