爬虫 代理池,Python 代理池,Python爬虫代理池搭建的方法步骤

爬虫 代理池,Python 代理池,Python爬虫代理池搭建的方法步骤

这篇文章主要介绍了计算机编程语言爬虫代理池搭建的方法步骤,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧

目录

一、为什么要搭建爬虫代理池二、搭建思路三、代码实现四、代理测试

一、为什么要搭建爬虫代理池

在众多的网站防爬措施中,有一种是根据互联网协议(互联网协议的缩写)的访问频率进行限制,即在某一时间段内,当某个互联网协议(互联网协议的缩写)的访问次数达到一定的阀值时,该互联网协议(互联网协议的缩写)就会被拉黑、在一段时间内禁止访问。

应对的方法有两种:

1.降低爬虫的爬取频率,避免互联网协议(互联网协议)被限制访问,缺点显而易见:会大大降低爬取的效率。

2.搭建一个互联网协议(互联网协议)代理池,使用不同的互联网协议(互联网协议)轮流进行爬取。

二、搭建思路

1、从代理网站(如:西刺代理、快代理、云代理、无忧代理)爬取代理IP;

2、验证代理互联网协议(互联网协议)的可用性(使用代理互联网协议(互联网协议)去请求指定URL,根据响应验证代理互联网协议(互联网协议)是否生效);

3、将可用的代理互联网协议(互联网协议)保存到数据库;

常用代理网站:西刺代理、云代理、IP海、无忧代理、飞蚁代理、快代理

三、代码实现

工程结构如下:

ipproxy.py

IPProxy代理类定义了要爬取的互联网协议(互联网协议)代理的字段信息和一些基础方法。

# -*-编码:utf-8 -*-

进口关于

导入时间

从设置导入代理统一资源定位器格式化程序

schema _ pattern=re编译(r ' http | https $ ',re .我)

ip _模式=re.compile(r'^([0-9]{1,3}.){3}[0-9]{1,3}$ ',re .我)

港口模式=re.compile(r'^[0-9]{2,5}$',re。我)

IPProxy类:

'''

{

架构:' http ',#代理的类型

ip': '127.0.0.1 ',#代理的互联网协议(互联网协议)地址

端口':' 8050 ',#代理的端口号

used_total': 11,#代理的使用次数

成功次数':5,#代理请求成功的次数

continuous_failed': 3,#使用代理发送请求,连续失败的次数

created_time': '2018-05-02' #代理的爬取时间

}

'''

def __init__(自身,模式,ip,端口,used_total=0,success_times=0,continuous_failed=0

创建时间=无):

'''初始化代理实例' ' '

如果架构==' '或(计划或理论的)纲要为无:

架构='http '

self.schema=schema.lower()

self.ip=ip

self.port=port

self.used _ total=used _ total

self.success_times=成功次数

自我。连续失败=连续失败

如果创建时间为无:

创建时间=时间。strftime(' % Y-% m-% d ',time.localtime(time.time()))

self.created_time=创建时间

def _get_url(self):

'''返回代理' ' ' url ' ' '

返回PROXY _ URL _ FORMATTER % { ' schema ':self。架构,“IP”:self . IP,“port”:self . port }

定义_检查_格式(自身):

'''如果代理字段格式良好,则返回没错,否则返回'假' ' '

如果自我模式不为没有人且self.ip不为没有人且自助端口不为无:

if schema _ pattern。匹配(自我。模式)和ip_pattern.match(自我. ip)和port_pattern.match(self.port):

返回真实的

返回错误的

def _is_https(self):

'''如果代理是https,则返回没错,否则返回'假' ' '

return self.schema=='https '

def _update(self,successed=False):

'''根据请求的响应结果更新代理'''

自我。use _ total=self。已用_总计1

如果成功:

self.continuous_failed=0

自我。成功次数=自己。成功次数1

否则:

打印(自我连续_失败)

自我。continuous _ failed=自我。continued _ failed 1

if __name__=='__main__ ':

proxy=IPProxy('HTTPS ',' 192.168.2.25 ',' 8080 ')

打印(代理. get_url())

打印(代理. check_format())

打印(代理. is_https())

settings.py

settings.py中汇聚了工程所需要的配置信息。

