Urllib3+BeautifulSoup+Thread：多线程爬⾍教程（爬取西刺代理IP）导⼊模块
from bs4 import BeautifulSoup
import urllib3
import urllib
import threading
Urllib3+BeautifulSoup+Thread：多线程爬⾍教程（爬取西刺代理IP）
爬⾍是学python中最好玩的学习⽅法，也是⽅便⽣活、⼯作中的有效利器，爬⾍帮我们⼯作，那会节省⼤量的⼈⼒。
在该博客中，分成5个步骤来简单教⼤家如何⽤好爬⾍。最后贴出源码。
1 函数：获取⽹页内容
#获取⽹页HTML码
def getContent(url):
http=urllib3.PoolManager()
quest('get',url)
#⽹页返回200表⽰可打开⽹页
print('⽹页 %s 返回代码：%s'%(url,response.status))
return response.data
2 获取每页的代理IP
Proxys=[]
pages=5
#爬去西刺10页的代理IP
for i in range(0,pages):
url='www.xicidaili/nn/%d'%(i+1)
content=getContent(url) #获取⽹页内容
if content!=None:
soup=BeautifulSoup(content,'lxml')
#分析⽹页源码得，tr标签中class=''或class='odd'都满⾜
Tags=soup.find_all('tr',attrs={'class':'odd'})+soup.find_all('tr',attrs={'class':''})[1:]
#获取该页所有的代理IP
for tag in Tags:
t=tag.find_all('td')
ip=t[1].get_text()
post=t[2].get_text()
protocol=t[5].get_text().lower() #变⼩写
python 爬虫教学Proxys.append(protocol+'://'+ip+':'+post)
print('共爬取了%d个代理IP'%len(Proxys))
代码结果：
⽹页 www.xicidaili/nn/1 返回代码：200
⽹页 www.xicidaili/nn/2 返回代码：200
⽹页 www.xicidaili/nn/3 返回代码：200
⽹页 www.xicidaili/nn/4 返回代码：200
⽹页 www.xicidaili/nn/5 返回代码：200
共爬取了500个代理IP
3 函数：测试代理IP是否有效
在这⾥⽤的是urllib设置代理IP来测试代理IP是否有效。
def testProxy(pro):
protocol=pro.split(':')[0]
#设置urllib中的代理IP
quest.build_opener( quest.ProxyHandler({protocol:pro}) )
#若能打开test_url则保存该代理IP
try:
quest.urlopen(test_url,timeout=timeout)
except:
print("代理IP：%s ⽆效"%pro)
else:
if pro not in alive_Proxys:
alive_Proxys.append(pro)
4 多线程调⽤上述函数
单线程测试代理IP是⼗分缓慢的，故使⽤多线程。但在使⽤多线程中要注意join⽅法阻塞线程到结束。
test_url='baidu'
n=10#线程数
timeout=3
alive_Proxys=[] #保存可⽤代理
pro=Proxys.pop()
while Proxys:
T=[]
for i in range(n):
t=threading.Thread(target=testProxy,args=(pro,))
T.append(t)
t.start()
if Proxys:
pro=Proxys.pop()
else:
break
#阻塞n个进程结束后再重新创建线程
for t in T:
t.join()
print("可⽤代理IP：%d"%len(alive_Proxys))
代码结果：
代理IP：114.228.74.205:6666 ⽆效
代理IP：222.77.48.46:47437 ⽆效
代理IP：223.241.118.205:18118 ⽆效
代理IP：123.163.178.116:25746 ⽆效
#中间省略了很多....
可⽤代理IP：339
5 写⼊⽂件
with open('','w') as fp:
for line in alive_Proxys:
fp.write("%s\n"%line)
print("写⼊完毕")
代码结果：
写⼊完毕
源码
from bs4 import BeautifulSoup
import urllib3
import urllib
import threading
#获取⽹页HTML码
def getContent(url):
http=urllib3.PoolManager()
quest('get',url)
print('⽹页 %s 返回代码：%s'%(url,response.status))
return response.data
#测试代理是否可⽤
def testProxy(pro):
protocol=pro.split(':')[0]
#设置urllib中的代理IP
quest.build_opener( quest.ProxyHandler({protocol:pro}) )
#若能打开test_url则保存该代理IP
try:
quest.urlopen(test_url,timeout=timeout)
except:
print("代理IP：%s ⽆效"%pro)
else:
if pro not in alive_Proxys:
alive_Proxys.append(pro)
#获取代理IP
Proxys=[]
pages=5
#爬去西刺10页的代理IP
for i in range(0,pages):
url='www.xicidaili/nn/%d'%(i+1)
content=getContent(url) #获取⽹页内容
if content!=None:
soup=BeautifulSoup(content,'lxml')
#分析⽹页源码得，tr标签中class=''或class='odd'都满⾜
Tags=soup.find_all('tr',attrs={'class':'odd'})+soup.find_all('tr',attrs={'class':''})[1:]
#获取该页所有的代理IP
for tag in Tags:
t=tag.find_all('td')
ip=t[1].get_text()
post=t[2].get_text()
protocol=t[5].get_text().lower()
Proxys.append(protocol+'://'+ip+':'+post)
print('共爬取了%d个代理IP'%len(Proxys))
#测试代理IP是否可⽤
test_url='baidu'
n=10#线程数
timeout=3
alive_Proxys=[] #保存可⽤代理
pro=Proxys.pop()
while Proxys:
T=[]
T=[]
for i in range(n):
t=threading.Thread(target=testProxy,args=(pro,))
T.append(t)
t.start()
if Proxys:
pro=Proxys.pop()
else:
break
#阻塞n个进程结束后再重新创建线程
for t in T:
t.join()
print("可⽤代理IP：%d"%len(alive_Proxys))
#写⼊⽂件
with open('','w') as fp:
for line in alive_Proxys:
fp.write("%s\n"%line)
print("写⼊完毕")
谢谢⼤家的浏览，
希望我的努⼒能帮助到您，
共勉！

版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。