創建IP代理

在爬取豆瓣電影時，由於使用了多線程，致使IP被封，因此使用代理來爬取就頗有必要了，本文主要是從西刺代理中爬取能夠的IP，主要有如下幾個功能：

一、獲取西祠代理第一頁高匿IP

二、使用代理IP訪問網易網判斷IP是否可能（使用多線程有效提高了驗證速度）

三、能夠在其餘的爬蟲中調用，如

1 from IPProxy import getProxy
2 ip_list = getProxy.ip_list_able

ip_list 即爲獲取的可用IP

源文件已放Github(https://github.com/createnewli/IPProxy)，具體代碼以下，歡迎交流討論

  1 # -*- coding: utf-8 -*-
  2 # @Date    : 2017-3-3 16:28:15
  3 # @Author  : youth-lee (you@example.org)
  4 # @Link    : http://www.cnblogs.com/youth-lee/
  5 # @Version : Ver 0.1
  6 
  7 '''
  8 一、從西刺代理往上獲取第一頁國內高匿的IP
  9 二、使用多線成檢驗獲取的IP是否可在訪問網易
 10 三、可調用
 11     from IPProxy import getProxy
 12     ip_list = getProxy.ip_list_able
 13 '''
 14 
 15 import requests
 16 import datetime
 17 from bs4 import BeautifulSoup
 18 from threading import Thread
 19 
 20 
 21 class getProxy():
 22     def __init__(self):
 23         self.user_agent = "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/42.0.2311.154 Safari/537.36 LBBROWSER"
 24         self.header = {"User-Agent": self.user_agent}
 25         self.start_url = "http://www.xicidaili.com/nn/"
 26         self.ip_list = []
 27         self.ip_list_able = []
 28 
 29         self.get_ip()
 30         self.check_ip_list()
 31 
 32         print("獲取了%s個可用代理" % len(self.ip_list_able))
 33 
 34     def getContent(self, url):
 35         # 獲取網頁內容並返回
 36         content = requests.get(url, headers=self.header)
 37         soup = BeautifulSoup(content.text, "lxml")
 38         return soup
 39 
 40     def get_ip(self):
 41         soup = self.getContent(self.start_url)
 42         tr_soup = soup.find("table", id="ip_list").find_all("tr")
 43         # 把第一個tr標籤剔除，第一個tr爲標題行
 44         tr_soup.pop(0)
 45         for tr in tr_soup:
 46             items = tr.find_all("td")
 47 
 48             if items is not []:
 49                 ip = items[1].get_text().strip()
 50                 port = items[2].get_text().strip()
 51                 ip_port = ip + ":" + port
 52                 self.ip_list.append(ip_port)
 53 
 54     def check_ip_list(self):
 55         # 使用多線程檢測ip是否可用
 56         threads = []
 57         for ip_port in self.ip_list:
 58             t = Thread(target=self.check_ip, args=[ip_port])
 59             t.start()
 60             threads.append(t)
 61 
 62         for t in threads:
 63             t.join()
 64 
 65     def check_ip(self, ip_port):
 66         # 若是能正常訪問則將ip放入到ip_list_able中
 67         if self.is_Alive(ip_port):
 68             self.ip_list_able.append(ip_port)
 69 
 70     def is_Alive(self, ip_port):
 71         # 使用代理IP去訪問網易，能成功則能使用
 72         proxy = {"http": ip_port}
 73         test_url = "http://www.163.com"
 74 
 75         try:
 76             test_content = requests.get(
 77                 test_url, headers=self.header, proxies=proxy, timeout=1)
 78             if test_content.status_code == 200:
 79                 # print("%s知足使用條件！" % (ip_port))
 80                 return True
 81             else:
 82                 # print("%s不知足使用條件！" % (ip_port))
 83                 return False
 84         except:
 85             # print("%s不知足使用條件！" % (ip_port))
 86             return False
 87 
 88 
 89 if __name__ == "__main__":
 90 
 91     time1 = datetime.datetime.now()
 92 
 93     obj = getProxy()
 94 
 95     print("得到%s個代理IP！" % len(obj.ip_list_able))
 96 
 97     for ip in obj.ip_list_able:
 98         print(ip)
 99 
100     time2 = datetime.datetime.now()
101     print("耗時：%s" % str(time2 - time1))

View Code