# 指定雷迪斯的主机名和端口

REDIS_HOST='localhost '

REDIS_PORT=6379

# 代理保存到雷迪斯键格式化字符串

proxy _ REDIS _ FORMATTER=' proxy:{ } '

# 已经存在的超文本传送协议代理和HTTPS代理集合

proxy _ REDIS _ EXISTED=' proxy:EXISTED '

# 最多连续失败几次

最大连续次数=3

# 代理地址的格式化字符串

PROXY _ URL _ FORMATTER=' %(schema)s://%(IP)s:%(port)s '

用户代理列表=[

Mozilla/5.0(Windows NT 6.1;WOW64) AppleWebKit/537.1 (KHTML,像壁虎一样)Chrome/22.0.1207.1 Safari/537.1 ',

Mozilla/5.0(X11;十字架i686 2268。111 .0)苹果WebKit/536.11(KHTML,像壁虎)Chrome/20。0 .1132 .57 Safari/536.11 ',

Mozilla/5.0(Windows NT 6.1;WOW64) AppleWebKit/536.6 (KHTML,像壁虎一样)Chrome/20.0.1092.0 Safari/536.6 ',

Mozilla/5.0(Windows NT 6.2)苹果WebKit/536.6(KHTML,像壁虎)铬/20。0 .1090 .0 Safari/536.6 ',

Mozilla/5.0(Windows NT 6.2;WOW64) AppleWebKit/537.1 (KHTML,像壁虎一样)Chrome/19.77.34.5 Safari/537.1 ',

Mozilla/5.0(X11;Linux x86 _ 64)苹果WebKit/536.5(KHTML,像壁虎)Chrome/19.0.1084.9 Safari/536.5 ',

Mozilla/5.0(Windows NT 6.0)苹果WebKit/536.5(KHTML,像壁虎)铬/19。0 .1084 .36英寸Safari/536.5英寸,

Mozilla/5.0(Windows NT 6.1;WOW64) AppleWebKit/536.3 (KHTML,像壁虎一样)Chrome/19.0.1063.0 Safari/536.3 ',

Mozilla/5.0(Windows NT 5.1)苹果WebKit/536.3(KHTML,像壁虎)铬/19。0 .1063 .0 Safari/536.3 ',

Mozilla/5.0(麦金塔;英特尔Mac OS X 10 _ 8 _ 0)苹果WebKit/536.3(KHTML像壁虎一样)Chrome/19.0.1063.0 Safari/536.3 ',

Mozilla/5.0(Windows NT 6.2)苹果WebKit/536.3(KHTML,像壁虎)铬/19。0 .1062 .0 Safari/536.3 ',

Mozilla/5.0(Windows NT 6.1;WOW64) AppleWebKit/536.3 (KHTML,像壁虎一样)Chrome/19.0.1062.0 Safari/536.3 ',

Mozilla/5.0(Windows NT 6.2)苹果WebKit/536.3(KHTML,像壁虎)铬/19。0 .1061 .1个Safari/536.3 ',

Mozilla/5.0(Windows NT 6.1;WOW64) AppleWebKit/536.3 (KHTML,像壁虎一样)Chrome/19.0.1061.1 Safari/536.3 ',

Mozilla/5.0(Windows NT 6.1)苹果WebKit/536.3(KHTML,像壁虎)铬/19。0 .1061 .1个Safari/536.3 ',

Mozilla/5.0(Windows NT 6.2)苹果WebKit/536.3(KHTML,像壁虎)铬/19。0 .1061 .0 Safari/536.3 ',

Mozilla/5.0(X11;Linux x86 _ 64)苹果WebKit/535.24(KHTML,像壁虎)Chrome/19。0 .1055 .1个Safari/535.24英尺,

Mozilla/5.0(Windows NT 6.2;WOW64) AppleWebKit/535.24 (KHTML,像壁虎一样)Chrome/19。0 .1055 .1 Safari/535.24英尺

]

# 爬取到的代理保存前先检验是否可用,默认真实的

PROXY_CHECK_BEFOREADD=True

# 检验代理可用性的请求地址,支持多个

PROXY _ CHECK _ URLS={ ' https ':[' https://Ican hazip。com '],' http':['http://icanhazip.com']}

proxy_util.py

