肥皂的小屋

import requests
from lxml import etree

url = 'https://www.xicidaili.com/nn/1'  # 需要爬取的url
req = requests.get(url)  # 用requests請求
content = req.content.decode("utf-8")  # 將內容解碼
tree = etree.HTML(content)  # 解析html文件
tr_nodes = tree.xpath('.//table[@id="ip_list"]/tr')[1:]  # 使用xpath解析所需要的標籤

import telnetlib

def verifyProxy(ip, port):
try:
telnetlib.Telnet(ip, port, timeout=5) # 用 telnet 連接一下，能連通說明代理可用
except:
pass
else:
print("此代理已驗證可用：{}".format(proxies))

if **name** == "**main**":
verifyProxy("113.13.177.80", "9999")

from multiprocessing import Pool

def example(proxies):
if type(proxies) == 'http':
with open("1.txt", 'a') as f:
f.write(proxies+"\n")
else:
with open("2.txt", 'a') as f:
f.write(proxies+"\n")

if **name** == "**main**":
proxy = []
with open("tmp.txt", 'r') as f:
proxy = f.read().splitlines()
pool = Pool() # 新建一個執行緒池
for line in proxy:
pool.apply_async(target=example, args=(line,)) # target 這個參數名可以省略，建議在最後一個參數後面加上一個,
pool.close() # 創建進程池之後必須關閉
pool.join() # 加入阻塞隊列，則要上面的代碼執行完畢之後，才會執行後面的代碼

#!/usr/bin/python

# -_- coding: utf-8 -_-

'''
@author: soapffz
@function: 多執行緒爬取西刺高匿代理並驗證可用性
@time: 2019-01-21
'''

import requests
from lxml import etree
import re
import telnetlib
import threading
from multiprocessing import Pool
import os
import timeit

path = os.path.join(os.path.expanduser("~")+"\\") # 爬下來未驗證的的代理先放在當前用戶根目錄下
http_tmp_path = os.path.join(path+"http_tmp.txt")
https_tmp_path = os.path.join(path+"https_tmp.txt")

# 驗證過的代理就放在當前用戶桌面

http_proxy_path = os.path.join(path+"Desktop"+"\\"+"http_proxy.txt")
https_proxy_path = os.path.join(path+"Desktop"+"\\"+"https_proxy.txt")
http_proxy = []
https_proxy = []

def get_nn_proxy(page_num):
url = "https://www.xicidaili.com/nn/{}".format(page_num)
headers = {
"User-Agent": 'Mozilla/5.0 (Windows NT 6.2; WOW64; rv:21.0) Gecko/20100101 Firefox/21.0'} # 本來還測試使用隨機 User-Agent 頭，後來發現沒用，西刺封的是 ip
req = requests.get(url, headers=headers)
print("正在爬取第{}頁的內容...".format(page_num))
content = req.content.decode("utf-8")
tree = etree.HTML(content) # 用 xpath 獲得總的 ip_list
tr_nodes = tree.xpath('.//table[@id="ip_list"]/tr')[1:]
for tr_node in tr_nodes:
td_nodes = tr_node.xpath('./td') # 用 xpath 獲得單個 ip 的標籤
speed = int(
re.split(r":|%", td_nodes[6].xpath('./div/div/@style')[0])[1]) # 獲得速度的值
conn_time = int(
re.split(r":|%", td_nodes[7].xpath('./div/div/@style')[0])[1]) # 獲得連接時間的值
if(speed <= 95 | conn_time <= 95): # 如果速度和連接時間都不理想，就跳過這個代理
continue
ip = td_nodes[1].text
port = td_nodes[2].text
proxy_type = td_nodes[4].text
ip_type = td_nodes[5].text.lower()
proxy = "{}:{}".format(ip, port)
if ip_type == 'http':
with open(http_tmp_path, 'a') as f:
f.write("http://{}".format(proxy)+"\n") # 獲得爬下來的 http 代理並存在臨時文件中
else:
with open(https_tmp_path, 'a') as f: # 獲得爬下來的 https 代理並存在臨時文件中
f.write("https://{}".format(proxy)+"\n")

def verifyProxy(proxies):
ree = re.split(r'//|:', proxies)
ip_type = ree[0]
ip = ree[2]
port = ree[3]
try:
telnetlib.Telnet(ip, port, timeout=5) # 用 telnet 連接一下，能連通說明代理可用
except:
pass
else:
print("此代理已驗證可用：{}".format(proxies))
if ip_type == 'http':
with open(http_proxy_path, 'a') as f:
f.write(proxies+"\n")
else:
with open(https_proxy_path, 'a') as f:
f.write(proxies+"\n")

def clear_cache(path):
if os.path.exists(path):
os.remove(path)

if **name** == "**main**":
start_time = timeit.default_timer()
clear_cache(http_tmp_path)
clear_cache(https_tmp_path)
clear_cache(http_proxy_path)
clear_cache(https_proxy_path)
pool = Pool()
for i in range(1, 11): # 這裡爬取的是 1 到 10 頁的高匿代理，自行根據需要修改參數
pool.apply_async(get_nn_proxy, args=(i,))
pool.close()
pool.join()
if not os.path.exists(http_tmp_path): # 如果 ip 被 ban 掉，則獲取到的每一頁代理都是空列表，不會生成文件
print("你的 ip 已經被西刺 ban 掉了，請使用 ipconfig /release 之後使用 ipconfig /renew 換 ip 或者掛 vpn！")
os._exit(0)
with open(http_tmp_path, 'r') as f:
http_proxy = f.read().splitlines()
Unhttp_proxy_num = len(http_proxy)
with open(https_tmp_path, 'r') as f:
https_proxy = f.read().splitlines()
Unhttps_proxy_num = len(https_proxy)
pool2 = Pool()
pool3 = Pool()
for proxies in http_proxy:
pool2.apply_async(verifyProxy, args=(proxies,))
for proxies in https_proxy:
pool3.apply_async(verifyProxy, args=(proxies,))
pool2.close()
pool3.close()
pool2.join()
pool3.join()
http_proxy.clear()
https_proxy.clear()
with open(http_proxy_path, 'r') as f:
http_proxy = f.read().splitlines()
http_proxy_num = len(http_proxy)
with open(https_proxy_path, 'r') as f:
https_proxy = f.read().splitlines()
https_proxy_num = len(https_proxy)
print(http_proxy)
print(https_proxy)
print("已存儲在 http_proxy 和 https_proxy 這兩個列表中並已將 txt 保存在桌面")
end_time = timeit.default_timer()
print("共爬取{}個 http 代理，可用{}個，爬取{}個 https 代理，可用{}個，總耗時{}s".format(Unhttp_proxy_num,
http_proxy_num, Unhttps_proxy_num, https_proxy_num, end_time-start_time))