【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 scrapy

class 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 requests

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 True
    except:
        pass
    return False

def 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 True
    except:
        pass
    return 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 redis

class 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 proxy

def 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.text
    except:
        pass
    return None

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

三、完整代码实现

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

import scrapy
import requests
import redis

class 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 True
    except:
        pass
    return False

def 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 True
    except:
        pass
    return False

class 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 proxy

def 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.text
    except:
        pass
    return None

if __name__ == '__main__':
    from multiprocessing import Process

    spider_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库来访问目标网站。