文章有点儿长，谨慎食用~
python 爬虫在爬取网页内容时，遭遇的最常见的反爬措施就是 ip 限制/封禁，对此最常见的解决方式就是设置 IP 代理池，每次请求时随机使用一个代理 IP 去访问资源。
网上有成熟的代理服务，但是小伙汁的爬虫需求多是非定期的自定义项目，使用付费代理并不划算，遂有了爬取免费代理并测试是否可用，进而构建一个可用代理 IP 池的想法。本项目亦可作为后续网络相关服务的子模块。
版本1：先通过 request 或者 selenium 进行爬取；
版本2（大概率是鸽了~）：学习并使用 scrapy 进行爬取；
0 项目逻辑架构
经过在编码过程中不断的修修改改，重重构构，整体逻辑终于是有了一个相对解耦的模式，由于项目相比于大型项目来说还是 just like a toy，所以核心模块就是一个通用的 Spider 父类模块，定义了整体的爬虫逻辑，其余针对特定网页的实例都继承自该 Spider。先看图吧：
- Spider 是基础爬虫类，定义了一些静态的属性和功能方法
- spider x 是实例爬虫，每个实例爬虫需要根据自己网页的结构需要，重写 pre_parse()、parse()、get_all_proxies() 方法
- Proxy Manager 是运行时的代理管理类（目前仅简单提供可用代理的临时存储功能）
- 项目运行前，将多个对象爬虫实例化后配置在配置文件中，组成爬虫链。这样主函数执行时会串行加载每个爬虫实例并运行其爬取逻辑（之所以不用并行是想着后面的爬虫能利用前面爬虫验证过的代理，所以将那些没有反爬的爬虫实例尽量配在爬虫链的前面）
下面会详细的介绍每个模块的具体实现细节，再介绍之前，让我再 bb 几句吧。这个项目主要是出于个人兴趣，作为一个初入 python 爬虫领域的菜鸟，利用业余时间在拖拖拉拉中写完了这个项目，写的过程中也逐渐学习了一些 python 的高级语法，例如装饰器、自定义异常、多进程异步操作等，收获还是蛮多的。此外，由于免费代理资源本身并不是很稳定，指望通过免费代理资源来构建一个鲁棒的代理池还是有点困难的，所以这个项目更多的还是当学习使用。
食用提醒：
- 前置知识
- 最好还是要对 python requests 库有点了解，包括 请求、jsonpath 解析网页资源 等
- python 类的继承、多态等
- 可能的收获
- python requests 的使用
- python 类的使用
- python 多进程、进程池
- python 装饰器
- python 自定义异常类并在程序中手动抛出并处理
- python 进度条、输出格式个性化定制等
- 一点点软件工程设计的思想
1 Spider 模块
这个模块一开始写得时候比较简单，但是在后续加入各种各样的实例爬虫后，为了解耦和鲁棒性，功能也在不断的完善，为了阅读方便，先上简单版本代码：
# _*_ coding : utf-8 _*_
"""
定义基础 Spider 类
"""
import sys
import time
import requests
# from wrappers import old_version_fun_wrapper, req_exceed_limit_wrapper, req_respose_none_wrapper
from tqdm import tqdm
from tools import check_proxy_icanhazip, check_proxy_900cha, get_free_proxy
from concurrent.futures import ProcessPoolExecutor, as_completed
import json
from config import *
from custom_exceptions import Request500Exception, TryWithSelfProxyLimitException
class Spider:
def __init__(self, *args, **kwargs):
self.url = kwargs.get('url')
self.headers = kwargs.get('headers')
self.req_type = kwargs.get('req_type') # get or post
self.data = kwargs.get('data') # for post
self.proxies = kwargs.get('proxies')
self.verify = kwargs.get('verify')
self.day = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time())).split(' ')[0].strip() # time tag
self.timeout = 3
self.proxy_try_num = 0 # 设置使用代理时的全局失败尝试次数
self.response = None
self.parse_urls = [] # 代理资源页入口
self.all_proxies = [] # 爬取到的所有代理
self.all_proxies_filter = [] # 验证后可用的所有代理
def pre_parse(self):
"""
识别当天代理信息资源页面
"""
pass
# @req_respose_none_wrapper # 可暂时忽略
def parse(self):
"""
解析代理
"""
pass
def get_all_proxies(self):
"""
获取所有 proxies
"""
pass
def update_attrs(self, *args, **kwargs):
"""
update the spider object's attrs
:param args:
:param kwargs:
:return:
"""
self.url = self.url if kwargs.get('url') is None else kwargs.get('url')
self.headers = self.headers if kwargs.get('headers') is None else kwargs.get('headers')
self.req_type = self.req_type if kwargs.get('req_type') is None else kwargs.get('req_type')
self.data = self.data if kwargs.get('data') is None else kwargs.get('data')
self.proxies = self.proxies if kwargs.get('proxies') is None else kwargs.get('proxies')
self.verify = self.verify if kwargs.get('verify') is None else kwargs.get('verify')
self.timeout = self.timeout if kwargs.get('timeout') is None else kwargs.get('timeout')
# @req_exceed_limit_wrapper # 可暂时忽略
def update_response(self, *args, **kwargs):
"""
recontruct a url request, and update the spider's response attribute
:param args:
:param kwargs:
:return:
"""
if self.data is None:
self.response = requests.get(self.url, headers=self.headers, timeout=self.timeout, proxies=self.proxies, verify=self.verify)
else:
self.response = requests.post(self.url, headers=self.headers, data=self.data, timeout=self.timeout, proxies=self.proxies, verify=self.verify)
return self.response
def filter_all_proxies_mp(self):
"""
测试代理 ip 可用性
多进程处理
"""
self.all_proxies_filter = dict()
# 进程池
pool = ProcessPoolExecutor(max_workers=50)
all_task = [pool.submit(check_proxy_900cha, proxy) for proxy in self.all_proxies]
for future in tqdm(as_completed(all_task), total=len(all_task), file=sys.stdout, desc='[{}] checking proxies...'.format(self.__class__.__name__)):
res, proxy = future.result()
if res:
self.all_proxies_filter['{}:{}'.format(proxy['ip'], proxy['port'])] = proxy
self.all_proxies_filter = self.all_proxies_filter.values()
return self.all_proxies_filter
def save_to_txt(self, file_name, all_proxies, add_day_tag=True):
"""
存文件
"""
if not os.path.isdir(os.path.dirname(file_name)):
os.makedirs(os.path.dirname(file_name))
if add_day_tag:
file_name = file_name.split('.')[0] + '_{}.'.format(self.day.replace('-', '_')) + file_name.split('.')[-1]
with open(file_name, 'a+', encoding='utf-8') as f:
for proxy in all_proxies:
f.write(json.dumps(proxy, ensure_ascii=False) + '\n')
def run(self):
"""
General spider running logic：
init -> face page url request -> (resource page collect) -> crawl all proxies -> check proxies' useability -> save
:return:
"""
# 1 爬取所有 proxies
self.all_proxies = self.get_all_proxies()
print('[{}] 爬取代理数：{}'.format(self.__class__.__name__, len(self.all_proxies)))
# 2 过滤可用代理
self.all_proxies_filter = self.filter_all_proxies_mp()
print('[{}] 可用代理数：{}'.format(self.__class__.__name__, len(self.all_proxies_filter)))
# 3 存储可用代理
# 默认存储路径配置在 config 中，如果想要另存，在子爬虫中重构 run() 方法即可
self.save_to_txt(os.path.join(useful_ip_file_path, useful_ip_file_name), self.all_proxies_filter)
print('[{}] run successed.'.format(self.__class__.__name__))
return list(self.all_proxies_filter)
if __name__ == '__main__':
help(Spider)
上面这段代码即对应着项目中所有实例爬虫的通用运行逻辑，步骤如下：
- 初始化相关参数
- run() 实例化爬虫运行入口
- pre_parse() 请求代理资源页
- get_all_proxies() 爬取逻辑的入口函数，开始对当前实例爬虫进行代理资源爬取
- parse() 对代理详情页进行解析
- filter_all_proxies_mp() 验证爬取到的代理的可用性
- save_to_txt() 将可用代理存到文件中
为了便于理解代理资源页、代理详情页，下面以站大爷的网页结构为例进行说明：
代理资源页：可包含多个具体的代理资源集合
代理详情页：即前面每一条资源的详情页面
实际使用中，假设我们已有了一个实例爬虫 SpiderX，只需通过以下方式来启动：
spider_obj = SpiderX()
spider_obj.run()
插播一下：
上面我这里直接上了进程池的版本，本来一开始写得是串行验证，但是速度太慢了，所以果断换多进程并发。如果对 python 多进程不太熟悉，可以先停一会儿去这里看下相关知识，啪的一下很快的：https://www.cnblogs.com/zishu/p/17300868.html
1.1 Config
存储路径等参数放在 config.py 中：
# _*_ coding : utf-8 _*_
import os
RETRY_LIMIT = 4 # 爬取失败时的重试次数
# 存储文件地址
useful_ip_file_path = os.path.join('D:/FreeIPProxyGettingPro', 'proxies_spider_results')
# 存储文件名称
useful_ip_file_name = 'useful_proxies_spiding.txt'
1.2 验证代理可用性
当爬取到代理时，需要验证其可用性，对可用代理才将其保存。一般来说，较为简单的验证逻辑就是使用该代理对百度首页进行访问，然后根据返回结果验证代理可用性。但是在实际使用过程中，会出现各种问题，比如代理访问并没有隐藏掉源 ip、百度返回验证页面、无法访问但返回一个正常的说明网页（非百度首页）等。
所以这里采用的是使用代理对 IP 查询网站（https://ip.900cha.com/）进行访问，解析网页结果，判断网页显示的 ip 是否与所使用的代理 ip 一致：
代码逻辑 tools.py：
# _*_ coding : utf-8 _*_
import requests
from lxml import etree
import time
def check_proxy_900cha(proxy, timeout=3, realtimeout=False):
"""
验证代理可用性
:param proxy:
:param timeout:
:param realtimeout:
:return:
"""
time.sleep(1) # 防止频繁访问给服务器带来过大压力
url = 'https://ip.900cha.com/'
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36',
}
proxies = {
'http': '{}:{}'.format(proxy['ip'], proxy['port']),
'https': '{}:{}'.format(proxy['ip'], proxy['port'])
}
try:
response = requests.get(url=url, headers=headers, proxies=proxies, timeout=timeout)
except Exception as e:
return False, None
else:
tree = etree.HTML(response.text)
ret_ip = tree.xpath('//div[@class="col-md-8"]/h3/text()')[0].strip()
if ret_ip == proxies['http'].split(':')[0]:
if realtimeout:
print(f'代理 {proxy["ip"]}:{proxy["port"]} 有效！')
return True, proxy
else:
return False, None
2 实例爬虫
搜集网上的一些免费代理资源，限于篇幅，这里以 3 个结构典型案例来展示。
2.1 seo 代理
可以说是最简单的一个代理页面了，入口页直接就放了 proxy 列表：
实例代码：
# _*_ coding : utf-8 _*_
from tools import *
from ProxiesSpider.spider import Spider # 前面的 Spider 类
# from wrappers import req_respose_none_wrapper
class SpiderSeo(Spider):
def __init__(self, *args, **kwargs):
url = 'https://proxy.seofangfa.com/'
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36',
}
super().__init__(url=url, headers=headers)
def pre_parse(self):
self.parse_urls = [
'https://proxy.seofangfa.com/'
]
# @req_respose_none_wrapper # 可暂时不管
def parse(self):
"""
解析代理
"""
content = self.response.text
tree = etree.HTML(content)
proxies_obj = tree.xpath('//table[@class="table"]/tbody/tr')
proxies = []
for proxy_obj in proxies_obj:
dic_ = {
'ip': proxy_obj.xpath('./td[1]/text()')[0].strip(),
'port': proxy_obj.xpath('./td[2]/text()')[0].strip(),
'position': proxy_obj.xpath('./td[4]/text()')[0].strip(),
'day': proxy_obj.xpath('./td[5]/text()')[0].strip().split(' ')[0],
}
proxies.append(dic_)
return proxies
def get_all_proxies(self):
"""
获取所有 proxies
"""
self.pre_parse()
for parse_url in self.parse_urls:
self.update_attrs(url=parse_url)
self.update_response()
proxies = self.parse()
self.all_proxies += proxies
return self.all_proxies
if __name__ == '__main__':
spider_seo = SpiderSeo()
spider_seo.run()
pre_parse() 用来获取代理资源页，但是 seo 没有，为了结构一致性，在其中直接填充详情页。
2.2 快代理
快代理的网页结构介于 seo 和站大爷之间，其也没有代理资源页，但是其有两份代理资源（普通 & 高匿），所以同 seo 一样，直接在 pre_parse() 函数中填充即可。
# _*_ coding : utf-8 _*_
from ProxiesSpider.spider import Spider
from tools import *
import time
import sys
# from wrappers import req_respose_none_wrapper
class SpiderKuai(Spider):
def __init__(self, *args, **kwargs):
kwargs['url'] = 'https://www.kuaidaili.com/free/inha/1/'
kwargs['headers'] = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36',
}
super().__init__(**kwargs)
def pre_parse(self):
self.parse_urls = [
'https://www.kuaidaili.com/free/intr/', # 国内普通代理
'https://www.kuaidaili.com/free/inha/' # 国内高匿代理
]
return self.parse_urls
# @req_respose_none_wrapper
def parse(self):
"""
解析代理
"""
content = self.response.text
tree = etree.HTML(content)
proxies_obj = tree.xpath('//div[@id="list"]//tbody/tr')
proxies = []
for proxy_obj in proxies_obj:
dic_ = {
'ip': proxy_obj.xpath('./td[@data-title="IP"]/text()')[0].strip(),
'port': proxy_obj.xpath('./td[@data-title="PORT"]/text()')[0].strip(),
'type': proxy_obj.xpath('./td[@data-title="类型"]/text()')[0].strip(),
'position': proxy_obj.xpath('./td[@data-title="位置"]/text()')[0].strip(),
'day': proxy_obj.xpath('./td[@data-title="最后验证时间"]/text()')[0].strip().split(' ')[0]
}
if dic_['day'] != self.day:
break
proxies.append(dic_)
return proxies
def get_all_proxies(self):
"""
获取所有 proxies
"""
# 1 先获取所有待采集的 proxy list 页
self.pre_parse()
# 2 对每个 proxy 信息页的资源进行解析
for parse_url in self.parse_urls:
time.sleep(3)
self.update_attrs(url=parse_url)
self.update_response()
# 3 获取资源页所有的 proxy
count = 1
pbar = tqdm(file=sys.stdout, desc='[{}] crawling all pages...'.format(self.__class__.__name__))
while True:
proxies = self.parse()
if len(proxies) == 0:
break
self.all_proxies += proxies
next_page = '{}{}/'.format(parse_url, count+1)
count += 1
time.sleep(3)
self.update_attrs(url=next_page)
self.update_response()
pbar.update(1)
pbar.close()
return self.all_proxies
if __name__ == '__main__':
spider_kuai = SpiderKuai()
spider_kuai.run()
基本结构和 seo 的逻辑一致，只是这里在代理详情页解析资源时要复杂些，因为是多页结构。
观察各页资源可以发现，快代理是将所有累计的免费代理都放在一起，并没有按天分区，所以这里要面对的问题有两个：
- 翻页爬取
- 翻页过程中对资源更新日期进行检测，一但资源声明周期超过当天，就停止继续爬取
所以我们这里直接采用一个 True 循环，在 parse() 中一旦遇到生成周期超过当天的资源后，就及时返回。这样在继续翻页并且下一页没有当天资源时，就会返回空列表，此时结束循环。
在测试过程中，我发现快代理的还是有着简单的反爬限制的：
- 当连续访问多页内容时，会返回 -10，获取不到具体数据；
- 一天内多次访问时，会封 IP；
其中针对第一种，只需要在多个连续请求之间 sleep() 一下即可。而对第二种限制，由于我们的爬虫正式逻辑是一天访问一次，所以正式运行时逻辑上不会存在封禁 ip 的情况，所以可以不做处理（当然后面也可以利用已获取的代理对快代理网站进行访问）。
\(插播一下\)：
这里还有个知识点，就是 python 进度条组件 tqdm 的使用，由于 tqdm 默认的输出模式和 print 是不一样的，会导致 tqdm 输出和 print 输出排版交混的问题，所以这里需要在 tqdm 中指定 file=sys.stdout，这样就不会出现上述问题。
2.3 站大爷
相对来说结构最为完善的网页，包含代理资源页、代理详情页。所以 pre_parse() 函数中需要先对代理资源页进行解析，获取当天的代理详情页链接，再进二级页面进行爬取。
# _*_ coding : utf-8 _*_
from lxml import etree
from ProxiesSpider.spider import Spider
import time
import sys
# from wrappers import req_respose_none_wrapper
class SpiderZdaye(Spider):
"""
zdaye 自有证书，需要设置 verify=False
"""
def __init__(self, *args, **kwargs):
url = 'https://www.zdaye.com/dayProxy.html'
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36',
}
# zdaye 封禁代理比较频繁，需要使用代理去访问资源
super().__init__(url=url, headers=headers, verify=False)
def pre_parse(self):
"""
代理资源页解析
:return:
"""
self.update_response()
# 该网页需对 request 结果指定 utf-8 编码
self.response.encoding = 'utf-8'
content = self.response.text
tree = etree.HTML(content)
proxy_page_info_obj = tree.xpath('//div[@class="thread_content"]/h3/a')
for ppio in proxy_page_info_obj:
title = ppio.xpath('./text()')[0].strip().split(' ')[0]
parse_day = title.split('日')[0].replace('年', '-').replace('月', '-')
if [int(x) for x in parse_day.split('-')] == [int(x) for x in self.day.split('-')]:
self.parse_urls.append(ppio.xpath('./@href')[0])
else:
break
return self.parse_urls
# @req_respose_none_wrapper
def parse(self):
"""
解析代理
"""
self.response.encoding = 'utf-8'
content = self.response.text
tree = etree.HTML(content)
proxies_obj = tree.xpath('//table[@id="ipc"]/tbody/tr')
proxies = []
for proxy_obj in proxies_obj:
dic_ = {
'ip': proxy_obj.xpath('./td[1]/text()')[0].strip().replace('"', '').strip(),
'port': proxy_obj.xpath('./td[2]/text()')[0].strip().replace('"', '').strip(),
'type': proxy_obj.xpath('./td[3]/text()')[0].strip(),
'position': proxy_obj.xpath('./td[5]/text()')[0].strip().split(' ')[0],
'isp': proxy_obj.xpath('./td[5]/text()')[0].strip().split(' ')[1] if len(proxy_obj.xpath('./td[5]/text()')[0].strip().split(' ')) > 1 else None,
'day': self.day
}
proxies.append(dic_)
return proxies
def get_all_proxies(self):
"""
获取所有 proxies
"""
# 1 先获取所有代理详情页的 url
self.pre_parse()
# 2 对每个 proxy 信息页的资源进行解析
for parse_url in self.parse_urls:
parse_url = 'https://www.zdaye.com' + parse_url
self.update_attrs(url=parse_url)
self.update_response()
# 3 获取详情页所有的 proxy
while True:
time.sleep(3) # 间隔爬取
proxies = self.parse()
if len(proxies) == 0:
break
self.all_proxies += proxies
next_tag = etree.HTML(self.response.text).xpath('//a[@title="下一页"]/@href')
if len(next_tag) == 0:
break
else:
next_page = 'https://www.zdaye.com' + etree.HTML(self.response.text).xpath('//a[@title="下一页"]/@href')[0]
self.update_attrs(url=next_page)
self.update_response()
return self.all_proxies
if __name__ == '__main__':
spider_zdy = SpiderZdaye()
spider_zdy.run()
站大爷的网站需要注意的主要有两点：
- 设置 verify=False 来关闭 SSL 验证；
- 该网站的反爬措施还蛮严厉的，上述代码中尽管对请求进行了简单的 sleep，但是只要多调试几次，还是会被封；
关于反爬，后面会讲解如何使用已爬取的代理来请求，并且站大爷这个网站还挺难搞的，后面会说到的。
3 爬虫链 & 使用代理绕过反爬
其实写到这里，如果要求不高的话，上面的功能已经可以做基本使用了，只需要挨个运行或者直接写一个主函数依次实例化并运行即可。但是上面已经说到了有些网站会有反爬限制，如果不使用代理，这个实例爬虫基本上就废掉了，所以下面我将多个实例爬虫排排队，串行执行，并在此过程中更新已有可用代理，并编写代理请求接口函数随机的获取一个可用代理，用来应对被限制的情况。
3.1 加载已有代理资源
看过 Spider 类代码的应该有印象，该项目将爬取并验证后的代理以字典的形式存到 txt 文件中，形式如下：
{"ip": "222.190.208.49", "port": "8089", "position": "江苏省泰州市", "isp": "电信", "day": "2023-04-13"}
{"ip": "36.137.106.110", "port": "7890", "position": "北京市", "isp": "移动", "day": "2023-04-13"}
{"ip": "182.241.132.30", "port": "80", "position": "云南省红河州", "isp": "电信", "day": "2023-04-13"}
并且多个实例爬虫是采用追加的形式向同一份文件追加写入的，那么我们自然可以简单的通过加载文件的方式来获取已有资源：
# 项目文件结构
-- proxies_spider_results:
-- useful_proxies_2023-04-11.txt
-- useful_proxies_2023-04-12.txt
-- ProxiesSpider:
-- seo_spider.py
-- kuai_spider.py
-- zdaye_spider.py
-- main.py
-- tools.py
import sys
import json
import random
import os
from config import useful_ip_file_path
here = os.path.dirname(__file__)
def get_latest_proxy_file(file_path):
"""
获取当前路径下的最新文件内容
:param file_path:
:return:
"""
file_latest = sorted(os.listdir(file_path))[-1]
with open(os.path.join(file_path, file_latest), 'r', encoding='utf=8') as f:
all_free_proxies = [json.loads(s.strip()) for s in f.readlines()]
return all_free_proxies
def get_free_proxy():
all_free_proxies = get_latest_proxy_file(useful_ip_file_path)
for i in tqdm(range(len(all_free_proxies)), file=sys.stdout, desc='choosing a useful proxy...'):
index = random.randint(0, len(all_free_proxies)-1)
proxy = all_free_proxies[index]
useful, proxy = check_proxy_900cha(proxy)
if useful:
return proxy
print('无可用 proxy ~')
return None
if __name__ == '__main__':
print(get_free_proxy())
这样，通过 get_free_proxy()函数，就可以很容易的从已有代理中随机挑选一个可用的代理。
3.2 使用代理
有了上面的代理获取函数，在遇到反爬时，我们只需要调用一下，如果能返回一个可用的代理，那么就可以拿着这个代理去重新请求网页资源。
这里我以 kuai spider 为例，我们可以设置超时时间为 0.01 s，来模拟访问失败的情况，然后 catch 这个 Exception 并重新使用代理访问正确的网页：
# _*_ coding : utf-8 _*_
# @Time : 2023/3/13 14:19
# @Author : jiang
# @File : kuai_proxy_spider
# Project : FreeIPProxyGettingPro
# 将上级目录加载进来
import sys
import os
sys.path.append(os.path.dirname(__name__))
from ProxiesSpider.spider import Spider
from lxml import etree
import time
import sys
from wrappers import req_respose_none_wrapper
from tqdm import tqdm
from tools import get_free_proxy
class SpiderKuai(Spider):
def __init__(self, *args, **kwargs):
kwargs['url'] = 'https://www.kuaidaili.com/free/inha/1/'
kwargs['headers'] = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36',
}
super().__init__(**kwargs)
def pre_parse(self):
self.parse_urls = [
'https://www.kuaidaili.com/free/intr/', # 国内普通代理
'https://www.kuaidaili.com/free/inha/' # 国内高匿代理
]
return self.parse_urls
@req_respose_none_wrapper
def parse(self):
"""
解析代理
"""
content = self.response.text
tree = etree.HTML(content)
proxies_obj = tree.xpath('//div[@id="list"]//tbody/tr')
proxies = []
for proxy_obj in proxies_obj:
dic_ = {
'ip': proxy_obj.xpath('./td[@data-title="IP"]/text()')[0].strip(),
'port': proxy_obj.xpath('./td[@data-title="PORT"]/text()')[0].strip(),
'type': proxy_obj.xpath('./td[@data-title="类型"]/text()')[0].strip(),
'position': proxy_obj.xpath('./td[@data-title="位置"]/text()')[0].strip(),
'day': proxy_obj.xpath('./td[@data-title="最后验证时间"]/text()')[0].strip().split(' ')[0]
}
if dic_['day'] != self.day:
break
proxies.append(dic_)
return proxies
def get_all_proxies(self):
"""
获取所有 proxies
"""
# 1 先获取所有待采集的 proxy list 页
self.pre_parse()
# 2 对每个 proxy 信息页的资源进行解析
for parse_url in self.parse_urls:
time.sleep(3)
self.update_attrs(url=parse_url)
self.update_response()
# 3 获取资源页所有的 proxy
count = 1
pbar = tqdm(file=sys.stdout, desc='[{}] crawling all pages...'.format(self.__class__.__name__))
while True:
proxies = self.parse()
if len(proxies) == 0:
break
self.all_proxies += proxies
next_page = '{}{}/'.format(parse_url, count+1)
count += 1
time.sleep(3)
self.update_attrs(url=next_page)
self.update_response()
pbar.update(1)
pbar.close()
return self.all_proxies
if __name__ == '__main__':
spider_kuai = SpiderKuai()
spider_kuai.timeout = 0.01
try:
spider_kuai.run()
except Exception as e:
print(e)
# 使用代理
proxy = get_free_proxy()
print('使用代理：', proxy)
if proxy is None:
proxies = None
else:
proxies = {
'http': '{}:{}'.format(proxy['ip'], proxy['port']),
'https': '{}:{}'.format(proxy['ip'], proxy['port']),
}
spider_kuai.timeout = 3
spider_kuai.run()
执行结果：
~\python_spider\FreeIPProxyGettingPro_TMP> kuai_proxy_spider.py
HTTPSConnectionPool(host='www.kuaidaili.com', port=443): Max retries exceeded with url: /free/intr/ (Caused by ConnectTimeoutError(<urllib3.connection.HTTPSConnection object at 0x00000183962DFB50>, 'Connection to www.kuaidaili.com timed out. (connect timeout=0.01)'))
choosing a useful proxy...: 6%|█████▊ | 1/17 [00:06<01:38, 6.13s/it]
使用代理： {'ip': '182.241.132.30', 'port': '80', 'position': '云南省红河州', 'isp': '电信', 'day': '2023-04-13'}
[SpiderKuai] crawling all pages...: 2it [00:07, 3.72s/it]
[SpiderKuai] crawling all pages...: 2it [00:07, 3.69s/it]
[SpiderKuai] 爬取代理数：42
[SpiderKuai] checking proxies...: 100%|████████████████████████████████████████████████████████████████████████████████████████████| 42/42 [00:07<00:00, 5.48it/s]
[SpiderKuai] 可用代理数：0
[SpiderKuai] run successed.
可以看到，基本逻辑没问题。
4 鲁棒性
上一节介绍了如何使用爬取到的代理访问资源页，但是处理方式还是不够优雅。
我们可以想象一下，真实的爬虫运行出错场景是怎样的，首先肯定一开始是正常请求，结果报错了，此时我们应该暂停一下再次访问（确保不是网络本身的问题），如果还是不行就通过 get_free_proxy()获取一个代理来进行请求，当然也不是无限次请求，我们暂且设置这种尝试次数不超过三次，之后如果还是报错，则停止这个爬虫。
4.1 request 请求出错处理逻辑
好了，有了这个基础的逻辑，我们就可以愉快的写代码了。首先，为了解耦，我们肯定是不会在每个实例爬虫中像上一节中那样进行 try catch 的，别忘了 Spider 这个类，对 url 进行 requset 请求操作被封装在了 update_response() 这个函数中，所以我们可以直接对该函数进行 try catch。
未完待续：在后面，会有，不会鸽~