大家好，我是小小明，今天我们计划搭建一个代理Ip池。当然相对爬取免费的代理IP自己搭建代理IP池，本人更推荐大家直接购买付费的代理IP，相对而言稳定而且省心好用。

那么对于穷人而言，为了训练一下自己的技术并省下几块钱，咱们还是决定自己搭建一个玩玩。

要搭建一个代理ip池，我的思路：

1. 爬虫定期爬取代理IP，验证代理iP有效性，有效则存入Redis数据库
2. 一个线程或进程定期检查代理ip池的有效性，无效则从中删除

代理IP的爬取与解析

下面，首先去选择几个免费代理ip的网址去爬一下，这里我选择了快代理。

快代理提供了高匿代理和普通代理两个页面供爬取，网址是：

• https://www.kuaidaili.com/free/inha
• https://www.kuaidaili.com/free/intr

爬取高匿测试一下：

import requests
import pandas as pd

headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.106 Safari/537.36'
}

i = 1
url = f'https://www.kuaidaili.com/free/inha/{i}/'
r = requests.get(url, headers=headers)
r.encoding = 'u8'
ip_df, = pd.read_html(r.text)
ip_df

代理IP爬了，但是是否真的能用呢？

代理IP的校验

我们需要验证一下，有个不错的测试网站是httpbin.org，我们可以通过它来读取自己的ip地址：

r = requests.get(f"http://httpbin.org/ip")
r.json()

结果大致形式：

{'origin': '116.xx.xx.xxx'}

然后以如下形式访问代理ip：

r = requests.get(f"http://httpbin.org/ip",
                 proxies={'http': "175.42.158.226:9999"})
r.json()

若抛出异常：

说明代理失效。

在我们的机器有公网IP时，我们也可以搭建自己的服务器用于校验代理IP有效性，flask代码：

from flask import *

app = Flask(__name__)


@app.route('/')
def index():
    return request.remote_addr


if __name__ == '__main__':
    app.run(host="0.0.0.0", port=8088)

基于此我们可以写一个校验代理IP的方法：

def proxie_ip_validity_check(proxie_ip, proxie_type='http', timeout=1):
    try:
        r = requests.get(f"{proxie_type}://httpbin.org/ip", headers=headers,
                         proxies={proxie_type: proxie_ip}, timeout=timeout)
        if r.status_code == 200:
            return True
        return False
    except Exception as e:
        return False

经过我个人测试，高匿代理似乎没有一个是可以用的，所以这次我们只爬取普通代理ip。

批量爬取普通代理ip

设置超时时间为2秒，取7页数据：

import requests
import pandas as pd

headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.106 Safari/537.36'
}

http_ip_pool = []
https_ip_pool = []
for page in range(1, 8):
    print("当前爬取页面数：", page)
    url = f'https://www.kuaidaili.com/free/intr/{page}'
    r = requests.get(url, headers=headers)
    r.encoding = 'u8'
    ip_df, = pd.read_html(r.text)
    for ip, port, proxie_type in ip_df[["IP", "PORT", "类型"]].values:
        pool = http_ip_pool if proxie_type.lower() == 'http' else https_ip_pool
        proxie_ip = f"{ip}:{port}"
        if proxie_ip_validity_check(proxie_ip, proxie_type.lower(), 2):
            pool.append(proxie_ip)
len(http_ip_pool), len(https_ip_pool)

经测试爬到26个有效的代理ip，全部都是http类型：

写入到Redis服务器

为了方便其他爬虫程序使用，我们将爬到的代理IP写入到Redis数据库中。

首先需要安装Redis，下载地址：https://github.com/MicrosoftArchive/redis/releases

我的是Windows平台选择了：https://github.com/microsoftarchive/redis/releases/download/win-3.0.504/Redis-x64-3.0.504.zip

解压后直接双击redis-server.exe即可启动Redis服务器：

安装让Python操作redis的库：

pip install redis

基本使用：

import redis

r = redis.Redis(host='192.168.3.31', port=6379, db=0)
r.set('name', 'zhangsan')  # 添加
print(r.get('name'))

