【python】python开源代理ip池

一、前言

随着互联网的不断发展，越来越多的应用需要使用高匿代理IP才能访问目标网站，而代理IP作为一种能够隐藏本机真实IP地址的工具，在网络抓取、搜索引擎排名、广告投放、反爬虫等方面有着广泛的应用场景。但是，由于代理IP的稳定性难以保证，而且容易被反爬虫机制识别和封杀，为解决这些问题，我们可以通过构建一个代理IP池来实现代理IP的有效管理和使用。

本文将介绍如何使用Python语言和开源爬虫框架Scrapy，构建一个高匿代理IP池，并提供完整的代码和案例。

二、代理IP池的构建流程

构建代理IP池的大致流程如下：

爬取代理IP数据
验证代理IP的有效性和匿名性
将有效的代理IP保存到代理IP池中
从代理IP池中随机获取一个代理IP并使用

接下来，我们将详细介绍每个步骤的实现方式。

2.1、爬取代理IP数据

在构建代理IP池之前，我们需要先收集一些代理IP数据。这里我们可以使用一些代理IP网站上的API接口或者直接爬取网站上的代理IP数据。

以爬取站大爷代理网站为例，其网站上提供了一个可以直接访问的代理IP列表页面，我们可以从该页面中提取出代理IP地址和端口号等信息。

以下为爬取站大爷代理网站的代理IP数据的代码实现：

import scrapyclass XiciSpider(scrapy.Spider):name = 'xici'allowed_domains = ['www.zdaye.com']start_urls = ['https://www.zdaye.com/']def parse(self, response):item_list = response.xpath('//table[@id="ip_list"]//tr')[1:]for item in item_list:ip = item.xpath('.//td[2]/text()').get()port = item.xpath('.//td[3]/text()').get()yield {'ip': ip,'port': port}

以上代码中，我们使用Scrapy框架爬取西刺代理网站，爬取其网页上的代理IP地址和端口号，并以字典的格式保存到内存中。

2.2、验证代理IP的有效性和匿名性

由于代理IP的可用性和匿名性是不稳定的，我们需要定期验证其有效性和匿名性。验证代理IP可用性的方法是通过访问目标网站并获取目标网页的数据，如果获取到了数据，则说明该代理IP可用；否则，则说明该代理IP不可用。

验证代理IP匿名性的方法是通过访问代理IP网站上的检测页面，检测代理IP是否真正地隐藏了客户端的真实IP地址。我们可以使用Python的requests库发起HTTP请求和获取HTTP响应，判断目标网站的状态码和返回的内容是否符合要求。

以下为验证代理IP有效性和匿名性的代码实现：

import requestsdef validate_proxy(proxy):url = 'http://httpbin.org/get'proxies = {'http': 'http://{ip}:{port}'.format(**proxy),'https': 'https://{ip}:{port}'.format(**proxy)}try:response = requests.get(url, proxies=proxies, timeout=10)if response.status_code == 200 and response.json().get('origin'):return Trueexcept:passreturn Falsedef check_anonymous(proxy):url = 'https://www.baidu.com/'headers = {'User-Agent': 'Mozilla/5.0'}proxies = {'http': 'http://{ip}:{port}'.format(**proxy),'https': 'https://{ip}:{port}'.format(**proxy)}try:response = requests.get(url, headers=headers, proxies=proxies, timeout=10)if response.status_code == 200 and 'baidu' in response.text:return Trueexcept:passreturn False

以上代码中，我们定义了两个函数validate_proxy和check_anonymous，用于验证代理IP的有效性和匿名性。其中，validate_proxy函数会访问httpbin.org网站，并从响应数据中获取客户端的ip地址，如果获取到了数据，则说明该代理IP可用；否则，则说明不可用。check_anonymous函数会访问百度首页，并判断访问结果是否为正常页面，如果是，则说明代理IP匿名性高；否则，则说明代理IP匿名性低。

2.3、将有效的代理IP保存到代理IP池中

当我们验证出一批有效的代理IP时，可以将其保存到代理IP池中，以便于后续使用。

以下为将代理IP保存到代理IP池中的代码实现：

import redisclass ProxyPool(object):def __init__(self):self.client = redis.Redis(host='localhost', port=6379, db=0, password='')def add(self, proxy):self.client.sadd('proxy_pool', str(proxy))def remove(self, proxy):self.client.srem('proxy_pool', str(proxy))def random(self):return eval(self.client.srandmember('proxy_pool'))def all(self):return [eval(p) for p in self.client.smembers('proxy_pool')]

