代理池的搭建有很多免费的开源的项目，这个具体我也看过，但是我是个小白，看不懂，如果要写也写不除来，那么这篇文章就来详细说明一下搭建的步骤.
一个合格的代理池必须拥有一个爬取代理IP的爬取器、一个验证IP可否使用的校验器、一个存储IP的数据库、调用这些的调度器以及可以供获取IP的接口（这里推荐flask，比较简单）。
代理爬取器
爬取器，首先要爬取的代理IP网站尽量是无需登录的，其次是对代理IP更新较快的，前者加快代理池的效率，后者增加代理池的质量。这里我对市面上部分代理网站进行爬取，当然一些常用的代理IP网站提供IP质量不高，比如西刺无忧66这些经常被爬取（西刺偶尔还会崩溃，估计是爬取的人有些多，网站保护性503），话说基本上每次谈到免费的ip，首选就是西刺代理呐😀，所以我们也简单做个演示，自己搭建的时候可选可不选。
那么我们使用requests来请求网址，使用PyQuery库或者正则都可以解析数据，创建crawl_ip.py文件用来写爬取器，具体代码参考:
def crawl_xici():
url='http://www.xicidaili.com/{}'
items=[]
for page in range(1,2):
items.append(('wt/{}'.format(page),'http://{}:{}'))
items.append(('wn/{}'.format(page),'https://{}:{}'))
#items列表中添加(wt/{},'http://{}:{}')
for item in items:
proxy_type,host=item
html=requests.get(url.format(proxy_type),headers=HEADERS).text#组合url
if html:
doc=py(html)
for proxy in doc('table tr').items():#doc('table tr')为对象
ip=proxy('td:nth-child(2)').text()#查找第二个td标签，取文本
port=proxy('td:nth-child(3)').text()#同理
if ip and port:
yield host.format(ip,port)#补全host的内容
def crawl_66ip():
"""
66ip代理:http://www.66ip.cn
可以使用json接口提取
http://www.66ip.cn/nmtq.php?getnum=100&isp=0&anonymoustype=0&area=0&proxytype={}&api=66ip
getnum为提取数量
proxytype为协议类型
:return:
"""
url='http://www.66ip.cn/nmtq.php?getnum=100&isp=0&anonymoustype=0&area=0&proxytype={}&api=66ip'
pattern="\d+\.\d+.\d+\.\d+:\d+"#定义正则提取代理
items=[(0,'http://{}'),(1,'https://{}')]
for item in items:
proxy_type,host=item
html=requests.get(url.format(proxy_type),headers=HEADERS).text
if html:
for proxy in re.findall(pattern,html):
yield host.format(proxy)
def crawl_kuaidaili():
"""
快代理:https://www.kuaidaili.com/free/inha/1
:param headers:
:return:
"""
url='https://www.kuaidaili.com/free/inha/{}/'
items=[p for p in range(1,2)]
for page in items:
html=requests.get(url.format(page),headers=HEADERS).text
if html:
doc=py(html)
for proxy in doc('table tr').items():
ip=proxy('td:nth-child(1)').text()
port=proxy('td:nth-child(2)').text()
if ip and port:
yield 'http://{}:{}'.format(ip,port)#只有http协议的代理
def crawl_ip3366():
"""
云代理:
:param headers:
:return:
"""
url='http://www.ip3366.net/?stype=1&page={}'
items=[p for p in range(1,2)]
for page in items:
html=requests.get(url.format(page),headers=HEADERS).text
if html:
doc=py(html)
for proxy in doc('table tr').items():
ip = proxy("td:nth-child(1)").text()
port = proxy("td:nth-child(2)").text()
proxy_type = proxy("td:nth-child(4)").text()
if ip and port and proxy_type:
yield "{}://{}:{}".format(proxy_type.lower(),ip,port)
解释一下代码，返回的代理一般都是(http/https)://ip:port格式的代理。那么爬取的方式就是对于一些免费代理网站分别进行抓取，封装成不同的函数来爬取，你也可以添加更多的免费代理网站。
你可以先创建一个settings.py文件，用来存放配置信息，具体代码参考:
HEADERS={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}
REQUEST_TIMEOUT=2
REQUEST_DELAY=2
#redis地址
REDIS_HOST='localhost'
#reids端口
REDIS_PORT=6379
#reids密码
REDIS_PASSWORD=None
#redis的集合名字
REDIS_KEY='myproxy'
#reids 连接池最大连接量
REDIS_MAX_CONNECTION=20
#REIDS SCORE 最大分数
MAX_SCORE=10
#REDIS SCORE 最小分数
MIN_SCORE=0
#REDIS SCORE 初始分数
INIT_SCORE=5
代理抓取写好应该完成代理池搭建的30%了，哈哈……之后我们就要尝试保存进数据库了，这里推荐基本上所有的代理池项目都会使用redis数据库，我们小白也不列外，毕竟其中的有序集合类型（sorted set）非常适合代理池cookies池的搭建。其中是有score的，也就是我们存入一个代理的同时也要给它一个分数，这方便我们之后对其校验以及取代理IP的优先级。那么我们创建redis_conn.py文件来写一个连接redis数据库的类,具体代码参考:
class RedisClient:
"""
redis 有序集合zset 权重即为分数
zadd 按照权重值添加
zrange 根据索引查看值
zrangebyscore 按照权重查看值
zremerangebyscore 根据值的权重删除
"""
def __init__(self,host=REDIS_HOST,port=REDIS_PORT,password=REDIS_PASSWORD):
client=redis.ConnectionPool(
host=host,
port=port,password=password,max_connections=REDIS_MAX_CONNECTION,
)
self.redis=redis.Redis(connection_pool=client)#创建redis连接实例
def add_proxy(self,proxy,score=INIT_SCORE):
"""
新增一个代理，初始化分数 INIT_SCORE < MAX_SCORE，确保在
运行完收集器后还没运行校验器就获取代理，导致获取到分数虽为 MAX_SCORE,
但实际上确是未经验证，不可用的代理
:param proxy:
:param score:
:return:
"""
if not self.redis.zscore(REDIS_KEY,proxy):
self.redis.zadd(REDIS_KEY,proxy,score)
def reduce_proxy_score(self,proxy):
"""
验证未通过，分数减一
:param proxy:
:return:
"""
score=self.redis.zscore(REDIS_KEY,proxy)
if score and score>MIN_SCORE:
self.redis.zincrby(REDIS_KEY,proxy,-1)
else:
self.redis.zrem(REDIS_KEY,proxy)
def increase_proxy_score(self,proxy):
"""
验证通过，分数加1
:param proxy:
:return:
"""
score=self.redis.zscore(REDIS_KEY,proxy)
if score and score<MAX_SCORE:
self.redis.zincrby(REDIS_KEY,proxy,1)
def pop_proxy(self):
"""
返回一个代理
:return:
"""
#第一次尝试取分数最高，就是最新可用的代理
fist_chance=self.redis.zrangebyscore(REDIS_KEY,MAX_SCORE,MAX_SCORE)
if fist_chance:
return random.choice(fist_chance)
else:
#第二次尝试取7-10分数的任意一个代理
second_chance=self.redis.zrangebyscore(
REDIS_KEY,MAX_SCORE-3,MAX_SCORE
)
if second_chance:
return random.choice(second_chance)
#最后一次随便取
else:
last_chance=self.redis.zrangebyscore(REDIS_KEY,MIN_SCORE,MAX_SCORE)
if last_chance:
return random.choice(last_chance)
def get_proxies(self,count=1):
"""
返回指定数量的代理，分数由高到低
:param count:
:return:
"""
proxies=self.redis.zrevrange(REDIS_KEY,0,count-1)
for proxy in proxies:
yield proxy.decode('utf-8')
def count_all_proxy(self):
"""
返回所有代理总数
:return:
"""
return self.redis.zcard(REDIS_KEY)
def count_score_proxies(self,score):
"""
返回指定分数总代理
:param score:
:return:
"""
if 0<=score<=10:
proxies=self.redis.zrangebyscore(REDIS_KEY,score,score)
return len(proxies)
return -1
def clear_proxy(self,score):
"""
删除分数小于等于score的代理
:param score:
:return:
"""
if 0<=score<=10:
proxies=self.redis.zrangebyscore(REDIS_KEY,0,score)
for proxy in proxies:
self.redis.zrem(REDIS_KEY,proxy)
return True
return False
def all_proxy(self):
"""
返回全部代理
:return:
"""
return self.redis.zrangebyscore(REDIS_KEY,MIN_SCORE,MAX_SCORE)
这就是写的对redis数据库的操作，至于我们如何把爬取的IP放入，就需要改一下我们的爬取器了，封装成一个类会比我写这样的方法好得多代码参考:
import requests,re
from pyquery import PyQuery as py
from settings import * #settings配置文件
from redis_conn import RedisClient#引入redis_conn.py中的RedisClient类
redis_con=RedisClient()#实例化，实现爬取存储
all_func=[]
def colection_funcs(func):
"""
装饰器，用于收集爬虫函数
:return:
"""
all_func.append(func)
return func
class Crawler:
"""
返回格式:http://host:port
"""
@staticmethod#将函数转换为静态方法
def run():
"""
启动收集器
:return:
"""
for func in all_func:
for proxy in func():
redis_con.add_proxy(proxy)#调用RedisClient实例中添加代理方法
@colection_funcs#添加之前写的爬取ip方法，在每个方法之前声明装饰器,会吧
网上有许多检验IP的方法，诸如requests.get telnet之类的，我们这里使用requests.get 去请求百度，看状态码，成功的话就在redis数据库中对相应的ip的分数加一，反之减一，为什么不先验证在存储到redis数据库中呢，我刚开始也是这么想的，据说，是由于这个ip可能在验证的时候失败，过段时间之后又可以了，所以就需要把ip加上那个分数，分数最低之时，就是删掉ip之时…..代码参考:
最后就是整个项目如何运行了，我们单纯跑调度器肯定是不合适的，因为这样redis中有代理了，但是我们要在爬虫项目中连接redis来获取代理，这一点是比较麻烦的，而且每次运行都要在本地跑一次代理池，这样肯定不符合程序员偷懒的初衷。正确的做法是做一个api接口，然后部署到云端，让云服务器爬取代理IP，这样我们就每次访问api接口就能得到我们的数据了，下面来写写接口，这里用的是Flask，因为简单
这个接口比较简单，实现了查看代理池数量，获取代理IP（推荐pop因为使用后就可以将代理IP删除）当然有许多低分的代理我们也可以通过接口调用将其删除。出于安全性，其实这个api应该加个校验的，不然别人一直用你的代理池就白费功夫了。
Python中代理的验证可以使用的是telnetlib.Telnet方法，爬取免费的西刺代理，具体代码可参考：
import redis,urllib.request
from bs4 import BeautifulSoup
import telnetlib
client=redis.Redis()
for i in range(1,3):#爬取前两页
url='https://www.xicidaili.com/nn/%d/'% i
req=urllib.request.Request(url)
req.add_header('User-Agent','Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1; TencentTraveler 4.0)')
response=urllib.request.urlopen(req)
html=response.read()
bsobj=BeautifulSoup(html,'lxml')
for i in range(100):
speed=float(bsobj.select('td')[6+i*10].div.get('title').replace('秒',''))#抓取速度
if speed<0.2:
ip=bsobj.select('td')[1+i*10].get_text()#ip
port=bsobj.select('td')[2+i*10].get_text()#端口
ssl=bsobj.select('td')[5+i*10].get_text().lower()#协议，需转换为小写
ip_address=ssl+'://'+ip+':'+port #拼接代理
try:
telnetlib.Telnet(host=ip,port=port,timeout=3)#验证
except:
print('fail')
else:
print('sucess:'+ip_address)#输出成功的代理那么我们再通过上面的代码进行测试的时候发现，验证成功的代理是有问题的，所以我们需要重新修改验证代理的代码，具体就是通过代理去请求某个网址，看状态码，具体代码参考:
声明：本站所有文章，如无特殊说明或标注，均为本站原创发布。任何个人或组织，在未征得本站同意时，禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益，可联系我们进行处理。