b'zhangsan'

考虑将代理IP存入Redis的set集合中，因为可以自动去重，测试一下：

r.sadd("proxie_ip", "ip1", "ip2")
r.sadd("proxie_ip", "ip2", "ip3")
r.smembers("proxie_ip")

{b'ip1', b'ip2', b'ip3'}

于是我们可以通过以下命令一次性存入Redis中：

if len(http_ip_pool)>0:
    r.sadd("proxie_ip_http", *http_ip_pool)
if len(https_ip_pool)>0:
    r.sadd("proxie_ip_https", *https_ip_pool)

存入Redis数据库后，Redis就已经相当于一个代理IP池，下面我们另外启动一个尝试，常识使用这些代理IP：

从代理Ip池中获取代理IP并使用

读取代理IP：

import redis

with redis.Redis(host='192.168.3.31', port=6379, db=0) as r:
    http_ip_bytes = r.smembers("proxie_ip_http")
    http_ip_pool = list(map(bytes.decode, http_ip_bytes))
http_ip_pool

['139.9.25.69:3128',
 '119.3.235.101:3128',
 '122.51.207.244:8888',
 '106.58.191.24:8888',
 '222.74.202.229:9999',
 '47.111.71.29:8081',
 '222.74.202.229:80',
 '222.74.202.244:80',
 '106.58.191.218:8888',
 '222.74.202.229:8080']

然后可以拿来模拟爬百度：

import requests
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.106 Safari/537.36'
}

for ip in http_ip_pool:
    try:
        r = requests.get(f"http://www.baidu.com/",
                         headers=headers, proxies={'http': ip}, timeout=1)
        print(r.status_code)
    except Exception as e:
        print(e)

HTTPConnectionPool(host='139.9.25.69', port=3128): Max retries exceeded with url: http://www.baidu.com/ (Caused by ConnectTimeoutError(<urllib3.connection.HTTPConnection object at 0x0000028D6A993248>, 'Connection to 139.9.25.69 timed out. (connect timeout=1)'))
200
200
HTTPConnectionPool(host='106.58.191.24', port=8888): Max retries exceeded with url: http://www.baidu.com/ (Caused by ConnectTimeoutError(<urllib3.connection.HTTPConnection object at 0x0000028D6B59C2C8>, 'Connection to 106.58.191.24 timed out. (connect timeout=1)'))
200
HTTPConnectionPool(host='47.111.71.29', port=8081): Max retries exceeded with url: http://www.baidu.com/ (Caused by ProxyError('Cannot connect to proxy.', RemoteDisconnected('Remote end closed connection without response')))
200
200
HTTPConnectionPool(host='106.58.191.218', port=8888): Max retries exceeded with url: http://www.baidu.com/ (Caused by ConnectTimeoutError(<urllib3.connection.HTTPConnection object at 0x0000028D6A9C2F88>, 'Connection to 106.58.191.218 timed out. (connect timeout=1)'))
200

可以看到只有一半的代理IP能在指定时间内成功访问百度，说明免费的代理IP就是不如收费的稳定。

定期爬取并清除失效的代理IP

只要再完成这部分，我们的代理IP池就能够初见雏形，冲！

最终完成代码：

import requests
import pandas as pd
import redis
from threading import Timer
import time

headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.106 Safari/537.36'
}


def proxie_ip_validity_check(proxie_ip, proxie_type='http', timeout=2):
    try:
        r = requests.get(f"{proxie_type}://httpbin.org/ip",
                         proxies={proxie_type: proxie_ip}, timeout=timeout)
        if r.status_code == 200:
            return True
        return False
    except:
        return False


redis_conn = redis.Redis(host='192.168.3.31', port=6379)


def crawl_proxy_ip2redis():
    for page in range(1, 20):
        print("当前爬取页面：", page)
        url = f'https://www.kuaidaili.com/free/intr/{page}'
        r = requests.get(url, headers=headers)
        r.encoding = 'u8'
        ip_df, = pd.read_html(r.text)
        for ip, port, proxie_type in ip_df[["IP", "PORT", "类型"]].values:
            proxie_ip = f"{ip}:{port}"
            proxie_type = proxie_type.lower()
            if proxie_ip_validity_check(proxie_ip, proxie_type, 2):
                redis_conn.sadd(f"proxie_ip_{proxie_type}", proxie_ip)


