欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

进击的爬虫:用Python搭建匿名代理池

程序员文章站 2022-03-13 16:37:29
常听到很多人抱怨自己的IP因爬虫次数太多而被网站屏蔽,不得不频繁使用各种代理IP,却又因为网上的公开代理大部分都是不能使用,而又要花钱花精力去申请VIP代理,几番波折又遭屏蔽。特此写一篇如何利用Python搭建代理池的文章,以降低时间及精力成本,实现自动化获取活跃代理IP的功能。 运作原理 一、 网 ......

       常听到很多人抱怨自己的ip因爬虫次数太多而被网站屏蔽,不得不频繁使用各种代理ip,却又因为网上的公开代理大部分都是不能使用,而又要花钱花精力去申请vip代理,几番波折又遭屏蔽。特此写一篇如何利用python搭建代理池的文章,以降低时间及精力成本,实现自动化获取活跃代理ip的功能。

 

运作原理

一、 网站代理获取

1.    爬免费代理网站的ip列表测试是否可用及是否是高匿

2.    若都是,则放进数据库,否则丢弃。

3.    重复第2步

 

二、 保证失效的代理能被尽快从代理池中挑出

1.    从爬虫数据库获取ip

2.    测试ip的可用性和匿名性

3.    如果可用且匿名,则保留,否则丢弃。

4.    重复第1步

 

说明①:可建立一个爬虫程序守护程序(daemon),有此方面需要的小伙伴可自行谷歌,在此不多做介绍。

说明②:可建立一个对外代理信息接口,无论你用nodejs或者flask/django或者php来写都没关系,在此也不多做介绍。

 

实现

       建议库: requests, beautifulsoup, re, sqlite3。

       其中,用requests库获取代理网站页面,用beautifulsoup和re两库来进行代理信息获取,用sqlite3来对这些信息进行存取。

       如果必要(如代理网站有反爬虫策略时),可用phantomjs替代requests,或用相应库进行数据清理(如base64解码)。

 

下面简单展示一下各部分的代码:

       首先是选择多个能爬取代理且不容易被屏蔽ip的网站,此处以proxy-list.org为例:

 

base_url = "https://proxy-list.org/english/index.php?p="

 

#ip地址及端口的正则

re_pattern_ip = re.compile("(.*):")

re_pattern_port = re.compile(":(.*)")

 

#网站有11页,所以循环11次获取所有代理ip及端口

for startingurl_param in range(1,11):

   html_proxypage = requests.get(base_url+str(startingurl_param)).content

   soup = bs(html_proxypage,"html.parser")

   for raw_proxyinfo in soup.find_all("ul",{"class":none}):

       #此网站有用base64简单对代理进行了加密,所以这里对其解码

       ip_port = base64.b64decode(raw_proxyinfo.find("li",{"class":"proxy"}).text.replace("proxy('","").replace("')",""))

       #接下来利用正则从网页数据中提取我们需要的信息

       ip = re.findall(re_pattern_ip, ip_port)[0]

       port = re.findall(re_pattern_port, ip_port)[0]

       type = raw_proxyinfo.find("li",{"class":"https"}).text

  

 

       接下来是一段简易代理池框架类的代码,提供代理数据库的添加、删除、可连接性检测、匿名性检测:

 

class proxypool:        

   #初始化爬虫池数据库

   def __init__(self,proxypooldb):

       self.proxypooldb = proxypooldb

       self.conn = sqlite3.connect(self.proxypooldb, isolation_level=none)

       self.cursor = self.conn.cursor()

       self.tb_proxypool = "tb_proxypool"

       self.cursor.execute("create table if not exists "+self.tb_proxypool+"(ip text unique, port integer, protocol text)")

 

   #添加代理ip进代理池的接口

   def addproxy(self, ip, port, protocol):  

       self.cursor.execute("insert or ignore into " + self.tb_proxypool+"(ip, port, protocol) values (?,?,?)", [ip,port,protocol])

 

   #检查代理的匿名性及可连接性

   def testconnection(self, ip, port, protocol):

       proxies = { protocol: ip+":"+port }

       try:

           origionalip = requests.get("http://icanhazip.com",timeout=req_timeout).content

           maskedip = requests.get("http://icanhazip.com", timeout=req_timeout,proxies=proxies).content

           if origionalip != maskedip:

               return true

           else:

               return false

       except:    

           return false

 

   #删除代理ip对应的数据库记录

   def delrecord(self, ip):

       self.cursor.execute("delete from "+self.tb_proxypool+" where ip=?",(ip,))

  

       下面是对代理池进行去“失效ip”的代码:    

 

   #循环代理池,逐行测试ip地址端口协议是否可用

def cleannonworking(self):

   for info in self.cursor.execute("select * from "+self.tb_proxypool).fetchall():

       ip = info[0]

       port = str(info[1])

       protocol = info[2].lower()

 

       isanonymous = self.testconnection(ip,port,protocol)

       if isanonymous == false:

           #这条代理的可用性失效了,从数据库里删除

           self.delrecord(ip)

 

#通过检测icanhazip.com回显来检测可用性及匿名性

def testconnection(self, ip, port, protocol):

       proxies = { protocol: ip+":"+port }

       try:

           origionalip = requests.get("http://icanhazip.com",timeout=req_timeout).content

           maskedip = requests.get("http://icanhazip.com", timeout=req_timeout,proxies=proxies).content

           if origionalip != maskedip:

               return true

           else:

               return false

       except:    

           return false

  

反思

       这个项目是我当年用python练手写的,以现在的程度再来回顾,逻辑不够严谨,各类功能太过耦合,不少段落需要重写,因为代码是在校园网内所跑,所以还需要考虑到网络连接的稳定性,这就造成部分代码之间的混乱关系。

       通过icanhazip.com来检测代理匿名性的方法或许有效,但却忽略了x-forwarded-for的http头,所以有很大风险,必须改进。

       验证代理池内代理的有效性,需要多线程,目前的方案效率太低。

 

完整代码

       放在此文章中的是代理池的核心代码,旨在提供各位读者能够自己实现的思路及参考,ubuntu 16.04及kali下用python 2.7测试可运行。