如何构建Python爬虫的代理池搭建流程？

本文共计1900个文字，预计阅读时间需要8分钟。

一、为什么要搭建爬虫代理？在众多网站防爬措施中，有一种是基于IP的访问频率限制。即在某一时间段内，当某个IP的访问次数达到一定阈值时，该IP就会被拉黑。

一、为什么要搭建爬虫代理池

在众多的网站防爬措施中，有一种是根据ip的访问频率进行限制，即在某一时间段内，当某个ip的访问次数达到一定的阀值时，该ip就会被拉黑、在一段时间内禁止访问。

应对的方法有两种：

1. 降低爬虫的爬取频率，避免IP被限制访问，缺点显而易见：会大大降低爬取的效率。

2. 搭建一个IP代理池，使用不同的IP轮流进行爬取。

二、搭建思路

1、从代理网站(如：西刺代理、快代理、云代理、无忧代理)爬取代理IP；

2、验证代理IP的可用性（使用代理IP去请求指定URL，根据响应验证代理IP是否生效）；

3、将可用的代理IP保存到数据库；

常用代理网站：西刺代理、云代理、IP海、无忧代理、飞蚁代理、快代理

三、代码实现

工程结构如下：

ipproxy.py

IPProxy代理类定义了要爬取的IP代理的字段信息和一些基础方法。

# -*- coding: utf-8 -*- import re import time from settings import PROXY_URL_FORMATTER schema_pattern = re.compile(r'%(ip)s:%(port)s' USER_AGENT_LIST = [ "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.1 (KHTML, like Gecko) Chrome/22.0.1207.1 Safari/537.1", "Mozilla/5.0 (X11; CrOS i686 2268.111.0) AppleWebKit/536.11 (KHTML, like Gecko) Chrome/20.0.1132.57 Safari/536.11", "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/536.6 (KHTML, like Gecko) Chrome/20.0.1092.0 Safari/536.6", "Mozilla/5.0 (Windows NT 6.2) AppleWebKit/536.6 (KHTML, like Gecko) Chrome/20.0.1090.0 Safari/536.6", "Mozilla/5.0 (Windows NT 6.2; WOW64) AppleWebKit/537.1 (KHTML, like Gecko) Chrome/19.77.34.5 Safari/537.1", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/536.5 (KHTML, like Gecko) Chrome/19.0.1084.9 Safari/536.5", "Mozilla/5.0 (Windows NT 6.0) AppleWebKit/536.5 (KHTML, like Gecko) Chrome/19.0.1084.36 Safari/536.5", "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1063.0 Safari/536.3", "Mozilla/5.0 (Windows NT 5.1) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1063.0 Safari/536.3", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_8_0) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1063.0 Safari/536.3", "Mozilla/5.0 (Windows NT 6.2) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1062.0 Safari/536.3", "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1062.0 Safari/536.3", "Mozilla/5.0 (Windows NT 6.2) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1061.1 Safari/536.3", "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1061.1 Safari/536.3", "Mozilla/5.0 (Windows NT 6.1) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1061.1 Safari/536.3", "Mozilla/5.0 (Windows NT 6.2) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1061.0 Safari/536.3", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/535.24 (KHTML, like Gecko) Chrome/19.0.1055.1 Safari/535.24", "Mozilla/5.0 (Windows NT 6.2; WOW64) AppleWebKit/535.24 (KHTML, like Gecko) Chrome/19.0.1055.1 Safari/535.24" ] # 爬取到的代理保存前先检验是否可用，默认True PROXY_CHECK_BEFOREADD = True # 检验代理可用性的请求地址，支持多个 PROXY_CHECK_URLS = {'icanhazip.com'],'icanhazip.com']}

proxy_util.py

proxy_util.py 中主要定义了一些实用方法，例如：proxy_to_dict(proxy)用来将IPProxy代理实例转换成字典；proxy_from_dict(d)用来将字典转换为IPProxy实例；request_page()用来发送请求；_is_proxy_available()用来校验代理IP是否可用。