def get_proxy_ip_pool(proxie_type='http'):
    http_ip_bytes = redis_conn.smembers(f"proxie_ip_{proxie_type}")
    http_ip_pool = list(map(bytes.decode, http_ip_bytes))
    return http_ip_pool


def clear_invalid_proxie_ip():
    print("开始清空失效代理ip:")
    for proxie_type in ['http', 'https']:
        for ip in get_proxy_ip_pool(proxie_type):
            if not proxie_ip_validity_check(ip, proxie_type, 2):
                print(proxie_type, ip, "失效")
                redis_conn.srem(f"proxie_ip_{proxie_type}", ip)


crawl_proxy_ip2redis()
while True:
    # 5分钟后清空一次失效的代理ip
    Timer(5 * 60, clear_invalid_proxie_ip, ()).start()
    # 10分钟后下载一次代理ip
    Timer(10 * 60, crawl_proxy_ip2redis, ()).start()
    # 每隔10分钟一个周期
    time.sleep(10 * 60)

目前测试，这个Redis代理IP池还算好使，作为一个雏形勉强能用。

可以通过Redis图形化工具查看代理IP池的存储情况：

我使用的工具叫Redis Desktop Manager，可以百度下载。

目前我个人还算满意。

总结

通过本文讲解了：

1. pandas超简代码带请求头解析表格
2. 查看访问IP的方法
3. 搭建简易检验代理Ip的网站
4. Redis数据库的基本操作
5. 代理ip在request库中的使用方法
6. Timer定时器的使用
7. Redis 图形化工具的使用介绍

我是小小明，希望你有所收获。

一、申请代理IP

如果一个用户对某个网站多次的访问，有可能会被识别为爬虫，因而限制其客户端 ip 的访问，对于一些比较正规的网站，反爬系统很强，最容易出现这种情况，所以有时候有必要使用代理IP，我一般选择使用随机动态的代理ip,这样可以保证每次访问时随机的一个用户而不是一个固定的用户。

话不多说，注册IPIDEA进去，注册就送免费的100M流量，有特殊需求不够再买：

http://www.ipidea.net/?utm-source=gejing&utm-keyword=?gejing

生成API:

点击生成链接

复制链接包存起来，等会用。

二、在selenium使用代理IP实战(一)

设置代理基本格式：

import requests
proxies = {
    'http': 'http://222.89.32.159:21079',
    'https': 'http://222.89.32.159:21079'
}
headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.163 Safari/537.36"
}
res = requests.get(url=urls,headers=headers,proxies=proxies)

我想了半天没想好到底哪些网站反爬强，所以我就随便找一个网站来测试了，你可以再去尝试逛一下自己学校的教务管理系统，12360,facebook等…

目标网址：

https://www.taobao.com/

所以定位就很容易：

driver.find_element_by_name('q')

之前写过一次使用代理玩爬虫，是requests模块，但是作为个人，我越往后面学，发现selenium用得反而越来越多，requests被逐渐抛弃一般，所以这里补充一个selenium添加代理。

方式很简单：

ops.add_argument('--proxy-server=http://%s' % a) #添加代理

注意这里的a格式为：ip:port

注意： 使用代理ip需要安装模块selenium-wire：

pip install selenium-wire

你应该是：

from seleniumwire import webdriver

而不是：

from selenium import webdriver

比如在X宝搜索：XX手机

完整代码：

from selenium import webdriver
from fake_useragent import UserAgent
from selenium.webdriver.chrome.options import Options

headers = {'User-Agent': UserAgent().random}
ops = Options()

driver = webdriver.Chrome(r'D:\360安全浏览器下载\chromedriver.exe')

api_url = '让你复制的代理api链接'

driver.get(api_url)

a = driver.find_element_by_xpath('/html/body/pre').text  # 获取代理

ops.add_argument('--proxy-server=http://%s' % a) #添加代理

