如何利用Python搭建代理池？

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

　　01 运作原理

　　一、 网站代理获取

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

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

　　3. 重复第2步

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

　　1. 从爬虫数据库获取IP

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

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

　　4. 重复第1步

　　说明①： 可建立一个爬虫程序守护程序（Daemon），有此方面需要的小伙伴可自行谷歌，在此不多做介绍。

　　说明②： 可建立一个对外代理信息接口，无论你用NodeJS或者Flask/Django或者PHP来写都没关系，在此也不多做介绍。

　　02 实现

　　建议库： requests, BeautifulSoup, re, sqlite3。

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

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

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

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

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

　　2.#IP地址及端口的正则

　　3.Re_Pattern_IP = re.compile("(.*):")

　　4.Re_Pattern_PORT = re.compile("

.*)")

　　5.#网站有11页，所以循环11次获取所有代理IP及端口

　　6.for startingURL_Param in range(1,11):

　　7.HTML_ProxyPage = requests.get(BASE_URL+str(startingURL_Param)).content

　　8.soup = bs(HTML_ProxyPage,"html.parser")

　　9.for Raw_ProxyInfo in soup.find_all("ul",{"class":None}):

　　10.#此网站有用Base64简单对代理进行了加密，所以这里对其解码

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

roxy('","").replace("')",""))

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

　　13.IP = re.findall(Re_Pattern_IP, ip_port)[0]

　　14.PORT = re.findall(Re_Pattern_PORT, ip_port)[0]

　　15.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

　　04 反思

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

　　通过icanhazip.com来检测代理匿名性的方法或许有效，但却忽略了X-Forwarded-For 的HTTP头，所以有很大风险，必须改进。

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

　　05 完整代码

　　放在此文章中的是代理池的核心代码，旨在提供各位读者能够自己实现的思路及参考。完整代码可在作者的Github主页中找到，Ubuntu
16.04及Kali下用Python
2.7测试可运行。

来源：中国统计网