# -*- coding: utf-8 -*- import random import logging import requests from ipproxy import IPProxy from settings import USER_AGENT_LIST, PROXY_CHECK_URLS # Setting logger output format logging.basicConfig(level=logging.INFO, format='[%(asctime)-15s] [%(levelname)8s] [%(name)10s ] - %(message)s (%(filename)s:%(lineno)s)', datefmt='%Y-%m-%d %T' ) logger = logging.getLogger(__name__) def proxy_to_dict(proxy): d = { "schema": proxy.schema, "ip": proxy.ip, "port": proxy.port, "used_total": proxy.used_total, "success_times": proxy.success_times, "continuous_failed": proxy.continuous_failed, "created_time": proxy.created_time } return d def proxy_from_dict(d): return IPProxy(schema=d['schema'], ip=d['ip'], port=d['port'], used_total=d['used_total'], success_times=d['success_times'], continuous_failed=d['continuous_failed'], created_time=d['created_time']) # Truncate header and tailer blanks def strip(data): if data is not None: return data.strip() return data base_headers = { 'Accept-Encoding': 'gzip, deflate, br', 'Accept-Language': 'en-US,en;q=0.9,zh-CN;q=0.8,zh;q=0.7' } def request_page(url, options={}, encoding='utf-8'): """send a request,get response""" headers = dict(base_headers, **options) if 'User-Agent' not in headers.keys(): headers['User-Agent'] = random.choice(USER_AGENT_LIST) logger.info('正在抓取: ' + url) try: response = requests.get(url, headers=headers) if response.status_code == 200: logger.info('抓取成功: ' + url) return response.content.decode(encoding=encoding) except ConnectionError: logger.error('抓取失败' + url) return None def _is_proxy_available(proxy, options={}): """Check whether the Proxy is available or not""" headers = dict(base_headers, **options) if 'User-Agent' not in headers.keys(): headers['User-Agent'] = random.choice(USER_AGENT_LIST) proxies = {proxy.schema: proxy._get_url()} check_urls = PROXY_CHECK_URLS[proxy.schema] for url in check_urls: try: response = requests.get(url=url, proxies=proxies, headers=headers, timeout=5) except BaseException: logger.info("< " + url + " > 验证代理 < " + proxy._get_url() + " > 结果： 不可用 ") else: if response.status_code == 200: logger.info("< " + url + " > 验证代理 < " + proxy._get_url() + " > 结果： 可用 ") return True else: logger.info("< " + url + " > 验证代理 < " + proxy._get_url() + " > 结果： 不可用 ") return False if __name__ == '__main__': headers = dict(base_headers) if 'User-Agent' not in headers.keys(): headers['User-Agent'] = random.choice(USER_AGENT_LIST) proxies = {"163.125.255.154:9797"} response = requests.get("www.baidu.com", headers=headers, proxies=proxies, timeout=3) print(response.content)

proxy_queue.py

代理队列用来保存并对外提供 IP代理，不同的代理队列内代理IP的保存和提取策略可以不同。在这里， BaseQueue 是所有代理队列的基类，其中声明了所有代理队列都需要实现的保存代理IP、提取代理IP、查看代理IP数量等接口。示例的 FifoQueue 是一个先进先出队列，底层使用 Redis 列表实现，为了确保同一个代理IP只能被放入队列一次，这里使用了一个Redis proxies::existed 集合进行入队前重复校验。