driver.delete_all_cookies()  #清楚cookies

driver.get('https://www.taobao.com/')

driver.find_element_by_name('q').send_keys('华为手机')

接下来是点击按钮：

确定点击的地方元素，然后使用click点击即可：

from selenium.webdriver import ActionChains

b= driver.find_element_by_class_name('search-button')  #定位搜索

ActionChains(driver).click(b).perform()

可能是触发了反扒机制吗？需要登录，我也不知道我的X宝账号密码，随便演示输入一下…剩下的自己操作
这里是账号密码分析：


所以定位账号密码如下，账号我设置的输入：chuanchuan,密码设置的输入：123456 瞎编的，具体根据你的实际账号来操作，我就不讲下去了，就是定位定位点点点

driver.find_element_by_name('fm-login-id').send_keys('chuanchuan')  # 输入账号

driver.find_element_by_name('fm-login-password').send_keys('123456')  # 输入密码

效果如下：

三、在selenium使用代理IP实战(二)

注意： 用代理爬外wang需要国外环境，为了演示我不得不买了一个国外环境测试，请看：国外环境服务器
比如：

https://www.facebook.com/

分析账号密码登录：

代码为：

from fake_useragent import UserAgent
import requests
from selenium import webdriver
from selenium.webdriver import ChromeOptions

headers = {'User-Agent': UserAgent().random}

api_url = '复制你的api'

res = requests.post(api_url, headers=headers, verify=True)
PROXY = res.text
print(PROXY)

ops = ChromeOptions()

ops.add_argument('--proxy-server=%s' % PROXY)  # 添加代理

driver = webdriver.Chrome(r'D:\360安全浏览器下载\chromedriver.exe', chrome_options=ops)

driver.get("https://m.facebook.com/")
driver.find_element_by_name('email').send_keys("你的账号")
driver.find_element_by_name('pass').send_keys('你的密码')
# 按钮
btnSubmit = driver.find_element_by_name('login')
btnSubmit.click()

效果如下：

我的账号被封了，后续操作不继续演示，可以根据我讲的selenium知识点自行操作，无非就是点点点定位定位保存保存。

三、selenium单个元素定位实战复习

3.1 定位填写

以微软搜索引擎为例：

https://cn.bing.com/?mkt=zh-CN

分析：

所以：

from selenium import webdriver

driver = webdriver.Chrome(r'D:\360安全浏览器下载\chromedriver.exe')

driver.get('https://cn.bing.com/?mkt=zh-CN')

driver.find_element_by_name('q').send_keys('川川菜鸟')

如下：

你也可以如下两种方式：

driver.find_element_by_id('sb_form_q').send_keys('川川菜鸟')

driver.find_element_by_class_name('sb_form_q').send_keys('川川菜鸟')

send_keys函数就是填写信息的意思。

3.2 点击搜索

分析：id或者class

b=driver.find_element_by_id('search_icon')

ActionChains(driver).click(b).perform()

如下：

上面是id定位，把class定位也写一下：

b = driver.find_element_by_class_name('search')
ActionChains(driver).click(b).perform()

有趣的是，定位class的时候是：search而不是search icon tooltip，个人认为可能是因为这个空格的原因，还好有爬虫经验，不然死卡在这个定位不对了。

3.3 完整代码

代理api请替换为你的，按照我的的方法去申请：

# coding=gbk
"""
作者：川川
公众号：玩转大数据
@时间  : 2022/3/3 17:11
群：428335755
"""
from selenium import webdriver
from selenium.webdriver import ActionChains

driver = webdriver.Chrome(r'D:\360安全浏览器下载\chromedriver.exe')

driver.get('https://cn.bing.com/?mkt=zh-CN')

# driver.find_element_by_name('q').send_keys('川川菜鸟')

# driver.find_element_by_id('sb_form_q').send_keys('川川菜鸟')

driver.find_element_by_class_name('sb_form_q').send_keys('川川菜鸟')

# b=driver.find_element_by_id('search_icon')
b = driver.find_element_by_class_name('search')
ActionChains(driver).click(b).perform()

