一、写在前面
有时候你的爬虫刚开始的时候可以正常运行，能够正常的爬取数据，但是过了一会，却出现了一个“403 Forbidden"，或者是”您的IP访问频率太高“这样的提示，这就意味着你的IP被ban了，好一点的情况是过一段时间你就能继续爬取了，坏一点的情况就是你的IP已经进入别人的黑名单了，然后你的爬虫就GG了。怎么办呢？我们可以通过设置代理来解决，付费代理的效果自然不必多说，但是对于学习阶段的人来说，我觉得爬取网上的免费代理来用是一个更好的选择，而这一篇博客就将教你怎么利用免费代理搭建属于你自己的代理池。
二、目标分析
要搭建一个代理池，需要三个模块：存储模块、爬取模块和测试模块。
存储模块：负责存储我们爬取下来的代理，首先我们需要保证这些代理不能有重复的，然后我们还要对代理是否可用进行标记，这里可以使用Redis数据库的SortedSet（有序集合）进行存储。
爬取模块：负责对一些提供免费代理的网站进行爬取，代理的形式是IP+端口，爬取下来之后保存到数据库里。
测试模块：负责对代理池中的代理的可用性进行测试，由于测试出错不一定就表明代理不可用，可能是因为网络问题或者请求超时等等，所以我们可以设置一个分数标识，100分标识可用，分数越低标识可用性越低，当分数低于一个阈值之后，就从代理池中移除。
三、具体实现
1、存储模块
这里使用的是Redis的有序集合。Redis 有序集合和集合一样也是string类型元素的集合，且不允许重复的成员，不同的是每个元素都会关联一个double类型的分数，Redis正是通过分数来为集合中的成员进行从小到大的排序。这样我们保存到数据库中的元素就是一个代理和一个分数，比如112.17.65.133:8060和100，这就表示112.17.65.133:8060这个代理是可用的。
对于新添加到数据库中的代理，设置的初始分数为10，添加之后会进行一次测试，如果可用就把分数改为100表明可用，如果测试的结果是不可用就把分数减1，当分数减到0后就从代理池中移除。这么做的意义在于一次测试不可用并不能代表这个代理完全不可用，有可能在之后的某次测试中是可用的，这样我们就减小了将一个原本可用的代理从代理池中移除出去的概率。
当我们想要从代理池中获取一个代理的时候，优先从分数为100的代理中随机获取，如果一个100分的代理都没有，则对所有代理进行排序，然后随机获取一个代理。由于我们使用的是随机获取，这样就能保证代理池中的所有代理都有被获取的可能性。
现在我们需要定义一个类来实现这个有序集合，还需要设置一些方法来实现添加代理、修改分数、获取代理等功能。具体代码如下：
"""
Version: Python3.5
Author: OniOn
Site: http://www.cnblogs.com/TM0831/
Time: 2019/2/12 14:54
"""
import redis
import random MAX_SCORE = 100 # 最高分
MIN_SCORE = 0 # 最低分
INITIAL_SCORE = 10 # 初始分数
REDIS_HOST = "localhost"
REDIS_PORT = 6379 class RedisClient:
def __init__(self):
self.db = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=0)
self.key = "proxies" def add(self, proxy, score=INITIAL_SCORE):
"""
将代理添加到代理池中
:param proxy: 代理
:param score: 分数
:return:
"""
if not self.is_exist(proxy):
self.db.zadd(self.key, proxy, score) def is_exist(self, proxy):
"""
判断代理池中是否存在该代理
:param proxy: 代理
:return: True or False
"""
if self.db.zscore(self.key, proxy):
return True
else:
return False def random(self):
"""
获取有效代理，先获取最高分代理，如果不存在，则按分数排名然后随机获取
:return: 代理
"""
result = self.db.zrangebyscore(self.key, MAX_SCORE, MAX_SCORE)
if len(result):
return random.choice(result)
else:
result = self.db.zrangebyscore(self.key, MIN_SCORE, MAX_SCORE)
if len(result):
return random.choice(result)
else:
print("代理池已空！") def decrease(self, proxy):
"""
代理分数减1分，若小于最低分，则从代理池中移除
:param proxy:
:return:
"""
if self.is_exist(proxy):
score = self.db.zscore(self.key, proxy)
if score > MIN_SCORE:
score -= 1
self.db.zadd(self.key, proxy, score)
else:
self.delete(proxy) def max(self, proxy):
"""
将代理分数设置为最高分
:param proxy: 代理
:return:
"""
if self.is_exist(proxy):
self.db.zadd(self.key, proxy, MAX_SCORE) def delete(self, proxy):
"""
从代理池中移除该代理
:param proxy: 代理
:return:
"""
if self.is_exist(proxy):
self.db.zrem(self.key, proxy) def all(self):
"""
获取代理池中的所有代理
:return:
"""
if self.count():
return self.db.zrange(self.key, MIN_SCORE, MAX_SCORE) def count(self):
"""
获取代理池中代理数量
:return:
"""
return self.db.zcard(self.key)
2、爬取模块
爬取模块比较简单，就是定义一个Crawler类来对一些提供免费代理的网站进行爬取。具体代码如下：
"""
Version: Python3.5
Author: OniOn
Site: http://www.cnblogs.com/TM0831/
Time: 2019/2/12 15:07
"""
import requests
from lxml import etree
from fake_useragent import UserAgent # 设置元类
class CrawlMetaClass(type):
def __new__(cls, name, bases, attrs):
attrs['__CrawlFuncCount__'] = 0
attrs['__CrawlFunc__'] = []
for k, v in attrs.items():
if 'crawl_' in k:
attrs['__CrawlFunc__'].append(k)
attrs['__CrawlFuncCount__'] += 1
# attrs['__CrawlFuncCount__'] = count
return type.__new__(cls, name, bases, attrs) class Crawler(object, metaclass=CrawlMetaClass):
def __init__(self):
self.proxies = [] # 代理列表
ua = UserAgent() # 使用随机UA
self.headers = {
"UserAgent": ua.random
} def get_proxies(self, callback):
"""
运行各个代理爬虫
:param callback: crawl函数名称
:return:
"""
for proxy in eval("self.{}()".format(callback)):
print("成功获取代理：", proxy)
self.proxies.append(proxy)
return self.proxies def crawl_kdd(self):
"""
快代理爬虫
:return:
"""
urls = ["https://www.kuaidaili.com/free/inha/{}/".format(i) for i in range(1, 4)]
for url in urls:
res = requests.get(url, headers=self.headers)
try:
et = etree.HTML(res.text)
ip_list = et.xpath('//*[@data-title="IP"]/text()')
port_list = et.xpath('//*[@data-title="PORT"]/text()')
for ip, port in zip(ip_list, port_list):
yield ip + ":" + port
except Exception as e:
print(e) def crawl_89ip(self):
"""
89IP爬虫
:return:
"""
urls = ["http://www.89ip.cn/index_{}.html".format(i) for i in range(1, 4)]
for url in urls:
res = requests.get(url, headers=self.headers)
try:
et = etree.HTML(res.text)
ip_list = et.xpath('//*[@class="layui-table"]/tbody/tr/td[1]/text()')
port_list = et.xpath('//*[@class="layui-table"]/tbody/tr/td[2]/text()')
ip_list = [i.strip() for i in ip_list]
port_list = [i.strip() for i in port_list]
for ip, port in zip(ip_list, port_list):
yield ip + ":" + port
except Exception as e:
print(e) def crawl_xc(self):
"""
西刺代理爬虫
:return:
"""
url = "https://www.xicidaili.com/?t=253"
res = requests.get(url, headers=self.headers)
try:
et = etree.HTML(res.text)
ip_list = et.xpath('//*[@id="ip_list"]/tr[3]/td[2]/text()')
port_list = et.xpath('//*[@id="ip_list"]/tr[3]/td[3]/text()')
for ip, port in zip(ip_list, port_list):
yield ip + ":" + port
except Exception as e:
print(e)
可以看到几个爬虫的方法名称都是以crawl开头的，主要是为了方便，如果要添加新的爬虫就只用添加crawl开头的方法即可。
这里我写了爬取快代理、89IP代理和西刺代理的爬虫，都是用xpath进行解析，也都定义了一个生成器，然后用yield返回爬取到的代理。然后定义了一个get_proxies()方法，将所有以crawl开头的方法都调用一遍，获取每个方法返回的结果并生成一个代理列表，最后返回这个代理列表。那么如何获取crawl开头的方法呢？这里借用了元类来实现。首先定义一个类CrawlMetaClass，然后实现了__new__()方法，这个方法的第四个参数attrs里包含了类的一些属性，所以我们可以遍历attrs中包含的信息，就像遍历一个字典一样，如果方法名以crawl开头，就将其添加到__CrawlFunc__中，这样我们就能获取crawl开头的方法了。
我们已经定义好了爬取的方法了，但是还需要定义一个类来执行这些方法，这里可以定义一个GetProxy类来实现爬取代理并保存到代理池中，具体代码如下：
"""
Version: Python3.5
Author: OniOn
Site: http://www.cnblogs.com/TM0831/
Time: 2019/2/14 12:19
"""
from ProxyPool.crawl import Crawler
from ProxyPool.pool import RedisClient class GetProxy:
def __init__(self):
self.crawler = Crawler()
self.redis = RedisClient() def get_proxy(self):
"""
运行爬虫爬取代理
:return:
"""
print("[INFO]Crawl Start...")
count = 0
for callback_label in range(self.crawler.__CrawlFuncCount__):
callback = self.crawler.__CrawlFunc__[callback_label]
# 获取代理
proxies = self.crawler.get_proxies(callback)
for proxy in proxies:
self.redis.add(proxy)
count += len(proxies)
print("此次爬取的代理数量为：{}".format(count))
print("[INFO]Crawl End...\n\n")
3、测试模块
我们已经将代理成功爬取下来并保存到代理池中了，但是我们还需要对代理的可用性进行测试。测试的方法就是使用requests库设置代理并发送请求，如果请求成功并且返回的状态码是200的话，就表明这个代理是可用的，然后我们就要将该代理的分数设置为100，反之如果出现请求失败、请求超时或者返回的状态码不是200的话， 就要将该代理的分数减1，如果分数等于0了，就要从代理池中移除。
这里可以定义一个TestProxy类来实现，具体代码如下：
"""
Version: Python3.5
Author: OniOn
Site: http://www.cnblogs.com/TM0831/
Time: 2019/2/14 14:24
"""
import time
import random
import requests
from fake_useragent import UserAgent
from ProxyPool.crawl import Crawler
from ProxyPool.pool import RedisClient class TestProxy:
def __init__(self):
self.crawler = Crawler()
self.redis = RedisClient()
ua = UserAgent() # 使用随机UA
self.headers = {
"UserAgent": ua.random
} def test(self):
"""
测试函数，测试代理池中的代理
:return:
"""
proxy_list = self.redis.all()
proxy_list = [i.decode('utf-8') for i in proxy_list] # 字节型转字符串型 print("[INFO]Test Start...")
for proxy in proxy_list:
self.request(proxy)
print("[INFO]Test End...\n\n") def request(self, proxy):
"""
测试请求函数
:param proxy:
:return:
"""
print("当前测试代理：{} 该代理分数为：{}".format(proxy, self.redis.db.zscore(self.redis.key, proxy)))
time.sleep(random.randint(1, 4))
try:
url = "https://www.baidu.com/"
proxies = {
"https": "https://" + proxy
}
res = requests.get(url, headers=self.headers, proxies=proxies, timeout=5) if res.status_code == 200:
print("代理可用，分数设置为100")
self.redis.max(proxy)
else:
print("错误的请求状态码，分数减1")
self.redis.decrease(proxy)
except:
print("代理请求失败，分数减1")
self.redis.decrease(proxy)
四、运行程序
这里我定义了一个Main类来实现，过程为先爬取代理，然后对代理池中的代理进行测试，最后从代理池中获取一个可用代理。具体代码如下：
"""
Version: Python3.5
Author: OniOn
Site: http://www.cnblogs.com/TM0831/
Time: 2019/2/14 14:26
"""
from ProxyPool.pool import RedisClient
from ProxyPool.get import GetProxy
from ProxyPool.test import TestProxy class Main:
def __init__(self):
self.gp = GetProxy()
self.tp = TestProxy()
self.db = RedisClient() def run(self):
"""
运行的主函数，先爬取代理，然后测试，最后获取一个有效代理
:return:
"""
self.gp.get_proxy()
self.tp.test()
proxy = self.db.random()
proxy = proxy.decode('utf-8')
print("从代理池中取出的代理为：{}".format(proxy)) if __name__ == '__main__':
m = Main()
m.run()
运行结果截图如下：
完整代码已上传到GitHub！