使用爬虫实现代理IP池之放弃篇
啥叫代理IP以及代理IP池
概念上的东西网上搜索一下就好了,这里简单科普一下(大部分会读这篇文章的人,基本是不需要我来科普的),白话说就是能联网并提供代理访问互联网的服务器,它提供的IP就叫代理IP,至于代理IP池,就是一组代理IP构成的池子,跟我们编程用到的连接池也啥区别。
这里重点要说的是代理IP从隐藏级别上分三类:
- 透明代理,服务器知道你用了代理,但同时也知道你的真实IP,说白了是不以隐藏自己IP为目的使用的,比如翻墙什么的
- 普通代理,服务器也知道你用了代理,但不知道你的真实IP
- 高匿代理,服务器不知道你用了代理,更不知道你的真实IP(在被访问者看来,高匿代理就是你),“李代桃僵”说的就是这个吧,_
为什么我们会需要代理IP池
结合代理IP的特性,可以想到一些应用场景(不全,个人总结,仅供参考):
- 由于我国国情,很多人有翻墙的需求,比如说我(开玩笑,我基本不翻墙的,找不到路子主要是)
- 有些特殊的网站,设置了IP白名单,这样白名单之外的终端想要访问就得找白名单里的主机作为跳板来访问
- 可以提高某些网站的访问速度,比如GitHub,这个倒是没墙,但架不住老是打不开啊,说多都是泪,今天准备用
acme.sh
来申请HTTPS证书,结果 。。。 - 对我等
程序猿
而言,最重要的是隐藏自己的真实IP,不管是写爬虫也好,还是做点什么“不道德”的事也好,这个很重要
其它用途就不一一列举了(其实我就知道这么多),总之我们会需要一个代理IP(池,一般来说写爬虫才会用到池)
怎样用(Python)实现一个代理IP池
先说明一下,只是最近本人在研究Python,所以什么事都想用Python来搞一搞(我本Javaer
),实际上实现代理IP池跟语言关系并不大。
说到实现代理IP池,方法有很多,比如:
通过扫描器,扫描互联网上的主机,找到合适的IP就记录下来;
正规点的或者有钱点的,弄一大批机器,自建代理IP池(僵尸网络应该也能实现,这个就是“不道德”的一种了);
但对程序猿
而言,最经济的方法还是用爬虫来实现,网上有很多这种资源(现在叫它资源吧,后面就会知道这都是套路),比如 西刺代理
(先声明我并不是打广告,只是我的目标是它,实际上被它坑了)
实现的大致思路是写一个爬虫从目标网站(我用的是 http://www.xicidaili.com/nn/)把“资源”爬下来,并进行有效性验证
# -*- coding: utf-8 -*-
import requests
from pyquery import PyQuery as pq
import config
# 实现一个爬虫,用于从:http://www.xicidaili.com/nn 上获取代理IP数据
# 默认只爬取前4页数据,可自行调整 query_max_page 参数来控制
default_url = 'http://www.xicidaili.com/nn/'
# 生成爬虫目标URL列表
target_urls = [default_url + str(i) for i in range(1, config.query_max_page + 1)]
def get_proxies(url):
"""
下载并解析目标网页中的代理数据
:param url:
:return:
"""
try:
# 下载代理列表页
response = requests.get(url,
headers={
'User-Agent': config.default_user_agent,
'Host': 'www.xicidaili.com',
})
response.raise_for_status()
html = response.text
# 解析该页,获取全部代理IP及协议
for tr in pq(html).find('tr:gt(0)'):
tds = pq(tr).find('td')
yield tds[5].text.strip().lower(), tds[1].text.strip(), tds[2].text.strip()
except Exception as e:
print(e)
def get_all_proxies():
"""
获取全部代理数据
:return:
"""
for url in target_urls:
yield from get_proxies(url)
if __name__ == '__main__':
# 测试逻辑
count = 1
for proxy in get_all_proxies():
# 测试解包正常
print(proxy, *proxy)
count += 1
print('一共获取到:{} 个代理IP'.format(count))
对代理IP的有效性验证,我采用了一种最直接的办法,使用代理IP的目的是为了能访问一些网站,所以我直接用代理IP来访问测试网站,通用且在指定时间内正常响应,那么就算有效的
# -*- coding: utf-8 -*-
# 代理IP有效性验证,使用requests库检查,其代理设置参考:
# http://cn.python-requests.org/zh_CN/latest/user/advanced.html#proxies
import re
import time
from concurrent.futures.thread import ThreadPoolExecutor
import requests
import config
from crawler import get_all_proxies
def concat_proxy(item):
"""
将元组组装成代理字符串
:param item:
:return:
"""
return '{}://{}:{}'.format(*item)
def validate(protocol, ip, port):
"""
检查代理IP有效性
:param protocol:
:param ip:
:param port:
:return:
(True, 1) 表示有效且优质
(True, 0) 表示有效但普通
False 表示无效
"""
if check(protocol, ip, port, config.better_timeout):
return True, 1
elif check(protocol, ip, port, config.normal_timeout):
return True, 0
else:
return False, None
def check(protocol, ip, port, timeout=0.5):
"""
内部函数,检查代理IP是否有效
:param protocol:
:param ip:
:param port:
:param timeout:
:return:
"""
protocol = protocol.lower()
try:
# 根据协议自动切换测试URL
target_url = config.target_https_url
if protocol == 'http':
target_url = config.target_http_url
# 对HTTPS只能使用HTTPS代理,对HTTP只能使用HTTP代理
resp = requests.get(target_url,
timeout=timeout,
proxies={
protocol: concat_proxy((protocol, ip, port))
},
headers={
'User-Agent': config.default_user_agent,
})
return resp.status_code == 200
except Exception:
pass
return False
def test_target_website():
"""
这仅是一段测试代码,请忽略
# 连续10次无代理请求,测试目标网站的响应速度
# 这段代码是为了测算代理有效性的超时时间标准
# 其它算无效代理IP(响应太慢,没有什么实用价值)
:return:
"""
begin = time.time()
success_count = 0
for _ in range(10):
try:
if requests.get(config.target_https_url).status_code == 200:
success_count += 1
except Exception as e:
print(e)
end = time.time()
# 成功请求 10 次,总耗时 0.922 秒,平均单次请求耗时 0.092 秒
print('成功请求 {} 次,总耗时 {:.3f} 秒,平均单次请求耗时 {:.3f} 秒'.
format(success_count, end - begin, (end - begin) / 10))
if __name__ == '__main__':
# test_target_website()
def validate_proxy(url):
return validate(*re.split(r'(://|:)', url)[::2])
# # 从 http://ip.zdaye.com/FreeIPlist.html 找了几个试下效果
# # 前3个可以,后面的不行了
# print(validate_proxy('https://218.207.212.86:80'))
# print(validate_proxy('http://114.250.25.19:80'))
# print(validate_proxy('http://101.96.9.154:8080'))
# print(validate_proxy('http://183.232.185.85:8080'))
# print(validate_proxy('https://177.71.77.202:20183'))
#
# # 从 http://www.mayidaili.com/ 找几个试下
# # 都是1年前的,估计早就不能用了吧
# print(validate_proxy('https://174.120.70.232:80'))
# print(validate_proxy('https://120.52.32.46:80'))
# # http://www.swei360.com/
# # 也是坑货
# print(validate_proxy('https://95.143.109.146:41258'))
# print(validate_proxy('https://5.160.63.214:8080'))
# print(validate_proxy('https://93.126.59.74:53281'))
# # http://www.ip3366.net/
# # 也不比上面的强
# print(validate_proxy('https://138.118.85.217:53281'))
# print(validate_proxy('https://58.251.251.139:8118'))
# 放弃治疗了 ~~~
代码仅供参考,没有优化过(也基本不会优化了),里面涉及一些配置最后面会贴出来,或者直接访问:https://github.com/zlikun/python-proxy-ip-pool 查看即可
核心就是上述两段逻辑代码,使用定时任务启动爬虫,比如每5分钟从目标网站上爬取新的数据(定时爬取的原因是目标网站数据也是不停在变化的),校验有效性,有效的就加入代理IP池中,无效的直接丢弃(也可以缓存下来,后面再次爬取时,直接忽略)。仅仅只是这样还不够,网上提供的免费代理IP稳定性通常不高,所以还需要一个任务定时检查代理IP池里的代理IP有效性,无效即从池出剔除
# -*- coding: utf-8 -*-
# 代理IP池程序入口
import re
import time
from threading import Thread
import schedule
import config
from crawler import get_all_proxies
from process import DataProcessor
from validate import validate, concat_proxy
# 数据处理器
dp = DataProcessor()
def crawler_task():
for item in get_all_proxies():
# cannot unpack non-iterable bool object
flag, level = validate(*item)
if not flag:
continue
else:
# 加入代理IP池中
dp.save(concat_proxy(item), level)
print('[{}]完成爬虫任务'.format(time.ctime()))
def validate_task():
# 从代理IP池中获取数据
for proxy, level in dp.query():
# 执行校验,剔除无效数据
flag, level2 = validate(*re.split(r'(://|:)', proxy)[::2])
if not flag:
dp.remove(proxy, level)
else:
# 加入代理IP池中
dp.save(proxy, level2)
# 如果代理级别发生变化,则移除原集合中的代理IP
if level != level2:
dp.remove(proxy, level)
print('[{}]完成校验任务'.format(time.ctime()))
if __name__ == '__main__':
print('[{}]启动代理IP池维护任务'.format(time.ctime()))
# 手动执行一次,定时任务将会在下一个周期才开始执行
crawler_task()
# 启动爬虫定时任务
schedule.every(config.crawler_task_interval).seconds.do(crawler_task)
# 启动IP有效性校验任务
schedule.every(config.validate_task_interval).seconds.do(validate_task)
while True:
schedule.run_pending()
time.sleep(1)
通过爬虫实现代理IP池的大致思路就是这样,其它代码有兴趣的自行看看
实际上一开始我设计了挺多功能,比如设计了一套查询API,如:https://api.zlikun.com/https/random 随机返回一个有效的HTTPS代理IP(说是随机,实际每次都一样,因为池中就这一个),并且提供了Docker版本,通过 docker pull zlikun/proxy-ip-pool
获取,但不建议使用,有兴趣研究思路或代码的看看就好,实际应用中不要使用(代码里的域名api.zlikun.com
是我个人域名,对应的主机性能并不高,请勿并发访问)
网上那些代理IP那从哪来的
这个其实前面有说过,免费代理IP大都是通过扫描器扫描到的,时效性、质量都不高,又因为免费用得人多,所以几乎找不到几个能用的,通过看我代码就可以知道,我测试了好几家免费代理IP网站,几乎没有一家能提供10个以上有效代理IP的。
另外使用代理IP并不安全,有些人或组织回调代理IP服务是有目的的,比如钓鱼等,做爬虫对自己影响不大,但用来翻墙或者加速访问等,请慎重,“科学上网”同时也要注意安全上网。
结语
原本希望把这个项目做得比较完善一些,性能也优化一下的,现在看来是没这个必要了。
最近因为学习Python,爬虫研究的比较多,感觉有点掉坑里了,把自己主业都耽误了。后面一段时间应该不会再写爬虫了,不过有兴趣交流的爬虫的,欢迎“骚扰”。
最后打个广告,本人目前处于离职状态,有缺人的单位,可以联系我:https://zlikun.com/