设计思想:
一、具体网站具体对待,请求成功与否与网站的差异很大:
相同的代理不同的网站的会给予不同的反馈,有的网站能请求成功,但是有的网站却不行;
所以在这里校验器接受一个形参url,增强对于不同网站的扩展性;
请求成功未必也能够返回正确的数据;这里有衍生到了校验多种校验方式:
二、超时校验:
此种校验较为简单,如果超时了,那么这个代理就不可用,timeout即可判断,在校验器模块内集成超时异常判断方法即可;
三、隐匿失败校验:
有些代理不是高匿代理,那么这些代理加了也和没加差不多,这里校验的方式也较为简单,为了确保自己是带着套出去的,那么就得有个判断,这里校验器接受一个形参ip,这个形参ip指的是你的公网ip,校验器内部拿着这个ip和和代理查询网站查询的数据进行对比,如果是一样的那么说明隐匿失败;提供一个比较好的查询接口:
http://api.ipify.org
返回就是一个一个字符串形式的ip,不含任何网页标签;
四、服务器响应数据有效性的校验:
有时候虽然你请求成功了,代理或许也是高匿的,但是,服务器返回的有可能一个报错信息页面比如 xxxxxx抱歉页面无法访问,所以这一块就要校验返回的数据是否包含你要的字段;
关于代理池:
最好动态刷新缓存
如果代理池是动态的:
需要考虑一个问题,上一次刷新的缓存要不要保留,在代理失败切换代理的时候是不是直接刷新缓存;答案是肯定的,每一次失败后都刷新缓存,道理很简单,代理ip有的时效性非常短,上一秒有效的不一定下一秒有效;
一方面提供代理池的接口会有校验,会过滤掉垃圾代理,最新返回的是经过校验有效的;
另一反面及时代理池接口没有加经过校验,其返回的最新数据可用心也是比较大的,越新越好;
提醒:此种每次失败刷新缓存的方法,适用于接口没有调用频率限制的情况。
如果代理池是静态的:
还是会有一个问题,是随机获取呢,还是遍历呢?
随机比较好,遍历会有一个问题,打个比方,如果第一个代理是有效的,那么每回只会用第一个代理,因为校验器设计上就是如果以上三个校验都是没毛病的就返回response;其他的都没机会了,对于目标站点来说伪装度就不够了,失去了代理的价值,每次请求都是一个ip那不就被轻易识破了吗!
所以这里用random.choice(),每回都是随机选一个
关于代理为空的情况:
解决思路:循环三次刷新缓存
如果代理池为空那么有两种解决方式,一种是切换代理池,另一种是取消代理;
切换代理池:如果有备用代理池,那么最好是切换代理池了,
取消代理:如果没有备用代理池,只能取消代理
设计实现:
动态池:
尝试失败就刷新缓存
静态池:
分为两种,一种是纯静态,一种是限制请求频率的api代理池;这边纯静态在代码内忽略,函数设置形参api接受api;对于限制请求评率的代理池,先循环遍历每一个代理,如果全部失败,再次调用api遍历,最多调用三次,如果三次都失败那么取消代理;
外层循环,
第0层校验代理池是否为空,如果为空循环3次刷新缓存;再次失败,返回无代理响应
第一层校验隐匿是否成功,失败后,循环一个代理池的长度;再次失败,返回无代理响应
第二层校验超时,失败循环一个代理池的长度;再次失败,返回无代理响应
第三层校验是否返回正确的数据,如果失败循环一个代理池的长度;再次失败,返回无代理响应
一共四层校验
如果所有的校验都失败,那么取消代理返回无代理响应
提醒:这里的代理api是写死的,因为不同的api返回的数据结构不一样,解析规则不同,所以其他代理api调用该校验器需要修改源码的解析规则
import requests
import random
"""快代理接口"""
# 代理校验类
class Proxy_Test:
# 接受ip,api,field,url,headers='',status=1留个参数
def __init__(self,ip,field,url,headers='',status=1):
# 您的公网ip
self.ip = ip
# 您调用代理池的api
self.api = 'http://dev.kdlapi.com/api/getproxy/?orderid=(你的账户id)&num=10&b_pcchrome=1&b_pcie=1&b_pcff=1&protocol=1&method=2&an_an=1&an_ha=1&sp1=1&sp2=1&sep=1'
# 您的代理池类型,动态还是静态;接受参数类型,int, 1表示动态,0表示静态
self.status = status
# 您的目标url源码中,如果请求成功包含的字段;
self.field = field
# 目标站点的url
self.url = url
# 目标站点的请求头
self.headers = headers
# 第二层校验计数器
self.num2 = 0
# 静态代理池,计数器,对刷新缓存次数计数
self.num3 = 1
# 代理校验,返回response,代理成功返回加代理的response,失败则无代理response
def proxy_test(self):
status = 1
# 第一层校验计数器
num1 = 0
# 第三层校验计数器
num4 = 0
# 第四层校验计数器
num5 = 0
while True:
# 刷新缓存
if status == 1:
# 返回一个带有proxies key的json value 为列表,,,,这一块是写死的api
self.proxy_list = requests.get(self.api).text.split('\r\n')
# print(self.proxy_list)
try:
# 第一层校验代理池是否为空,不为空返回一个随机proxies
get_pool = self.get_pool()
proxies = get_pool['proxies']
length = get_pool['length']
print('ok')
# 判断是否能低延时高匿成功
try:
# print('高匿不验了')
# 第二层校验,是否高匿成功
response_ip = requests.get('http://api.ipify.org',
proxies = proxies,
timeout=4,verify=False).text
# 校验有没有高匿成功,ip为您本机的公网ip
if response_ip == self.ip:
1 / 0
# 第三层校验,是否超时
try:
response = self.wrapper_proxy(proxies)
# 第四层校验,是否正确响应
try:
content = response.text
# print(content)
if self.field in content:
print('恭喜,代理成功!')
li = [1,response]
return li
else:
1/0
except:
li = self.failed_retry(status, length, '正确响应校验')
if li:
return li
except:
li = self.failed_retry(status, length, '超时校验')
if li:
return li
except:
li = self.failed_retry(status,length,'高匿')
if li:
return li
except:
# 循环三次刷新缓存
num1 += 1
if num1 > 3:
# 三次都失败后,不使用代理返回response
print('代理池为空,不使用代理')
response = self.remove_proxy()
li = [0,response]
return li
print("代理数据库为空,循环刷新------第{}次".format(num1))
# 返回随机proxy
def get_pool(self):
# 随机选择,如果是空抛出异常
proxy = random.choice(self.proxy_list)
# 代理池的数量;
length = len(self.proxy_list)
# print('代理池长度{}'.format(length))
proxies = {'http': 'http://{}'.format(proxy)}
result = {'proxies':proxies,'length':length}
print(result)
return result
# 不设置代理返回response
def remove_proxy(self):
# 通过判断headers是否传递来决定是否加上请求头
if self.headers == '':
response = requests.get(self.url,timeout=20)
return response
else:
response = requests.get(url=self.url,headers = self.headers,timeout=20)
return response
# 添加代理请求
def wrapper_proxy(self,proxies):
# 通过判断headers是否传递来决定是否加上请求头
if self.headers == '':
response = requests.get(self.url,proxies = proxies,timeout=20)
return response
else:
response = requests.get(url=self.url,headers = self.headers,proxies=proxies,timeout=20)
return response
# 第二层以及以后的失败重试
def failed_retry(self,status,length,level):
# 当失败次数大于等于代理池的长度的时候,取消代理返回response
self.num2 += 1
print('{}失败{}次'.format(level,self.num2))
# 如果是静态代理池
if self.status == 0:
# 修改status状态,不刷新缓存
status = 0
# 遍历完静态池 有刷新三次缓存的机会,机会用完取消代理返回response
if status == 0:
if self.num2 >= length:
self.num3 += 1
if self.num3 < 4:
status = 1
self.num2 = 0
else:
print('{}失败,取消代理'.format(level))
response = self.remove_proxy()
li = [0,response]
return li
if status == 1:
if self.num2 >= length:
self.num2 = 0
print('{}失败,取消代理'.format(level))
response = self.remove_proxy()
li = [0,response]
return li
if __name__ == '__main__':
# 接受ip,api,field,url,headers='',status=1留个参数
api = 'http://localhost:8899/api/v1/proxies?anonymous=true'
ip = '211.161.244.119'
field = '学习网络爬虫'
url = 'https://blog.csdn.net/winterto1990/article/details/51220307'
headers = {"Host":"blog.csdn.net",
"User-Agent":"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/67.0.3396.62 Safari/537.36"
}
proxy = Proxy_Test(ip=ip,field=field,url=url,headers=headers)
response = proxy.proxy_test()[1]
print(response.text)
网友评论