前言
为啥要用代理？ 不管是爆破还是爬虫，经常都会遇见ip被封掉的情况，此时，自建一个可靠的代理池就显得很有必要了
代理有哪些分类 以代理的匿名度来分，可以分为透明代理、匿名代理和高度匿名代理
- 使用透明代理，对方服务器可以知道你使用了代理，并且也知道你的真实IP。
- 使用匿名代理，对方服务器可以知道你使用了代理，但不知道你的真实IP。
- 使用高匿名代理，对方服务器不知道你使用了代理，更不知道你的真实IP。
此篇文章就是爬取国内比较出名的西刺代理(还有一个快代理也比较出名，爬取步骤和这个差不多)的高匿代理并且验证可用性
思路
获取信息
首先，我们需要用requests库把网页爬下来，然后查看我们需要的信息在哪个标签页里
可以看到每一个代理的所有信息都包含在一个tr标签中，而每一个信息又都是放在一个td标签中的:
那么我们就可以使用lxml库来解析，用xpath来寻找标签，简单示例代码如下：
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解析所需要的标签用xpath获得指定标签页的内容可以参考这篇文章：https://www.cnblogs.com/lei0213/p/7506130.html
在获得每个端口ip、端口、类型、速度、连接时间等信息之后，我们可以进行筛选：
- 将http、https类型的代理ip分类
- 将速度及连接时间不理想的ip过滤掉
验证有效性
之后，我们可以将代理ip拿去验证
验证方法的话，网上大部分是再将代理ip加载请求头里面拿去请求某个网站，比如百度
如果返回状态码是200的话，则此代理ip有效
但是这样太过于依赖网络并且效率也不高，看到一篇文章是用每个ip用telnet去连接一下，如果能连接上，则是有效的
这部分主要的一个参数就是超时时间，也就是等待时间，建议设置为10秒
用telnet验证代理ip有效性的简单示例如下：
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")用多线程优化
在获取信息，验证有效性都可行之后，发现效率太低了，主要耗费时间的地方有这些：
- 请求多页耗费时间
- 验证代理需要有一个超时等待时间，后面的队伍都需要等待
那么我们可以使用multiprocessing库中的Pool线程池配合apply_async()这个方法使用
这个线程池可以配合apply()方法使用，但是apply()是阻塞的。
首先主进程开始运行，碰到子进程，操作系统切换到子进程，等待子进程运行结束后，在切换到另外一个子进程，直到所有子进程运行完毕。然后在切换到主进程，运行剩余的部分。这样跟单进程串行执行没什么区别。
apply_async 是异步非阻塞的。
即不用等待当前进程执行完毕，随时根据系统调度来进行进程切换。首先主进程开始运行，碰到子进程后，主进程仍可以先运行，等到操作系统进行进程切换的时候，在交给子进程运行。可以做到不等待子进程执行完毕，主进程就已经执行完毕，并退出程序。
由于我们是获取到代理、验证有效性就行，不要求顺序，不需要像爬取小说那样爬取之后需要按照章节顺序存入文件
所以我们这里使用apply_async是最合适的。
multiprocessing库中的Pool线程池配合apply_async()这个方法的简单示例代码如下：
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))
提示：由于代理的时效性，不建议长期存储，建议“即用即爬”，另外请勿重复使用，短时间内使用两三次就会封掉ip，如果ip被封的话挂vpn即可
效果展示
可以看到免费的高匿ip的有效个数还是很少的，376个http代理只有53个可用，409和https代理只有72个可用。