代理IP的使用

• 先了解一下百度百科定义的IP
  

• 为什么要使用代理IP?
  反爬(反网络爬虫)

• 示例:
  测试网址

http://httpbin.org/get

• 用浏览器先访问测试网址下看看
  

• 再用我们写的代码简单请求一下网页看看

import requests
url='http://httpbin.org/get'

html=requests.get(url=url).text
print(html)
"""
{
  "args": {},
  "headers": {
    "Accept": "*/*",
    "Accept-Encoding": "gzip, deflate",
    "Host": "httpbin.org",
    "User-Agent": "python-requests/2.23.0",
    "X-Amzn-Trace-Id": "Root=1-5ff704d4-3841771516040beb29f6066f"
  },
  "origin": "1.192.244.128",
  "url": "http://httpbin.org/get"
}
"""

疑惑???

“User-Agent”: “python-requests/2.23.0”

网站如何来判定是人类正常访问还是爬虫程序访问? —> 检查请求头！！！
我们是不是需要发送请求时重构一下User-Agent???
添加 headers参数！！！

那就使用fake_useragent 模块
让它伪造一个出来再试试

import requests
from fake_useragent import UserAgent
url='http://httpbin.org/get'

headers={'User-Agent':UserAgent().random}

html=requests.get(url=url,headers=headers).text
print(html)
"""
{
  "args": {},
  "headers": {
    "Accept": "*/*",
    "Accept-Encoding": "gzip, deflate",
    "Host": "httpbin.org",
    "User-Agent": "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:17.0) Gecko/20100101 Firefox/17.0.6",
    "X-Amzn-Trace-Id": "Root=1-5ff7a4de-05f8d7bf49dfe85e3be31d79"
  },
  "origin": "1.192.244.128",
  "url": "http://httpbin.org/get"
}
"""

“User-Agent”: “Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:17.0) Gecko/20100101 Firefox/17.0.6”

添加好headers的参数就可以了吗?
这还不行吗!!!

一些网站不但检测 请求头
一个IP 异常请求频繁(访问频率过多) 封禁?

“origin”: “1.192.244.128”

这是一项就是博主的IP, 我怎么确定这就是我的IP呢?
查询一下:果真如此

找个免费的代理IP来包装下

1. 定义
   代替你原来的IP地址去对接网络的IP地址
2. 作用
   隐藏自身真实IP, 避免被封
3. 获取代理IP网站
   快代理、全网代理、代理精灵、… …

这次加上代理IP再去请求下

import requests
from fake_useragent import UserAgent
url='http://httpbin.org/get'
headers={'User-Agent':UserAgent().random}

# 参数类型
# proxies
# proxies = {'协议': '协议://IP:端口号'}
proxies = {
            'http': 'http://{}'.format('8.129.28.247:8888'),
            'https': 'https://{}'.format('8.129.28.247:8888'),
        }

html=requests.get(url=url,headers=headers,proxies=proxies).text
print(html)
"""
{
  "args": {},
  "headers": {
    "Accept": "*/*",
    "Accept-Encoding": "gzip, deflate",
    "Host": "httpbin.org",
    "User-Agent": "Mozilla/5.0 (Windows NT 6.2; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1667.0 Safari/537.36",
    "X-Amzn-Trace-Id": "Root=1-5ff7a71d-10b181340f8dc04f7514dfba"
  },
  "origin": "8.129.28.247",
  "url": "http://httpbin.org/get"
}
"""

“origin”: “8.129.28.247”

这次和我们加入的IP一样, 而不是我们自己的IP直接去请求了
但这一个够用吗?

别着急往下看啦~~~

建立代理IP池

构建一个IP池
每次让它随机提供一个来请求不就解决一个IP请求频繁而导致封掉了
这样就达到我们想要的结果了

• 定义一个测试函数

import requests
from fake_useragent import UserAgent

test_url = 'http://httpbin.org/get'

headers = {'User-Agent': UserAgent().random}

