平時(shí)當(dāng)我們需要爬取一些我們需要的數(shù)據(jù)時(shí),總是有些網(wǎng)站禁止同一IP重復(fù)訪(fǎng)問(wèn),這時(shí)候我們就應(yīng)該使用代理IP,每次訪(fǎng)問(wèn)前偽裝自己,讓“敵人”無(wú)法察覺(jué)。
oooooooooooooooOK,讓我們愉快的開(kāi)始吧!
這個(gè)是獲取代理ip的文件,我將它們模塊化,分為三個(gè)函數(shù)
注:文中會(huì)有些英文注釋?zhuān)菫榱藢?xiě)代碼方便,畢竟英文一兩個(gè)單詞就ok了
#!/usr/bin/python#-*- coding:utf-8 -*-"""author:dasuda"""import urllib2import reimport socketimport threadingfindIP = [] #獲取的原始IP數(shù)據(jù)IP_data = [] #拼接端口后的IP數(shù)據(jù)IP_data_checked = [] #檢查可用性后的IP數(shù)據(jù)findPORT = [] #IP對(duì)應(yīng)的端口available_table = [] #可用IP的索引def getIP(url_target): patternIP = re.compile(r'(?<=<td>)[/d]{1,3}/.[/d]{1,3}/.[/d]{1,3}/.[/d]{1,3}') patternPORT = re.compile(r'(?<=<td>)[/d]{2,5}(?=</td>)') print "now,start to refresh proxy IP..." for page in range(1,4): url = 'http://www.xicidaili.com/nn/'+str(page) headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64)"} request = urllib2.Request(url=url, headers=headers) response = urllib2.urlopen(request) content = response.read() findIP = re.findall(patternIP,str(content)) findPORT = re.findall(patternPORT,str(content)) #assemble the ip and port for i in range(len(findIP)): findIP[i] = findIP[i] + ":" + findPORT[i] IP_data.extend(findIP) print('get page', page) print "refresh done!!!" #use multithreading mul_thread_check(url_target) return IP_data_checkeddef check_one(url_check,i): #get lock lock = threading.Lock() #setting timeout socket.setdefaulttimeout(8) try: ppp = {"http":IP_data[i]} proxy_support = urllib2.ProxyHandler(ppp) openercheck = urllib2.build_opener(proxy_support) urllib2.install_opener(openercheck) request = urllib2.Request(url_check) request.add_header('User-Agent',"Mozilla/5.0 (Windows NT 10.0; WOW64)") html = urllib2.urlopen(request).read() lock.acquire() print(IP_data[i],'is OK') #get available ip index available_table.append(i) lock.release() except Exception as e: lock.acquire() print('error') lock.release()def mul_thread_check(url_mul_check): threads = [] for i in range(len(IP_data)): #creat thread... thread = threading.Thread(target=check_one, args=[url_mul_check,i,]) threads.append(thread) thread.start() print "new thread start",i for thread in threads: thread.join() #get the IP_data_checked[] for error_cnt in range(len(available_table)): aseemble_ip = {'http': IP_data[available_table[error_cnt]]} IP_data_checked.append(aseemble_ip) print "available proxy ip:",len(available_table)一、getIP(url_target):主要函數(shù) 傳入?yún)?shù)是:驗(yàn)證代理IP可用性的網(wǎng)址,推薦ipchina
獲取代理IP,從http://www.xicidaili.com/nn/網(wǎng)站獲取,它是一個(gè)提供免費(fèi)代理IP的網(wǎng)站,但是里面的IP不是全都能用,而且結(jié)合你的實(shí)際地理位置、網(wǎng)絡(luò)情況、訪(fǎng)問(wèn)的目標(biāo)服務(wù)器等情況,能用的大概不到20%,至少我的情況是這樣。
訪(fǎng)問(wèn)http://www.xicidaili.com/nn/網(wǎng)站使用正常方式,返回的網(wǎng)頁(yè)內(nèi)容通過(guò)正則查詢(xún)獲得需要的IP和對(duì)應(yīng)端口,代碼如下:
patternIP = re.compile(r'(?<=<td>)[/d]{1,3}/.[/d]{1,3}/.[/d]{1,3}/.[/d]{1,3}')patternPORT = re.compile(r'(?<=<td>)[/d]{2,5}(?=</td>)')...findIP = re.findall(patternIP,str(content))findPORT = re.findall(patternPORT,str(content))關(guān)于如何構(gòu)造正則表達(dá)式,可以參考其他的文章:
獲取的IP保存在findIP中,對(duì)應(yīng)的端口在findPORT中,兩者按索引對(duì)應(yīng),獲取一頁(yè)IP正常數(shù)量為100.
接下來(lái)進(jìn)行IP和端口拼接
最后進(jìn)行可用性檢查
二、check_one(url_check,i):線(xiàn)程函數(shù)
本次訪(fǎng)問(wèn)url_check還是使用正常方式訪(fǎng)問(wèn),當(dāng)訪(fǎng)問(wèn)網(wǎng)頁(yè)有返回時(shí),則說(shuō)明本代理IP可用,則記錄下當(dāng)前索引值,用于后面將所有可用IP取出。
三、mul_thread_check(url_mul_check):多線(xiàn)程生成
本函數(shù)開(kāi)啟多線(xiàn)程檢查代理IP可用性,每個(gè)IP開(kāi)啟一個(gè)線(xiàn)程進(jìn)行檢查。
本工程直接調(diào)用getIP(),并傳入用于檢查可用性的網(wǎng)址,即可返回一個(gè)列表,里面是經(jīng)可用性檢查的ip列表,格式為
['ip1:port1','ip2:port2',....]
以上這篇python獲取代理IP的實(shí)例分享就是小編分享給大家的全部?jī)?nèi)容了,希望能給大家一個(gè)參考,也希望大家多多支持VEVB武林網(wǎng)。
新聞熱點(diǎn)
疑難解答
圖片精選
網(wǎng)友關(guān)注