肥皂的小屋

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
@fucntion: 多线程爬取西刺高匿代理并验证可用性
@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))