# -*- coding: utf-8 -*- from proxy_util import logger import json import redis from ipproxy import IPProxy from proxy_util import proxy_to_dict, proxy_from_dict, _is_proxy_available from settings import PROXIES_REDIS_EXISTED, PROXIES_REDIS_FORMATTER, MAX_CONTINUOUS_TIMES, PROXY_CHECK_BEFOREADD """ Proxy Queue Base Class """ class BaseQueue(object): def __init__(self, server): """Initialize the proxy queue instance Parameters ---------- server : StrictRedis Redis client instance """ self.server = server def _serialize_proxy(self, proxy): """Serialize proxy instance""" return proxy_to_dict(proxy) def _deserialize_proxy(self, serialized_proxy): """deserialize proxy instance""" return proxy_from_dict(eval(serialized_proxy)) def __len__(self, schema='www.kuaidaili.com/free/inha/{}/', 'type': '国内高匿', 'page': 1}, {'url': 'www.kuaidaili.com/free/intr/{}/', 'type': '国内普通', 'page': 1}]) kuaidailiCrawler._start_crawl() def run_feiyi(): feiyidailiCrawler = FeiyiDailiCrawler(queue=fifo_queue, website='飞蚁代理', urls=[{'url': 'www.feiyiproxy.com/?page_id=1457', 'type': '首页推荐'}]) feiyidailiCrawler._start_crawl() def run_wuyou(): wuyoudailiCrawler = WuyouDailiCrawler(queue=fifo_queue, website='无忧代理', urls=[{'url': 'www.data5u.com/free/index.html', 'type': '首页推荐'}, {'url': 'www.data5u.com/free/gngn/index.shtml', 'type': '国内高匿'}, {'url': 'www.data5u.com/free/gnpt/index.shtml', 'type': '国内普通'}]) wuyoudailiCrawler._start_crawl() def run_iphai(): crawler = IPhaiDailiCrawler(queue=fifo_queue, website='IP海代理', urls=[{'url': 'www.iphai.com/free/ng', 'type': '国内高匿'}, {'url': 'www.iphai.com/free/np', 'type': '国内普通'}, {'url': 'www.iphai.com/free/wg', 'type': '国外高匿'}, {'url': 'www.iphai.com/free/wp', 'type': '国外普通'}]) crawler._start_crawl() def run_yun(): crawler = YunDailiCrawler(queue=fifo_queue, website='云代理', urls=[{'url': 'www.ip3366.net/free/?stype=1&page={}', 'type': '国内高匿', 'page': 1}, {'url': 'www.ip3366.net/free/?stype=2&page={}', 'type': '国内普通', 'page': 1}, {'url': 'www.ip3366.net/free/?stype=3&page={}', 'type': '国外高匿', 'page': 1}, {'url': 'www.ip3366.net/free/?stype=4&page={}', 'type': '国外普通', 'page': 1}]) crawler._start_crawl() def run_xici(): crawler = XiCiDailiCrawler(queue=fifo_queue, website='西刺代理', urls=[{'url': 'www.xicidaili.com/', 'type': '首页推荐'}, {'url': 'www.xicidaili.com/nn/{}', 'type': '国内高匿', 'page': 1}, {'url': 'www.xicidaili.com/nt/{}', 'type': '国内普通', 'page': 1}, {'url': 'www.xicidaili.com/wn/{}', 'type': '国外高匿', 'page': 1}, {'url': 'www.xicidaili.com/wt/{}', 'type': '国外普通', 'page': 1}]) crawler._start_crawl() if __name__ == '__main__': run_xici() run_iphai() run_kuai() run_feiyi() run_yun() run_wuyou()

爬取西刺代理时，后台日志示例如下：

Redis数据库中爬取到的代理IP的数据结构如下:

四、代理测试

接下来，使用爬取好的代理来请求 icanhazip.com进行测试，代码如下：

# -*- coding: utf-8 -*- import random import requests from proxy_util import logger from run import fifo_queue from settings import USER_AGENT_LIST from proxy_util import base_headers # 测试地址 url = 'icanhazip.com' # 获取代理 proxy = fifo_queue.pop(schema='218.66.253.144:80代理请求成功后将代理重新放回队列，并将 Redis 中该代理的used_total 、success_times 、continuous_failed三个字段信息进行了相应的更新。

项目地址：github.com/pengjunlee/ipproxy_pool.git

到此这篇关于Python爬虫代理池搭建的方法步骤的文章就介绍到这了,更多相关Python爬虫代理池搭建内容请搜索易盾网络以前的文章或继续浏览下面的相关文章希望大家以后多多支持易盾网络！