0x01 寫在前面
常聽到很多人抱怨自己的IP因爬蟲次數太多而被網站屏蔽，不得不頻繁使用各種代理IP，卻又因為網上的公開代理大部分都是不能使用，而又要花錢花精力去申請VIP代理，幾番波折又遭屏蔽。特此寫一篇如何利用Python搭建代理池的文章，以降低時間及精力成本，實現自動化獲取活躍代理IP的功能。
說明①：可建立一個爬蟲程序守護程序（Daemon），有此方面需要的小夥伴可自行谷歌，在此不多做介紹。
說明②：可建立一個對外代理信息接口，無論你用NodeJS或者Flask/Django或者PHP來寫都沒關係，我會在未來的更新中加上這個功能，在這篇文章中也不多做介紹。
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
0x05 完整代碼
放在此文章中的是代理池的核心代碼，旨在提供各位讀者能夠自己實現的思路及參考。完整代碼可在我的Github主頁中找到(ProxyPool)，Win7 64位、Ubuntu 16.04及Kali下用Python 2.7測試可運行。
分享到:
閱讀更多 繁華落盡and曲終人散 的文章
