python网络爬虫代理_python爬虫实战python3批量抓取并验证快代理免费代理地址

特别声明&＃xff1a;本文为原创&＃xff0c;可自由转载、引用&＃xff0c;但需署名作者且注明文章出处&＃xff0c;如有侵权请联系&＃xff01;

使用python写个爬虫批量抓取并验证快代理免费代理地址import requests

import threading

from bs4 import BeautifulSoup

from queue import Queue

headers&＃61;{

&＃39;User-Agent&＃39;:&＃39;Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/71.0.3578.98 Safari/537.36&＃39;,

}

# 分隔符

Separator&＃61;&＃39;|&＃39;

data_queue&＃61;Queue()

def get_list(page):

countNum&＃61;0

All_proxy&＃61;[]

for p in range(1,page&＃43;1):

url&＃61;&＃39;https://www.kuaidaili.com/free/inha/{}/&＃39;.format(p)

print(url)

r&＃61;requests.get(url,headers&＃61;headers)

soup&＃61;BeautifulSoup(r.text,&＃39;lxml&＃39;)

trs &＃61; soup.find(&＃39;table&＃39;, class_&＃61;&＃39;table&＃39;).find_all(&＃39;tr&＃39;)

for tr in trs[1:]:

tds &＃61; tr.find_all(&＃39;td&＃39;)

ip&＃61;tds[0].text.strip()

port&＃61;tds[1].text.strip()

anony&＃61;tds[2].text.strip()

protocol&＃61;tds[3].text.strip()

locate&＃61;tds[4].text.strip()

time&＃61;tds[6].text.strip()

proxy_str&＃61;ip&＃43;Separator&＃43;port&＃43;Separator&＃43;protocol&＃43;Separator&＃43;anony&＃43;Separator&＃43;locate&＃43;Separator&＃43;time

data_queue.put(proxy_str)

All_proxy.append(proxy_str)

print(proxy_str)

countNum &＃43;&＃61; 1

text_save(&＃39;Alldaili_kuai.txt&＃39;,All_proxy)

return countNum

def verifyProxyList(items):

myurl &＃61; &＃39;http://www.baidu.com/&＃39;

while 1:

print(&＃39;线程 %s&＃xff1a;启动&＃xff1b;还有%d条数据待处理&＃xff01;&＃39;%(threading.current_thread().name,data_queue.qsize()))

if data_queue.empty():

break

ll&＃61;data_queue.get()

line&＃61;ll.strip().split(Separator)

daili&＃61;&＃39;{}://{}:{}&＃39;.format(line[2].lower(),line[0],line[1])

if line[2].lower()&＃61;&＃61;&＃39;http&＃39;:

proxies&＃61;{

&＃39;http&＃39;:daili,

}

else:

proxies&＃61;{

&＃39;https&＃39;:daili,

}

try:

r&＃61;requests.get(url&＃61;myurl,headers&＃61;headers,proxies&＃61;proxies)

items.append(ll&＃43;Separator&＃43;&＃39;验证成功&＃39;)

print (daili&＃43;" 连接成功&＃xff01;")

except Exception as e:

# items.append(ll&＃43;Separator&＃43;&＃39;验证失败&＃39;)

print (daili&＃43;" 连接失败&＃xff01;")

print(&＃39;线程 %s&＃xff1a;结束&＃xff1b;还有%d条数据待处理&＃xff01;&＃39;%(threading.current_thread().name,data_queue.qsize()))

def text_save(filename, data):#filename为写入txt文件的路径&＃xff0c;data为要写入数据列表.

file &＃61; open(filename,&＃39;a&＃39;)

for i in range(len(data)):

s &＃61; str(data[i]).replace(&＃39;[&＃39;,&＃39;&＃39;).replace(&＃39;]&＃39;,&＃39;&＃39;)#去除[],这两行按数据不同&＃xff0c;可以选择

s &＃61; s.replace("&＃39;",&＃39;&＃39;).replace(&＃39;,&＃39;,&＃39;&＃39;) &＃43;&＃39;\n&＃39; #去除单引号&＃xff0c;逗号&＃xff0c;每行末尾追加换行符

file.write(s)

file.close()

print("保存文件成功")

def main():

page&＃61;int(input(&＃39;请输入你要爬取的页码总数&＃xff1a;&＃39;))

if page>10:

page&＃61;10

# page&＃61;10

countNum&＃61;get_list(page)

print(&＃39;共采集了%d页&＃xff0c;共%d条数据&＃xff01;&＃39;%(page,countNum))

print(&＃39;下面准备开始逐一验证&＃xff0c;请稍后......&＃39;)

all_thread &＃61; []

verify_list&＃61;[]

for i in range(1,31):

t &＃61; threading.Thread(target&＃61;verifyProxyList,name&＃61;"线程"&＃43;str(i),args&＃61;(verify_list,))

all_thread.append(t)

t.start()

for t in all_thread:

t.join()

text_save(&＃39;verified_kuaidaili.txt&＃39;,verify_list)

print(&＃39;验证结束&＃xff0c;共有%d条信息有效&＃39;%len(verify_list))

print ("All Done.")

if __name__ &＃61;&＃61; &＃39;__main__&＃39;:

main()

赏

文章来源&＃xff1a;

zyglz

版权声明&＃xff1a;本文为原创&＃xff0c;可自由转载、引用&＃xff0c;但需署名作者且注明文章出处&＃xff0c;如有侵权请联系&＃xff01;