proxy_util.py中主要定义了一些实用方法,例如:代理_到_字典(代理)用来将IPProxy代理实例转换成字典;代理人_从_字典用来将字典转换为IPProxy实例;请求_页面()用来发送请求;_is_proxy_available()用来校验代理互联网协议(互联网协议)是否可用。

# -*-编码:utf-8 -*-

随机导入

导入日志记录

导入请求

从ipproxy导入IPProxy

从设置导入用户代理列表、代理检查统一资源定位器

#设置记录器输出格式

伐木。基本配置(级别=日志记录.信息,

format='[%(ASC time)-15s][%(关卡名)8s][%(名字)10s]-%(消息)s(%(文件名)s:%(行号)s)',

datefmt='%Y-%m-%d %T '

)

logger=日志记录。获取记录器(_ _ name _ _)

def proxy_to_dict(proxy):

d={

' schema': proxy.schema,

' ip': proxy.ip,

' port': proxy.port,

' used_total': proxy.used_total,

' success _ times ':代理。成功次数

' continuous _ failed ':代理。连续_失败

' created _ time ':代理。创建时间

}

返回d

def proxy_from_dict(d):

返回IP代理(schema=d[' schema '],ip=d['ip'],port=d['port'],used_total=d['used_total'],

success _ times=d[' success _ times '],continuous _ failed=d[' continuous _ failed '],

created_time=d['created_time'])

#截断标题和尾部空白

定义条带(数据):

如果数据不为无:

返回data.strip()

返回数据

base_headers={

Accept-Encoding': 'gzip,deflate,br ',

'接受-语言:“恩——我们,恩;q=0.9,zh-CN;q=0.8,zh;q=0.7 '

}

def request_page(url,options={}编码='utf-8 '):

'''发送请求,获得响应' ' '

headers=dict(base_headers,* *选项)

如果"用户代理"不在headers.keys()中:

标题['用户代理]=random.choice用户代理列表)

logger.info('正在抓取:' url)

尝试:

response=requests.get(url,headers=headers)

如果响应。状态_代码==200:

logger.info('抓取成功:' url)

返回回应。内容。解码(编码=编码)

除了连接错误:

logger.error('抓取失败网址)

不返回

def _is_proxy_available(proxy,options={}):

'''检查代理是否可用' ' '

headers=dict(base_headers,* *选项)

如果"用户代理"不在headers.keys()中:

