寫(xiě)了一個(gè)爬取免費(fèi)代理網(wǎng)站的爬蟲(chóng),時(shí)間倉(cāng)促,本著工具類(lèi)能用就行的原則,所以很多細(xì)節(jié)沒(méi)有做。
由于免費(fèi)代理有很多沒(méi)法用,所以寫(xiě)了檢測(cè)可用的功能,只保存可用的ip,并且以字典形式保存,方便requests中的proxies直接調(diào)用。
想要一次性爬取10頁(yè)的內(nèi)容,如果不開(kāi)多線程速度有點(diǎn)慢。
先上爬蟲(chóng)代碼:
import requests
import re, os, datetime, time, random
from pyquery import PyQuery as pq
from bs4 import BeautifulSoup as bs
user_agents = ['Mozilla/5.0 (Windows NT 6.1; rv:2.0.1) Gecko/20100101 Firefox/4.0.1','Mozilla/5.0 (Windows; U; Windows NT 6.1; en-us) AppleWebKit/534.50 (KHTML, like Gecko) Version/5.1 Safari/534.50','Opera/9.80 (Windows NT 6.1; U; en) Presto/2.8.131 Version/11.11']
headers = {'User-Agent':random.choice(user_agents)}
ip = []
port = []
typ = []
http_url = 'https://www.xicidaili.com/nt/' # 爬取目標(biāo),免費(fèi)代理網(wǎng)站
test_url = 'http://icanhazip.com' # 檢測(cè)ip的網(wǎng)站
ip_list = []
r = requests.get(http_url, headers=headers, timeout=30)
r.raise_for_status()
r.encoding = r.apparent_encoding
html = r.text
doc = pq(html)
page_url = []
# 頁(yè)碼列表,頁(yè)碼改變時(shí),底下750這個(gè)數(shù)值也要改變,每一頁(yè)50個(gè)ip,可以自己算一下
page = ['1', '2', '3', '4', '5','6','7','8','9','10']
for i in page:
page_url.append('https://www.xicidaili.com/nt/' + i)
for i in page_url:
odd = doc('.odd')
for i in odd.items():
for i in i('td:nth-child(2)').items():
ip.append(i.text())
for i in odd.items():
for i in i('td:nth-child(3)').items():
port.append(i.text())
for i in odd.items():
for i in i('td:nth-child(6)').items():
typ.append(i.text())
def get_ip(i, path):
url = {}
url[ip_list[750+i]] = ip_list[750+i] + ":" + ip_list[i] + ":" + ip_list[500+i]
try:
response = requests.get(test_url, headers=headers, proxies=url, timeout=30) # 此段代碼檢測(cè)代理是否可用,可用的話就保存到txt文件
print(response.status_code)
if response.status_code == 200:
with open(path, 'a+', encoding='utf-8') as f:
f.write(str(url) + '\n')
except requests.ConnectionError as e:
print(e.args)
ip_list = ip + port + typ
def ip_check(ip_list):
path = 'C:\\Users\\hj506\\Desktop\\ip.txt' # 保存路徑
for i in range(250):
get_ip(ip, path)
ip_check(ip_list)
多線程需要引入threading庫(kù),只需要把ip_check函數(shù)改一下即可,其他不用動(dòng):
import threading
def ip_check(ip_list):
path = 'C:\\Users\\hj506\\Desktop\\ip.txt' # 保存路徑
for i in range(250):
t = threading.Thread(target=get_ip, args=(i, path))
t.start()