怎樣簡單的搭建一個免費的IP代理池
之前寫過一篇python實戰專案二:獲取IP代理的文章,不過說實話,這個程式有幾點不足,以至於不能愉快玩耍之後,我就重新整理了思路,又寫了一個關於獲取免費IP代理的程式碼。在這兒我想寫反思一下之前這個程式碼的主要不足:
第一點,由於資料很雜,所以在提取資訊時頻繁的使用了迴圈,但是迴圈使用的太頻繁會使得程式執行的速度效率降低,而字典的索引效率就高效的多,所以這一次不必使用迴圈的地方就不使用迴圈,改用字典。
第二點,爬取下來的IP不是都能用的,所以得檢驗一下。我之前用的方法就是直接將得到的IP訪問某個網站,若狀態碼status_code為200,則說明可用。但是實際上得到的IP不能用的還是很多。所以這次我增加了一些檢驗的條件。
這次我還是選擇西刺代理。接下來就開始說一下過程:
首先先匯入相關的模組
import requests
from lxml import etree
import re
import time
然後定義函式,爬取網頁資訊
def get_html(url): # 獲取西刺代理高匿代理的每一頁的html,return: 返回html headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 ' '(KHTML, like Gecko) Chrome/68.0.3440.106 Safari/537.36'} try: response = requests.get(url, headers=headers) if response.status_code == 200: return response.text else: print('IP不能用') except Exception as e: print(e)
得到響應後,就是要提取頁面的資訊了,先用xpath表示式得到資訊,包括IP地址、是否匿名、型別、連線時間、存活時間等,提取之後的內容用字典來盛裝,然後返回該字典
def get_data(html): # 獲取每一頁的IP資訊, return:返回這一頁IP的資訊(封裝成字典形式) info = etree.HTML(html) try: address = info.xpath('//tr[@class="odd" or class=""]/td[2]/text()') # IP地址 ports = info.xpath('//tr[@class="odd" or class=""]/td[3]/text()') # 埠 anonymous = info.xpath('//tr[@class="odd" or class=""]/td[5]/text()') # 匿名形式 http_https = info.xpath('//tr[@class="odd" or class=""]/td[6]/text()') # http or https speed = info.xpath('//tr[@class="odd" or class=""]/td[7]/div[1]/@title') # 連線速度 speed_width = info.xpath('//tr[@class="odd" or class=""]/td[7]/div[1]/div/@style') # 連線速度的比例 conn_time = info.xpath('//tr[@class="odd" or class=""]/td[8]/div[1]/@title') # 連線時間 conn_time_width = info.xpath('//tr[@class="odd" or class=""]/td[8]/div[1]/div/@style') # 連線時間的比例 life = info.xpath('//tr[@class="odd" or class=""]/td[9]/text()') # 存活時間 test = info.xpath('//tr[@class="odd" or class=""]/td[10]/text()') # 檢驗時間 data_info = {} # print(len(address)) for i in range(0, len(address)): data_info[str(i+1)] = {'IP地址': address[i]+':'+ports[i], '是否匿名': anonymous[i], '型別': http_https[i], '速度': eval((re.compile('(.*?)秒').findall(speed[i]))[0]), '速度比例': eval((re.compile('width:(.*?)%').findall(speed_width[i]))[0]), '連線時間': eval((re.compile('(.*?)秒').findall(conn_time[i]))[0]), '耗時比例': eval((re.compile('width:(.*?)%').findall(conn_time_width[i]))[0]), '存活時間': eval((re.compile('(\d+).*?').findall(life[i]))[0]), '驗證時間': test[i]} return data_info # 返回名為data_info的字典,字典的每個鍵值對就是一個IP的資訊{{'1':{}},{'2':{}}} except Exception as er: print(er)
接下來就將上面得到的字典資訊存入檔案,存入之前要先進行第一次的檢驗,就是篩選出存活時間>100天,還有速度等限制條件的,這樣至少可以保證得到的IP生命力是比一般的那些頑強。
def save_to_file(data):
# 將得到的符合要求的IP寫入檔案;param data: 所有的IP集合(字典形式的)
try:
# print(data['1'])
# print(type(data['2']['耗時比例']), type(data['2']['存活時間']))
with open('IP代理池.csv', 'w', encoding='utf-8') as f:
for i in range(0, len(data)):
if data[str(i+1)]['速度比例'] >= 70 and data[str(i+1)]['耗時比例'] >= 75 and \
data[str(i+1)]['存活時間'] >= 100:
# 存活時間要>=100天,速度和耗時也有相關的限制
f.write(str(data[str(i+1)]))
f.write('\n')
else:
pass
except Exception as er:
print(er)
然後就是主函式,在這個函式中實現呼叫其他函式,實現整個程式的功能,因為只是用來檢驗能否用該方法得到有效的IP,所以我的迴圈range(1, 2)只爬取了一頁的內容。
def main():
# 主邏輯函式,實現得到IP的功能;param page: 預設為爬取1頁,其他函式呼叫時刻自行改變該值;return: None
base_url = 'http://www.xicidaili.com/nn/'
for i in range(1, 2):
url = base_url + str(i)
print('爬取第{}頁'.format(str(i)).center(20, '='))
html = get_html(url) # 得到當前爬取頁面的html資訊
data = get_data(html) # 解析當前頁面,得到想要獲取的資訊
save_to_file(data) # 將得到的資訊寫入檔案儲存
if i % 2 == 0: # 每爬取兩頁,停3秒
time.sleep(3)
到這兒整個程式執行之後就可以得到有效的IP了,部分結果如下圖
但是這隻得到相對存活率高一點的IP,有些IP可能已經被一些網站加入黑名單了。那麼我得到這個IP代理池之後,我怎麼使用呢?就是檢驗的第二重,用帶爬取的網站去檢驗,檢驗剛剛得到的檔案中的那些有效IP,用這些IP去訪問待爬取的網頁,若返回狀態碼是200,則說明還能用,不能用的直接移除就行。
兩種方法,一種是你在執行你的其他爬蟲的時候,先將該程式執行一遍,得到有效IP,然後在那個爬蟲裡讀取這個程式結束後的裝有效代理IP的檔案就行;另一種方法就是在這個程式中寫一個函式,作為那個函式呼叫的介面,在那個程式中匯入這個程式,然後執行的結果就是會先執行這個程式,函式返回的是一個有效IP的列表,在那個爬蟲裡直接使用random模組的choice方法就可以隨機得到一個該列表的一個IP,每次執行得到IP的都是一個隨機的,也會更好的防止被封。兩種方法其實差不多,只不過第二種方法會將該爬蟲獲取有效IP代理的時間也會在那個程式中消耗,所以這個就看個人選擇。我寫的介面是
def get_ip():
# 介面函式,其他爬蟲可以直接匯入這個程式(模組)之後呼叫該方法就行,返回的就是有效的IP列表
test_url = "https://blog.csdn.net/"
headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36\
(KHTML, like Gecko) hrome/68.0.3440.106 Safari/537.36'}
ip_data = []
try:
with open('IP代理池.csv', 'r', encoding='utf-8') as f:
for item in f.readlines():
ip_data.append(eval(item)['IP地址']) # 讀取檔案並將IP全部提取出來檢驗
except Exception as e:
print(e)
# print(ip_data)
for data in ip_data:
proxy = {'http': 'http://'+data}
try:
r = requests.get(test_url, headers=headers, proxies=proxy)
if r.status_code == 200:
pass # 若此IP有效,則萬事大吉
# print('{}可用'.format(proxy).center(20, '='))
else:
ip_data.remove(data) # 若不可用,則移除
# print('{}不可用'.format(proxy).center(20, '%'))
except Exception as e:
print(e)
# print('得到{}個有用個IP'.format(len(ip_data))
return ip_data