标题['用户代理]=random.choice用户代理列表)

proxy={ proxy。架构:代理服务器._get_url()}

CHECK _ URLS=PROXY _ CHECK _ URLS[PROXY。模式]

对于检查网址中的网址:

尝试:

response=requests.get(url=url,proxies=proxies,headers=headers,timeout=5)

除了基本异常:

伐木工。信息(' URL '验证代理代理' get_url()'结果:不可用)

否则:

如果响应。状态_代码==200:

伐木工。信息(' URL '验证代理代理' get_url()'结果:可用)

返回真实的

否则:

伐木工。信息(' URL '验证代理代理' get_url()'结果:不可用)

返回错误的

if __name__=='__main__ ':

headers=dict(base_headers)

如果"用户代理"不在headers.keys()中:

标题['用户代理]=random.choice用户代理列表)

proxy={ ' https ':' https://163。125 .255 .154:9797 ' }

响应=请求。得到(' https://www。百度一下。' com ',头=头,代理=代理,超时=3)

打印(响应。内容)

代理_队列。巴拉圭

代理队列用来保存并对外提供互联网协议(互联网协议)代理,不同的代理队列内代理互联网协议(互联网协议)的保存和提取策略可以不同。在这里,基本队列是所有代理队列的基类,其中声明了所有代理队列都需要实现的保存代理知识产权,提取代理知识产权,查看代理互联网协议(互联网协议)数量等接口。示例的先进先出。比较LIFO队列是一个先进先出队列,底层使用雷迪斯列表实现,为了确保同一个代理互联网协议(互联网协议)只能被放入队列一次,这里使用了一个雷迪斯代理:已存在集合进行入队前重复校验。

# -*-编码:utf-8 -*-

从代理工具导入记录器

导入json

导入雷迪斯

从ipproxy导入IPProxy

从代理工具导入proxy_to_dict,proxy_from_dict,_ is _ proxy _可用

从设置导入代理_ REDIS _已存在,代理_REDIS_FORMATTER,MAX_CONTINUOUS_TIMES,PROXY_CHECK_BEFOREADD

'''

代理队列基类

'''

类别基本队列(对象):

def __init__(自身,服务器):

''初始化代理队列实例

因素

-

服务器:StrictRedis

雷迪斯客户端实例

'''

self.server=服务器

def _serialize_proxy(self,proxy):

'''序列化代理实例' ' '

返回代理到字典(代理)

def _deserialize_proxy(self,serialized_proxy):

'''反序列化代理实例' ' '

返回proxy _ from _ dict(eval(序列化_代理))

def __len__(self,schema='http '):

'''返回队列的长度' ' '

引发NotImplementedError

定义推送(自身、代理、需要检查):

'''推送代理' ' '

引发NotImplementedError

def pop(self,schema='http ',timeout=0):

'''弹出代理' ' '

引发NotImplementedError

类FIFO队列(BaseQueue):

'''先进先出队列' ' '

def __len__(self,schema='http '):

'''返回队列的长度' ' '

返回自我。服务器。llen(proxy _ REDIS _ formatter。格式(架构))

def push(self,proxy,need _ CHECK=PROXY _ CHECK _ before add):

'''推送代理' ' '

如果需要检查并且代理不可用(代理):

返回

elif代理。连续失败最大连续次数而非自我._是_已存在(代理):

key=proxy _ REDIS _ formatter。格式(代理。模式)

self.server.rpush(key,json.dumps(self ._序列化_代理(proxy),确保_ascii=False))

def pop(self,schema='http ',timeout=0):

'''弹出代理' ' '

如果超时时间为0:

p=自我。服务器。bl pop(proxy _ REDIS _ formatter。格式(架构。lower()),超时)

if isinstance(p,tuple):

p=p[1]

否则:

p=自我。服务器。lpop(proxy _ REDIS _ formatter。格式(架构。下()))

如果p:

p=自我。_反序列化_代理(p)

自我。服务器。srem(proxy _ REDIS _ EXISTED,p._get_url())

返回p

定义已存在(自身,代理):

添加=自我。服务器。sadd(proxy _ REDIS _ EXISTED,proxy ._get_url())

添加的返回==0

if __name__=='__main__ ':

r=redis .StrictRedis(host='localhost ',port=6379)

queue=FifoQueue(r)

proxy=IPProxy('http ',' 218.66.253.144 ',' 80 ')

队列。推送(代理)

proxy=queue.pop(schema='http ')

打印(代理. get_url())

代理_爬虫。巴拉圭

ProxyBaseCrawler是所有代理爬虫的基类,其中只定义了一个_start_crawl()方法用来从搜集到的代理网站爬取代理知识产权。

# -*-编码:utf-8 -*-

从lxml导入etree

从ipproxy导入IPProxy

从代理工具导入条,请求页,记录器

类ProxyBaseCrawler(对象):

def __init__(self,queue=None,website=None,URL=[]):

self.queue=queue

self.website=网站

self.urls=urls

定义_开始_爬行(自身):

引发NotImplementedError

类KuaiDailiCrawler(proxybase爬虫):#快代理

定义_开始_爬行(自身):

对于self.urls中的url_dict:

logger.info('开始爬取【‘自我。网站']:[' URL _ dict[' type ']']')

has_more=True

url=无

在…期间拥有_更多:

如果url_dict.keys()和str.find(url_dict['url'],' {} ')中的页面!=-1:

url=url_dict['url'].格式(字符串(url_dict['page']))

URL _ dict[' page ']=URL _ dict[' page ']1

否则:

url=url_dict['url']

有_更多=假

html=etree .HTML(请求页面(网址))

tr _ list=html。XPath('//table[@ class=' table table-bordered table-striped ']/tbody/tr ')

对于tr_list中的tr:

ip=tr.xpath(' ./TD[@ data-title=' IP ']/text()')[0]if len(

tr.xpath(' ./td[@data-title='IP']/text()'))其他无

port=tr.xpath(' ./TD[@ data-title=' PORT ']/text()')[0]if len(

tr.xpath(' ./td[@data-title='PORT']/text()'))否则无

schema=tr.xpath(' ./td[@data-title='类型]/text()')[0] if len(

tr.xpath(' ./td[@data-title='类型]/text()'))否则无

proxy=IP proxy(schema=strip(schema),ip=strip(ip),port=strip(port))

如果代理. check_format():

self.queue.push(代理)

如果tr_list为无:

有_更多=假

类FeiyiDailiCrawler(proxybase爬虫):#飞蚁代理

定义_开始_爬行(自身):

对于self.urls中的url_dict:

logger.info('开始爬取【‘自我。网站']:[' URL _ dict[' type ']']')

has_more=True

url=无

在…期间拥有_更多:

如果url_dict.keys()和str.find(url_dict['url'],' {} ')中的页面!=-1:

url=url_dict['url'].格式(字符串(url_dict['page']))

URL _ dict[' page ']=URL _ dict[' page ']1

否则:

url=url_dict['url']

有_更多=假

html=etree .HTML(请求页面(网址))

tr _ list=html。XPath('//div[@ id=' main-content ']//table/tr[position()1]')

对于tr_list中的tr:

ip=tr.xpath(' ./TD[1]/text()')[0]if len(tr。XPath(./td[1]/text()'))否则无

port=tr.xpath(' ./TD[2]/text()')[0]if len(tr。XPath(./td[2]/text()'))否则无

schema=tr.xpath(' ./TD[4]/text()')[0]if len(tr。XPath(./td[4]/text()'))否则无

proxy=IP proxy(schema=strip(schema),ip=strip(ip),port=strip(port))

如果代理. check_format():

self.queue.push(代理)

如果tr_list为无:

有_更多=假

类WuyouDailiCrawler(proxybase爬虫):#无忧代理

定义_开始_爬行(自身):

对于self.urls中的url_dict:

logger.info('开始爬取【‘自我。网站']:[' URL _ dict[' type ']']')

has_more=True

url=无

在…期间拥有_更多:

如果url_dict.keys()和str.find(url_dict['url'],' {} ')中的页面!=-1:

url=url_dict['url'].格式(字符串(url_dict['page']))

URL _ dict[' page ']=URL _ dict[' page ']1

否则:

url=url_dict['url']

有_更多=假

html=etree .HTML(请求页面(网址))

ul _ list=html。XPath('//div[@ class=' wlist '][2]//ul[@ class=' L2 ']')

对于ul_list中的ul:

ip=ul.xpath(' ./span[1]/Li/text()')[0]if len(ul。XPath(./span[1]/li/text()'))其他无

port=ul.xpath(' ./span[2]/Li/text()')[0]if len(ul。XPath(./span[2]/li/text()'))其他无

schema=ul.xpath(' ./span[4]/Li/text()')[0]if len(ul。XPath(./span[4]/li/text()'))其他无

proxy=IP proxy(schema=strip(schema),ip=strip(ip),port=strip(port))

如果代理. check_format():

self.queue.push(代理)

如果ul_list为无:

有_更多=假

IPhaiDailiCrawler类(ProxyBaseCrawler): # IP海代理

定义_开始_爬行(自身):

对于self.urls中的url_dict:

logger.info('开始爬取【‘自我。网站']:[' URL _ dict[' type ']']')

has_more=True

url=无

在…期间拥有_更多:

如果url_dict.keys()和str.find(url_dict['url'],' {} ')中的页面!=-1:

url=url_dict['url'].格式(字符串(url_dict['page']))

URL _ dict[' page ']=URL _ dict[' page ']1

否则:

url=url_dict['url']

有_更多=假

html=etree .HTML(请求页面(网址))

tr _ list=html。XPath('//table//tr[position()1]')

对于tr_list中的tr:

ip=tr.xpath(' ./TD[1]/text()')[0]if len(tr。XPath(./td[1]/text()'))否则无

port=tr.xpath(' ./TD[2]/text()')[0]if len(tr。XPath(./td[2]/text()'))否则无

schema=tr.xpath(' ./TD[4]/text()')[0]if len(tr。XPath(./td[4]/text()'))否则无

proxy=IP proxy(schema=strip(schema),ip=strip(ip),port=strip(port))

如果代理. check_format():

self.queue.push(代理)

如果tr_list为无:

有_更多=假

类YunDailiCrawler(proxybase爬虫):#云代理

定义_开始_爬行(自身):

对于self.urls中的url_dict:

logger.info('开始爬取【‘自我。网站']:[' URL _ dict[' type ']']')

has_more=True

url=无

在…期间拥有_更多:

如果url_dict.keys()和str.find(url_dict['url'],' {} ')中的页面!=-1:

url=url_dict['url'].格式(字符串(url_dict['page']))

URL _ dict[' page ']=URL _ dict[' page ']1

否则:

url=url_dict['url']

有_更多=假

html=etree .HTML(请求页面(网址,编码='gbk '))

tr _ list=html。XPath('//table/tbody/tr ')

对于tr_list中的tr:

ip=tr.xpath(' ./TD[1]/text()')[0]if len(tr。XPath(./td[1]/text()'))否则无

port=tr.xpath(' ./TD[2]/text()')[0]if len(tr。XPath(./td[2]/text()'))否则无

schema=tr.xpath(' ./TD[4]/text()')[0]if len(tr。XPath(./td[4]/text()'))否则无

proxy=IP proxy(schema=strip(schema),ip=strip(ip),port=strip(port))

如果代理. check_format():

self.queue.push(代理)

如果tr_list为无:

有_更多=假

西戴利克劳勒类(ProxyBaseCrawler): #西刺代理

定义_开始_爬行(自身):

对于self.urls中的url_dict:

logger.info('开始爬取【‘自我。网站']:[' URL _ dict[' type ']']')

has_more=True

url=无

在…期间拥有_更多:

如果url_dict.keys()和str.find(url_dict['url'],' {} ')中的页面!=-1:

url=url_dict['url'].格式(字符串(url_dict['page']))

URL _ dict[' page ']=URL _ dict[' page ']1

否则:

url=url_dict['url']

有_更多=假

html=etree .HTML(请求页面(网址))

tr _ list=html。XPath('//table[@ id=' IP _ list ']//tr[@ class!='副标题']')

对于tr_list中的tr:

ip=tr.xpath(' ./TD[2]/text()')[0]if len(tr。XPath(./td[2]/text()'))否则无

port=tr.xpath(' ./TD[3]/text()')[0]if len(tr。XPath(./td[3]/text()'))否则无

schema=tr.xpath(' ./TD[6]/text()')[0]if len(tr。XPath(./td[6]/text()'))其他无

if schema.lower()=='http '或schema.lower()=='https ':

proxy=IP proxy(schema=strip(schema),ip=strip(ip),port=strip(port))

如果代理. check_format():

self.queue.push(代理)

如果tr_list为无:

有_更多=假

run.py

通过run.py启动各个代理网站爬虫。

# -*-编码:utf-8 -*-

导入雷迪斯

从代理队列导入从队列

从设置导入雷迪斯主机、REDIS端口

从代理爬虫导入WuyouDailiCrawler,FeiyiDailiCrawler,KuaiDailiCrawler,IPhaiDailiCrawler,YunDailiCrawler,\

西戴利克劳勒

r=redis .StrictRedis(host=REDIS_HOST,port=REDIS_PORT)

fifo_queue=FifoQueue(r)

def run_kuai():

kuaidailiCrawler=kuaidailiCrawler(queue=FIFO _ queue,website='快代理[国内高匿]',

URLs=[{ ' URL ':' https://www。快戴笠。' com/free/inha/{ }/',' type ':'国内高匿,

' page': 1},

{ ' URL ':' https://www。快戴笠。' com/free/intr/{ }/',' type ':'国内普通,

第页':1}])

kuaidailiCrawler ._start_crawl()

费祎():

feiyidailiCrawler=feiyidailiCrawler(queue=FIFO _ queue,website='飞蚁代理,

URLs=[{ ' URL ': ' http://www。飞一proxy。com/?page _ id=1457 ','键入':'首页推荐'}])

飞翼戴立克劳勒._start_crawl()

吴优():

wuyoudailiCrawler=wuyoudailiCrawler(queue=FIFO _ queue,website='无忧代理,

URLs=[{ ' URL ': ' http://www。数据5u。com/free/index。html ','键入':'首页推荐'},

{'url': 'http://www.data5u.com

/free/gngn/index.shtml', 'type': '国内高匿'}, {'url': 'http://www.data5u.com/free/gnpt/index.shtml', 'type': '国内普通'}]) wuyoudailiCrawler._start_crawl() def run_iphai(): crawler = IPhaiDailiCrawler(queue=fifo_queue, website='IP海代理', urls=[{'url': 'http://www.iphai.com/free/ng', 'type': '国内高匿'}, {'url': 'http://www.iphai.com/free/np', 'type': '国内普通'}, {'url': 'http://www.iphai.com/free/wg', 'type': '国外高匿'}, {'url': 'http://www.iphai.com/free/wp', 'type': '国外普通'}]) crawler._start_crawl() def run_yun(): crawler = YunDailiCrawler(queue=fifo_queue, website='云代理', urls=[{'url': 'http://www.ip3366.net/free/?stype=1&page={}', 'type': '国内高匿', 'page': 1}, {'url': 'http://www.ip3366.net/free/?stype=2&page={}', 'type': '国内普通', 'page': 1}, {'url': 'http://www.ip3366.net/free/?stype=3&page={}', 'type': '国外高匿', 'page': 1}, {'url': 'http://www.ip3366.net/free/?stype=4&page={}', 'type': '国外普通', 'page': 1}]) crawler._start_crawl() def run_xici(): crawler = XiCiDailiCrawler(queue=fifo_queue, website='西刺代理', urls=[{'url': 'https://www.xicidaili.com/', 'type': '首页推荐'}, {'url': 'https://www.xicidaili.com/nn/{}', 'type': '国内高匿', 'page': 1}, {'url': 'https://www.xicidaili.com/nt/{}', 'type': '国内普通', 'page': 1}, {'url': 'https://www.xicidaili.com/wn/{}', 'type': '国外高匿', 'page': 1}, {'url': 'https://www.xicidaili.com/wt/{}', 'type': '国外普通', 'page': 1}]) crawler._start_crawl() if __name__ == '__main__': run_xici() run_iphai() run_kuai() run_feiyi() run_yun() run_wuyou()

爬取西刺代理时,后台日志示例如下:

Redis数据库中爬取到的代理IP的数据结构如下:

四、代理测试

?接下来,使用爬取好的代理来请求 http://icanhazip.com?进行测试,代码如下:

# -*- coding: utf-8 -*- import random import requests from proxy_util import logger from run import fifo_queue from settings import USER_AGENT_LIST from proxy_util import base_headers # 测试地址 url = 'http://icanhazip.com' # 获取代理 proxy = fifo_queue.pop(schema='http') proxies = {proxy.schema:proxy._get_url()} # 构造请求头 headers = dict(base_headers) if 'User-Agent' not in headers.keys(): headers['User-Agent'] = random.choice(USER_AGENT_LIST) response = None successed = False try: response = requests.get(url,headers=headers,proxies = proxies,timeout=5) except BaseException: logger.error("使用代理< "+proxy._get_url()+" > 请求 < "+url+" > 结果: 失败 ") else: if (response.status_code == 200): logger.info(response.content.decode()) successed = True logger.info("使用代理< " + proxy._get_url() + " > 请求 < " + url + " > 结果: 成功 ") else: logger.info(response.content.decode()) logger.info("使用代理< " + proxy._get_url() + " > 请求 < " + url + " > 结果: 失败 ") # 根据请求的响应结果更新代理 proxy._update(successed) # 将代理返还给队列,返还时不校验可用性 fifo_queue.push(proxy,need_check=False)

使用?http://218.66.253.144:80?代理请求成功后将代理重新放回队列,并将 Redis 中该代理的?used_total 、success_times 、continuous_failed三个字段信息进行了相应的更新。

项目地址:https://github.com/pengjunlee/ipproxy_pool.git

到此这篇关于Python爬虫代理池搭建的方法步骤的文章就介绍到这了,更多相关Python爬虫代理池搭建内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们!

郑重声明:本文由网友发布,不代表盛行IT的观点,版权归原作者所有,仅为传播更多信息之目的,如有侵权请联系,我们将第一时间修改或删除,多谢。

留言与评论(共有 条评论)
   
验证码: