前言
在实际的爬虫应用中,爬虫程序经常会通过代理服务器来进行网络访问,以避免访问过于频繁而受到网站服务器的限制。但是,代理服务器的IP地址也可能被目标网站限制,导致无法正常访问。这时候,我们需要在代理IP被封后立刻换下一个IP继续任务,以保证爬虫的正常运行。
本文将介绍在Python中如何实现代理IP的动态切换,并给出相关的代码案例。在讲解具体实现方法之前,我们先了解一下代理服务器的基本原理。
一、 代理服务器的工作原理
代理服务器是一种在客户端与服务器之间进行转发的服务器。当客户端向服务器发起网络请求时,代理服务器会先接收这个请求,然后再将请求转发给目标服务器,最后将目标服务器返回的响应结果再转发给客户端。代理服务器在这个过程中扮演了中间人的角色,可以对客户端和服务器之间的通信进行拦截和修改。
代理服务器对爬虫程序的作用主要体现在以下两个方面:
- 隐藏客户端的真实IP地址,保护客户端的隐私和安全;
- 分散客户端的网络访问,降低被目标服务器封禁的风险。
代理服务器有多种工作模式,其中最常用的模式是HTTP代理。HTTP代理是基于HTTP协议的代理模式,客户端的HTTP请求会先被发送到代理服务器上,然后再由代理服务器转发给目标服务器,例如以下代码:
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库来发送HTTP请求,其中proxies参数指定了HTTP代理服务器的地址和端口。需要注意的是,这里使用的代理服务器是本机上的一个HTTP代理服务器,如果要使用其他代理服务器,需要替换IP地址和端口号。
为了实现代理IP的动态切换,我们需要了解如何使用Python来自动获取可用的代理IP列表,并在IP被封后自动切换到下一个可用的IP。接下来,我们将详细介绍这个过程。
二、获取可用的代理IP列表
获取可用的代理IP列表有多种方法,其中一种常用的方法是从代理IP网站上爬取代理IP信息。代理IP网站上通常会提供免费的代理IP列表,我们只需要对其进行爬取和验证即可得到可用的代理IP列表。
以下是一个实现自动获取代理IP列表的示例代码:
import requests
from bs4 import BeautifulSoup
import time
def get_proxy_list():
# 获取代理IP列表的URL
url = "http://www.example.com/proxy_list.html"
# 发送请求获取页面内容
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函数,用于从网站上获取代理IP列表。该函数通过requests库发送HTTP请求,然后使用BeautifulSoup库解析HTML页面,获取代理IP列表。
接下来,我们定义了一个test_proxy函数,用于测试代理IP的可用性。该函数使用requests库发送HTTP请求,如果请求成功返回了200状态码,则认为该代理IP可用。
最后,我们定义了一个get_available_proxies函数,用于获取可用的代理IP列表。该函数遍历原始代理IP列表,依次测试每个代理IP的可用性,将可用的代理IP添加到新的列表中。
注意,在测试代理IP的可用性时,我们需要设置一个较短的超时时间,以避免因为等待时间过长而浪费时间。此外,由于测试代理IP的过程很可能会失败,因此我们还需要添加异常处理逻辑,确保程序不会因为一个代理IP的失效而停止运行。
三、实现代理IP的动态切换
在获取可用的代理IP列表后,我们需要实现代理IP的动态切换。具体思路是,在向目标服务器发送HTTP请求前,先从代理IP列表中选取一个可用的代理IP,如果该代理IP不能正常工作,则切换到下一个可用的代理IP,直到找到能正常工作的代理IP为止。
以下是一个实现代理IP的动态切换的示例代码:
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"
# 发送请求获取页面内容
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):
# 发送HTTP请求
while True:
# 从代理IP列表中随机选择一个IP
proxy =
get_random_proxy()
proxies = {
'http': 'http://{}'.format(proxy),
'https': 'https://{}'.format(proxy)
}
try:
# 发送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函数,用于从代理IP列表中随机选择一个代理IP,并在需要时动态更新可用的代理IP列表。
接下来,我们定义了一个make_request函数,用于发送HTTP请求。该函数在调用get_random_proxy函数获取代理IP后,使用requests库发送HTTP请求,并在请求成功后返回响应结果。如果请求失败,则说明代理IP失效,需要从可用的代理IP列表中移除该代理IP,并重新选择一个代理IP进行请求。
最后,在程序的主函数中,我们定义了一个URL地址,并调用make_request函数发送HTTP请求。如果请求成功,则输出响应内容。
至此,我们已经完成了代理IP的动态切换功能的实现。接下来,我们对上述代码进行修改,加入一些必要的异常处理逻辑和日志记录功能。
四、异常处理和日志记录
在实际的爬虫应用中,我们经常会遇到各种意外情况,例如代理IP失效、网络连接超时、目标网站返回错误响应等。为了保证程序的稳定性和可靠性,我们需要对这些情况进行合理的异常处理和日志记录。
以下是一个加入异常处理和日志记录的示例代码:
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"
# 发送请求获取页面内容
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):
# 发送HTTP请求
while True:
# 从代理IP列表中随机选择一个IP
proxy = get_random_proxy()
proxies = {
'http': 'http://{}'.format(proxy),
'https': 'https://{}'.format(proxy)
}
try:
# 发送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模块提供了一些常见的网络请求异常类型,我们可以通过捕获这些异常类型来实现异常处理。logging模块则提供了一个日志记录器,我们可以使用它来记录程序运行时的异常和错误信息。
接下来,在程序的主函数中,我们调用了一个init_logging函数,用于初始化日志记录器。该函数设置了日志记录器的级别、格式和输出文件,并返回一个记录器实例。
最后,在make_request函数中,我们通过try-except语句对网络请求中可能出现的异常进行了捕获和处理。例如,如果代理服务器返回了错误码,我们将该代理IP从列表中移除,并记录警告日志。如果发生超时错误、连接错误或其他未知错误,我们直接记录警告日志,并在下一次循环中重试。
至此,我们已经完成了对代理IP的动态切换功能的实现,并加入了必要的异常处理和日志记录功能。
总结
为了实现在代理IP被封后立即切换到下一个IP,我们可以在爬虫程序中加入一个代理IP池,定时从可用的代理IP列表中随机选择一个IP,并发送HTTP请求。如果请求失败,我们可以将失败的代理IP从列表中移除,并在下一次选择IP时避开此IP。同时,我们需要加入必要的异常处理和日志记录功能,以保证程序的稳定性和可靠性。这样,即使某个代理IP被封,我们也能够及时切换到下一个可用的IP,继续执行爬虫任务。