Python3.x：免費代理ip的批量獲取並入庫
一、簡介
網絡爬蟲的世界，向來都是一場精彩的攻防戰。現在許多網站的反爬蟲機制在不斷的完善，其中最令人頭疼的，莫過於直接封鎖你的ip。但是道高一尺魔高一丈，在爬取網頁的時候，使用上代理ip，便可以有效的避免自己的ip被封鎖。
想要使用代理ip，目前你可以去相應的代理網站購買代理ip（如果是大型的項目還是推薦去購買），也可以去使用一些代理網站提供的免費的代理ip，不過這些ip還是存在很多問題的，有些不可用，有些不穩定，有些時效短。不過如果量大的話，還是有不少可以使用的。
基於這個目的，利用Python的requests庫寫了一個簡單的批量獲取免費代理ip的程序，其中包括“下載+驗證”程序。下面將簡單介紹代碼思路和使用方法。
二、Python實現思路
1，確定獲取免費代理ip的網頁
目前有些提供免費代理ip網站有以下三類情況：
- 所有的免費代理ip信息在網頁標簽中
- 所有的免費代理ip信息在網頁標簽中，不過使用了一些隱藏標簽
- 所有的免費代理ip信息在圖片中
本文選擇第一類網頁進行提取（這一類的網站的數量也是較多的，基本滿足小規模的使用需求），此選取了以下5個網站：
- 無憂代理 : http://www.data5u.com/
- 快代理 : https://www.kuaidaili.com/
- 小舒代理 : http://www.xsdaili.com/
- 西刺代理 : http://www.xicidaili.com/
- 89免費代理: http://www.89ip.cn/
2，下載代理IP（parse_html.py）
2.1 獲取網頁
解析一個網頁，第一步就是先獲取頁面。因為有多個頁面要獲取，為了方便就編寫一個獲取頁面的函數，便於之后進行調用。函數如下：
# 1. 無憂代理 : http://www.data5u.com/ # 2. 快代理 : https://www.kuaidaili.com/ # 3. 小舒代理 : http://www.xsdaili.com/ # 4. 西刺代理 : http://www.xicidaili.com/ # 5. 89免費代理: http://www.89ip.cn/ from bs4 import BeautifulSoup import requests import re def get_html(url, open_proxy=False, ip_proxies=None): """ 獲取頁面的html文件 :param url: 待獲取頁面的鏈接 :param open_proxy: 是否開啟代理，默認為False :param ip_proxies: 若開啟，代理地址 :return: """ try: pattern = re.compile(r'//(.*?)/') host_url = pattern.findall(url)[0] headers = { "Host": host_url, "User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64; rv:60.0) Gecko/20100101 Firefox/60.0", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2", "Accept-Encoding": "gzip, deflate", "Connection": "keep-alive", } if open_proxy: # 判斷是否開啟代理 proxies = {"http": "http://" + ip_proxies, } # 設置代理，例如{"http": "http://103.109.58.242:8080", } res = requests.get(url, headers=headers, proxies=proxies, timeout=5) else: res = requests.get(url, headers=headers, timeout=5) res.encoding = res.apparent_encoding # 自動確定html編碼 print("Html頁面獲取成功 " + url) return res.text # 只返回頁面的源碼 except Exception as e: print("Html頁面獲取失敗 " + url) print(e)
這個函數設置了三個參數，分別是：待獲取頁面的鏈接、是否開啟代理、代理地址。最終的返回值為網頁的源碼文本。
其中headers部分是偽裝為瀏覽器請求頭，這個函數還設置了利用代理去獲取頁面，也就是說你可以利用已有的代理去獲取這些頁面。
2.2 保存IP
當我們對獲取的每個頁面進行解析時，都會獲取頁面上的代理ip，同樣的，每個頁面獲取的代理ip都要保存下來，將代理ip寫入txt文件。函數如下：
def save_ip(data, save_path): """ 將獲取的ip信息保存到文件中 :param data: 代理ip數據，數據類型為列表 :param save_path: 代理ip保存路徑 :return: """ try: print("總共獲取 " + str(len(data)) + " 條數據") with open(save_path, "a") as f: for i in range(len(data)): f.write(data[i]) f.close() print("文件保存成功") except Exception as e: print("文件保存失敗！！！") print(e)
2.3 頁面分析
由於不同的網站，代理ip在其中的所在位置也都不同，函數如下：
def get_data5u_free_ip(ip_proxies, save_path, open_proxy): """ 獲取無憂代理的免費ip :param ip_proxies: 要使用的代理ip（這里是用代理ip去爬代理ip） :param save_path: 代理ip保存路徑 :param open_proxy: 是否開啟代理，默認為False :return: """ url_list = [ "http://www.data5u.com/free/index.shtml", "http://www.data5u.com/free/gngn/index.shtml", "http://www.data5u.com/free/gnpt/index.shtml", "http://www.data5u.com/free/gwgn/index.shtml", "http://www.data5u.com/free/gwpt/index.shtml" ] ip_list_sum = [] # 代理ip列表 for i in range(5): res_text = get_html(url_list[i], open_proxy=open_proxy, ip_proxies=ip_proxies) # 抓取錯誤頁面，主動報異常 if res_text.find("錯誤") != -1: raise AttributeError('錯誤頁面') # 頁面解析 soup = BeautifulSoup(res_text, "html.parser") tags = soup.find_all("ul", class_="l2") for tag in tags: ip_list = [] ip_info_format = "" sps = tag.find_all("li") for sp in sps: ip_info = sp.get_text() ip_list.append(ip_info) for j in range(len(sps)): # 格式化IP信息 if j == len(sps) - 1: ip_info_format += str(ip_list[j]) + "\n" else: ip_info_format += str(ip_list[j]) + "___" ip_list_sum.append(ip_info_format) save_ip(ip_list_sum, save_path) def get_kuaidaili_free_ip(ip_proxies, save_path, open_proxy): """ 獲取快代理的免費ip :param ip_proxies: 要使用的代理ip（這里是用代理ip去爬代理ip） :param save_path: 代理ip保存路徑 :param open_proxy: 是否開啟代理，默認為False :return: """ ip_list_sum = [] # 代理ip列表 for i in range(10): # 獲取頁數 res_text = get_html("https://www.kuaidaili.com/ops/proxylist/" + str(i+1) + "/", open_proxy=open_proxy, ip_proxies=ip_proxies) # 頁面解析 soup = BeautifulSoup(res_text, "html.parser") tags = soup.find_all("div", id="freelist") for tag in tags: ip_list = [] sps = tag.find_all("td") for sp in sps: ip_info = sp.get_text() ip_list.append(ip_info) for j in range(10): # 每頁100條數據 ip_info_format = "" for k in range(8): # 每條6個內容 if k == 7: ip_info_format += str(ip_list[(j * 8 + k)]) + "\n" else: ip_info_format += str(ip_list[(j * 8 + k)]) + "___" ip_list_sum.append(ip_info_format) save_ip(ip_list_sum, save_path) def get_xsdaili_free_ip(ip_proxies, save_path, open_proxy): """ 獲取小舒代理的免費ip :param ip_proxies: 要使用的代理ip（這里是用代理ip去爬代理ip） :param save_path: 代理ip保存路徑 :param open_proxy: 是否開啟代理，默認為False :return: """ url = "http://www.xsdaili.com/" url_list = [] home_page = get_html(url, open_proxy=open_proxy, ip_proxies=ip_proxies) # 首頁解析 home_soup = BeautifulSoup(home_page, "html.parser") home_tags = home_soup.find_all("div", class_="title") for home_tag in home_tags: home_url = home_tag.a["href"] new_url = "http://www.xsdaili.com" + str(home_url) url_list.append(new_url) # 頁面解析 ip_list_sum = [] for i in range(len(url_list)): # 頁面頁數 res_text = get_html(url_list[i], open_proxy=open_proxy, ip_proxies=ip_proxies) # 頁面解析 soup = BeautifulSoup(res_text, "html.parser") tags = soup.find("div", class_="cont") ip_info = tags.get_text() ip_info_temp = ip_info.replace("\r\n\t\t\t\t\t\t\t", "") ip_list = re.split(r'[:@# \] ]', ip_info_temp) # 分割字符串 for j in range(100): # 每頁100條數據 ip_info_format = "" for k in range(6): # 每條6個內容 if k == 5: ip_info_format += str(ip_list[(j * 6 + k)]) + "\n" else: ip_info_format += str(ip_list[(j * 6 + k)]) + "___" ip_list_sum.append(ip_info_format) save_ip(ip_list_sum, save_path) def get_xicidaili_free_ip(ip_proxies, save_path, open_proxy): """ 獲取西刺代理的免費ip :param ip_proxies: 要使用的代理ip（這里是用代理ip去爬代理ip） :param save_path: 代理ip保存路徑 :param open_proxy: 是否開啟代理，默認為False :return: """ ip_list_sum = [] for i in range(10): # 獲取頁數 res_text = get_html("http://www.xicidaili.com/nn/" + str(i+1), open_proxy=open_proxy, ip_proxies=ip_proxies) # 抓取錯誤頁面，主動報異常 # print(res_text) if res_text.find("錯誤") != -1: # 錯誤頁面 raise AttributeError('錯誤頁面') elif res_text == "block": # 空白頁面 raise AttributeError('錯誤頁面') # 頁面解析 soup = BeautifulSoup(res_text, "html.parser") tags = soup.find_all("tr", class_="") for tag in tags: ip_list = [] ip_ths = tag.find_all("td") for ip_th in ip_ths: ip_info = ip_th.get_text().replace("\n", "") if ip_info != "": ip_list.append(ip_info) try: ip_info_format = "" for k in range(7): # 每條6個內容 if k == 6: ip_info_format += str(ip_list[k]) + "\n" else: ip_info_format += str(ip_list[k]) + "___" ip_list_sum.append(ip_info_format) except Exception as e: # print(e) pass save_ip(ip_list_sum, save_path) def get_89ip_free_ip(ip_proxies, save_path, open_proxy): """ 獲取89免費代理的免費ip :param ip_proxies: 要使用的代理ip（這里是用代理ip去爬代理ip） :param save_path: 代理ip保存路徑 :param open_proxy: 是否開啟代理，默認為False :return: """ ip_list_sum = [] for i in range(10): # 獲取頁數 res_text = get_html("http://www.89ip.cn/index_" + str(i+1) + ".html", open_proxy=open_proxy, ip_proxies=ip_proxies) # 抓取錯誤頁面，主動報異常 if res_text.find("錯誤") != -1: # 錯誤頁面 raise AttributeError('錯誤頁面') # 頁面解析 soup = BeautifulSoup(res_text, "html.parser") tags = soup.find_all("tbody") for tag in tags: ip_ths = tag.find_all("tr") for ip_th in ip_ths: ip_tds = ip_th.find_all("td") ip_list = [] for ip_td in ip_tds: ip_info = re.split(r'[\t\n ]', ip_td.get_text()) # 分割字符串 for j in range(len(ip_info)): if ip_info[j] != "": ip_list.append(ip_info[j]) ip_info_format = "" for k in range(len(ip_list)): # 每條6個內容 if k == len(ip_list) - 1: ip_info_format += str(ip_list[k]) + "\n" else: ip_info_format += str(ip_list[k]) + "___" ip_list_sum.append(ip_info_format) save_ip(ip_list_sum, save_path)
3，驗證代理ip（check_ip.py）
3.1 文件查重
寫一個簡單的函數用於查重，查重思路就是獲取txt文件中的每一行元素，組成列表。對列表使用自帶的set函數，注意：這種方法會改變列表原來的順序。函數如下：
from bs4 import BeautifulSoup import requests import time def check_repeat(path): """ 檢查文件中每一行的內容是否重復，刪除重復內容 :param path: 文件路徑 :return: """ try: # 讀取文件 data_list = [] with open(path, "r") as fr: lines = fr.readlines() fr.close() for line in lines: data_list.append(line) new_data_list = list(set(data_list)) # 查重 file_name = path.split("/") print(file_name[-1] + "文件共有 " + str(len(data_list)) + " 條數據") print("經過查重,現在共有 " + str(len(new_data_list)) + " 條數據") # 保存文件 with open(path, "w") as f: for i in range(len(new_data_list)): f.write(new_data_list[i]) f.close() print(file_name[-1] + "文件查重成功") except Exception as e: print("文件查重失敗！！！") print(e)
3.2 代理IP格式化
網頁獲取的ip的格式基本為以下格式：
5.135.66.232___8554___透明___http___法國___XXXX___XX___7.869 秒___4秒前 220.230.120.101___8286___高匿___http___韓國___XXXX___XX___4.14 秒___11秒前
從頁面解析出來的代理ip的信息中間都是使用“—”進行隔開，為了方便直接使用，在此需要將上述的的格式轉換為以下格式：
5.135.66.232:8554 220.230.120.101:8286
函數如下：
def ip_format(read_path, save_path): """ 將文件中的代理ip進行格式化轉換，並進行查重 :param read_path: 讀取待轉換的代理ip的文件路徑 :param save_path: 轉換完成的代理ip的保存路徑 :return: """ data_list = [] with open(read_path, "r") as fr: lines = fr.readlines() fr.close() for line in lines: new_line = line.split("___") ip_format_line = new_line[0].replace(" ", "") + ":" + new_line[1] + "\n" data_list.append(ip_format_line) with open(save_path, "a") as fs: for i in range(len(data_list)): fs.write(data_list[i]) fs.close() print("文件保存成功") fs.close()
3.3 代理IP驗證
由於免費的代理ip很多都是無法使用，或是不穩定，或是時效短。所以驗證代理ip是否可用，就非常有必要。主要驗證原理：使用代理ip去訪問網頁，判斷是否能夠正常訪問。在此我選擇的網站是“站長之家”，這個網站可用直接返回你當前使用的ip以及ip所在地。這里需要注意的是訪問前可以設定連接超時的時間如果訪問時間超過一定時間，就直接跳過這個代理ip。建議是設定在2秒內，具體的可以看以下函數：
def ip_test(ip_proxies): """ 驗證單個代理ip是否可用 :param ip_proxies: 待驗證ip，例如：101.96.10.36:88 :return: """ url = "http://ip.chinaz.com/" headers = { "Host": "ip.chinaz.com", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64; rv:60.0) Gecko/20100101 Firefox/60.0", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2", "Accept-Encoding": "gzip, deflate", "Referer": "https://blog.csdn.net/Winterto1990/article/details/51220307", "Connection": "keep-alive", "Upgrade-Insecure-Requests": "1", "Cache-Control": "max-age=0", } proxies = {"http": "http://" + ip_proxies, } # 設置代理 res = requests.get(url, headers=headers, proxies=proxies, timeout=1) # timeout為設定的相應時長，建議在2秒內 # 解析網頁 soup = BeautifulSoup(res.text, "html.parser") info_list = soup.find_all("p", {"class": "getlist pl10"}) for info in info_list: is_local = info.get_text() print(info.get_text()) return is_local.find("XXX.XXX.XXX.XXX") # 判斷是否為本地的地址
3.4. 批量驗證代理ip
最后便是批量的對代理ip進行驗證，實際上這就是調用3.3.中驗證代理ip中的程序。具體程序如下：
def ip_batch_inspection(read_path, save_path): """ 驗證多個代理ip是否可用 :param read_path: 代理ip文件路徑 :param save_path: 驗證可用的代理ip保存路徑 :return: """ with open(read_path, "r") as fr: lines = fr.readlines() fr.close() count = 0 file_name = read_path.split("/") print(file_name[-1] + "文件共有 " + str(len(lines)) + " 條數據") for line in lines: count += 1 ip_proxies = line.replace("\n", "") try: is_local = ip_test(ip_proxies) # 如果是本地ip，返回值為大於0數值 if is_local < 0: with open(save_path, "a") as fs: fs.write(ip_proxies + "\n") except Exception as e: pass # print("ip不可用") print("驗證中......%.2f%%" % (count/len(lines)*100)) print("驗證完畢")
4. 使用的例程（get_proxy.py）
將上述的函數都整合起來，就能夠實現批量獲取免費的代理ip。
# @File : get_proxy # @Description: 下載+驗證，獲取可用ipfrom check_ip import * from parse_html import * def main(): today = time.strftime("%Y_%m_%d") # 當前日期 ip_pools_path = "ip_proxy\\" + today + "_ip_pools.txt" # 原始ip保存路徑 ip_format_pools_path = "ip_proxy\\" + today + "_ip_format_pools.txt" # 格式化后ip保存路徑 ip_use_path = "ip_proxy\\" + today + "_ip_use.txt" # 可用ip保存路徑 open_proxy = True # 是否要開啟代理模式 if not open_proxy: # 不開啟代理模式，直接獲取代理ip get_data5u_free_ip(None, ip_pools_path) get_kuaidaili_free_ip(None, ip_pools_path) get_xsdaili_free_ip(None, ip_pools_path) get_xicidaili_free_ip(None, ip_pools_path) get_89ip_free_ip(None, ip_pools_path) else: # 開啟代理模式，獲取代理ip available_ip_path = "ip_proxy\\ip_use.txt" # 目前可用的代理ip的保存路徑 ip_use_list = [] with open(available_ip_path, "r") as fr: ip_use_lines = fr.readlines() for ip_use_line in ip_use_lines: ip_use_line_new = ip_use_line.replace("\n", "") ip_use_list.append(ip_use_line_new) for i in range(len(ip_use_list)): # 獲取ip建立IP池 try: print("正在使用第" + str(i) + "條代理ip") get_data5u_free_ip(ip_use_list[i], ip_pools_path, open_proxy) break except: pass for i in range(len(ip_use_list)): # 獲取ip建立IP池 try: print("正在使用第" + str(i) + "條代理ip") get_kuaidaili_free_ip(ip_use_list[i], ip_pools_path, open_proxy) break except: pass for i in range(len(ip_use_list)): # 獲取ip建立IP池 try: print("正在使用第" + str(i) + "條代理ip") get_xsdaili_free_ip(ip_use_list[i], ip_pools_path, open_proxy) break except: pass for i in range(len(ip_use_list)): # 獲取ip建立IP池 try: print("正在使用第" + str(i) + "條代理ip") get_xicidaili_free_ip(ip_use_list[i], ip_pools_path, open_proxy) break except: pass for i in range(len(ip_use_list)): # 獲取ip建立IP池 try: print("正在使用第" + str(i) + "條代理ip") get_89ip_free_ip(ip_use_list[i], ip_pools_path, open_proxy) break except: pass # 篩選ip進行查重 ip_format(ip_pools_path, ip_format_pools_path) check_repeat(ip_format_pools_path) # 驗證ip可用性 ip_batch_inspection(ip_format_pools_path, ip_use_path) if __name__ == '__main__': main()