#                      参数 IP 地址
def test_proxy(proxy):
    '''测试代理IP是否可用'''
    proxies = {
        'http': 'http://{}'.format(proxy),
        'https': 'https://{}'.format(proxy),
    }
    # 参数类型
    # proxies
    # proxies = {'协议': '协议://IP:端口号'}
    # timeout 超时设置 网页响应时间3秒 超过时间会抛出异常
    try:
        resp = requests.get(url=test_url, proxies=proxies, headers=headers, timeout=3)
        
        # 查看状态码   
        if resp.status_code == 200:
            print(proxy, '\033[31m可用\033[0m')
            
        else:
            print(proxy, '不可用')


    except Exception as e:
        print(proxy, '不可用')

get一下HTTP状态码

完整代码

• 本次涉及到xpath对此不了解的同学可以去了解博主之前的博客
• 也可以去菜鸟教程自行了解一下
• 本次以 快代理 提供的免费IP 获取为例

# 建立属于自己的开放代理IP池
import requests
import random
import time
from lxml import etree
from fake_useragent import UserAgent

class IpPool:
    def __init__(self):
        # 测试ip是否可用url
        self.test_url = 'http://httpbin.org/get'
        # 获取IP的 目标url
        self.url = 'https://www.89ip.cn/index_{}.html'

        self.headers = {'User-Agent': UserAgent().random}
        # 存储可用ip
        self.file = open('ip_pool.txt', 'wb')

    def get_html(self, url):
        '''获取页面'''
        html = requests.get(url=url, headers=self.headers).text

        return html

    def get_proxy(self, url):
     	'''数据处理  获取ip 和端口''' 
        html = self.get_html(url=url)
        # print(html)
       
        elemt = etree.HTML(html)
        
        ips_list = elemt.xpath('//table/tbody/tr/td[1]/text()')
        ports_list = elemt.xpath('//table/tbody/tr/td[2]/text()')

        for ip, port in zip(ips_list, ports_list):
            # 拼接ip与port
            proxy = ip.strip() + ":" + port.strip()
            # print(proxy)
            
            # 175.44.109.195:9999
            self.test_proxy(proxy)

    def test_proxy(self, proxy):
        '''测试代理IP是否可用'''
        proxies = {
            'http': 'http://{}'.format(proxy),
            'https': 'https://{}'.format(proxy),
        }
        # 参数类型
        # proxies
        # proxies = {'协议': '协议://IP:端口号'}
        # timeout 超时设置 网页响应时间3秒 超过时间会抛出异常
        try:
            resp = requests.get(url=self.test_url, proxies=proxies, headers=self.headers, timeout=3)
           # 获取 状态码为200 
            if resp.status_code == 200:
                print(proxy, '\033[31m可用\033[0m')
                # 可以的IP 写入文本以便后续使用
                self.file.write(proxy)
                
            else:
                print(proxy, '不可用')

        except Exception as e:
            print(proxy, '不可用')

    def crawl(self):
        '''执行函数'''
        # 快代理每页url 的区别
        # https://www.kuaidaili.com/free/inha/1/
        # https://www.kuaidaili.com/free/inha/2/
        # .......
		# 提供的免费ip太多
        # 这里只获取前100页提供的免费代理IP测试
        for i in range(1, 101):
            # 拼接完整的url
            page_url = self.url.format(i)
            # 注意抓取控制频率
            time.sleep(random.randint(1, 4))
            self.get_proxy(url=page_url)

        # 执行完毕关闭文本
        self.file.close()


if __name__ == '__main__':
    ip = IpPool()
    ip.crawl()

测试完这里博主的脸可能比较黑吧, 竟没几个可以用的!!!
由于提供的免费IP可用的机率很小
想构建自己的IP池的小伙伴, 可以去获取其它代理, 提供的免费代理IP.

这里给大家提供几个博主当时测试时可以使用的IP

159.203.44.177:3128

203.202.245.62:80

8.210.88.234:3128

89.187.177.106:80

89.187.177.106:80

96.113.165.182:3128

IP的响应速度根据你机器所在的地理位置不同而有差异

作者：淮南子.
来源：CSDN
版权声明：本文为博主原创文章，原创不易，请尊重原创转载请附上博文链接！