以上代码中，我们使用Redis数据库作为代理IP池的存储工具，通过在Redis中创建一个set类型的key值proxy_pool，来存储所有可用的代理IP。在这里，我们定义了一个名为ProxyPool的类，用于对代理IP池进行增删改查的操作。

2.4、从代理IP池中随机获取一个代理IP并使用

当我们需要使用代理IP访问目标网站时，可以从代理IP池中随机获取一个代理IP，并将其添加到requests库的代理参数proxies中，以便于使用代理IP来访问目标网站。

以下为从代理IP池中随机获取一个代理IP的代码实现：

def get_random_proxy():proxy_pool = ProxyPool()proxy = proxy_pool.random()while not validate_proxy(proxy):proxy_pool.remove(proxy)proxy = proxy_pool.random()return proxydef download(url):proxy = get_random_proxy()proxies = {'http': 'http://{ip}:{port}'.format(**proxy),'https': 'https://{ip}:{port}'.format(**proxy)}headers = {'User-Agent': 'Mozilla/5.0'}try:response = requests.get(url, headers=headers, proxies=proxies, timeout=10)if response.status_code == 200:return response.textexcept:passreturn None

以上代码中，我们使用get_random_proxy随机获取一个代理IP，并使用validate_proxy函数验证该代理IP是否可用，如果该代理IP不可用，则从代理IP池中删除该代理IP，并重新获取。当获取到可用的代理IP后，我们将其添加到requests库的参数proxies中，并使用requests库来访问目标网站。

三、完整代码实现

将以上代码整合到一个文件中，得到完整的高匿代理IP池的实现方式，代码如下：

import scrapy
import requests
import redisclass XiciSpider(scrapy.Spider):name = 'xici'allowed_domains = ['www.zdaye.com']start_urls = ['http://www.zdaye.com/']def parse(self, response):item_list = response.xpath('//table[@id="ip_list"]//tr')[1:]for item in item_list:ip = item.xpath('.//td[2]/text()').get()port = item.xpath('.//td[3]/text()').get()proxy = {'ip': ip,'port': port}if validate_proxy(proxy) and check_anonymous(proxy):add_proxy(proxy)def validate_proxy(proxy):url = 'http://httpbin.org/get'proxies = {'http': 'http://{ip}:{port}'.format(**proxy),'https': 'https://{ip}:{port}'.format(**proxy)}try:response = requests.get(url, proxies=proxies, timeout=10)if response.status_code == 200 and response.json().get('origin'):return Trueexcept:passreturn Falsedef check_anonymous(proxy):url = 'https://www.baidu.com/'headers = {'User-Agent': 'Mozilla/5.0'}proxies = {'http': 'http://{ip}:{port}'.format(**proxy),'https': 'https://{ip}:{port}'.format(**proxy)}try:response = requests.get(url, headers=headers, proxies=proxies, timeout=10)if response.status_code == 200 and 'baidu' in response.text:return Trueexcept:passreturn Falseclass ProxyPool(object):def __init__(self):self.client = redis.Redis(host='localhost', port=6379, db=0, password='')def add(self, proxy):self.client.sadd('proxy_pool', str(proxy))def remove(self, proxy):self.client.srem('proxy_pool', str(proxy))def random(self):return eval(self.client.srandmember('proxy_pool'))def all(self):return [eval(p) for p in self.client.smembers('proxy_pool')]def add_proxy(proxy):proxy_pool = ProxyPool()proxy_pool.add(proxy)def get_random_proxy():proxy_pool = ProxyPool()proxy = proxy_pool.random()while not validate_proxy(proxy):proxy_pool.remove(proxy)proxy = proxy_pool.random()return proxydef download(url):proxy = get_random_proxy()proxies = {'http': 'http://{ip}:{port}'.format(**proxy),'https': 'https://{ip}:{port}'.format(**proxy)}headers = {'User-Agent': 'Mozilla/5.0'}try:response = requests.get(url, headers=headers, proxies=proxies, timeout=10)if response.status_code == 200:return response.textexcept:passreturn Noneif __name__ == '__main__':from multiprocessing import Processspider_process = Process(target=scrapy.cmdline.execute, args=('scrapy crawl xici',))spider_process.start()while True:url = input('请输入需要访问的URL：')print(download(url))spider_process.join()

在代码中，我们首先定义了XiciSpider类，用于爬取西刺代理网站上的代理IP数据。然后，我们使用requests库对代理IP进行验证，并使用Redis来保存可用的代理IP。最后，我们定义了get_random_proxy函数，从代理IP池中随机获取一个代理IP，并使用requests库来访问目标网站。