python代理池搭建

熟悉爬虫的，必定会熟悉各种反爬机制。今天就讲一下自己如何建立ip代理池的。

一个合格的代理池必须拥有一个爬取代理IP的爬取器、一个验证IP可否使用的校验器、一个存储IP的数据库、调用这些的调度器以及可以供获取IP的接口（这里推荐flask，比较简单）。

先来说说爬取器，首先要爬取的代理IP网站尽量是无需登录的，其次是对代理IP更新较快的，前者加快代理池的效率，后者增加代理池的质量。这里我对市面上部分代理网站进行爬取，当然一些常用的代理IP网站提供IP质量不高，比如西刺无忧66这些经常被爬取（西刺偶尔还会崩溃，估计是爬取的人有些多，网站保护性503）

    def crawl_xici():"""西刺代理：http://www.xicidaili.com"""url = "http://www.xicidaili.com/{}"items = []for page in range(1, 21):items.append(("wt/{}".format(page), "http://{}:{}"))items.append(("wn/{}".format(page), "https://{}:{}"))for item in items:proxy_type, host = itemhtml = requests(url.format(proxy_type))if html:doc = pyquery.PyQuery(html)for proxy in doc("table tr").items():ip = proxy("td:nth-child(2)").text()port = proxy("td:nth-child(3)").text()if ip and port:yield host.format(ip, port)def crawl_zhandaye():"""站大爷代理：http://ip.zdaye.com/dayProxy.html"""url = 'http://ip.zdaye.com/dayProxy.html'html = requests(url)sttrs = re.findall('<H3 class="title"><a href="(.*?)">', html, re.S)for sttr in sttrs:new_url = url[:28] + sttr[9:]new_html = requests_other(new_url)get_div = re.search("<div class=\"cont\">(.*?)</div>", new_html, re.S).group(1)print(get_div)results = re.findall("<br>(.*?)@(.*?)#\[(.*?)\]", get_div, re.S)for result in results:yield "{}://{}".format(result[1].lower(), result[0])def crawl_66ip():"""66ip 代理：http://www.66ip.cn19-04-30可用"""url = ("http://www.66ip.cn/nmtq.php?getnum=100&isp=0""&anonymoustype=0&area=0&proxytype={}&api=66ip")pattern = "\d+\.\d+.\d+\.\d+:\d+"items = [(0, "http://{}"), (1, "https://{}")]for item in items:proxy_type, host = itemhtml = requests(url.format(proxy_type))if html:for proxy in re.findall(pattern, html):yield host.format(proxy)def crawl_kuaidaili():"""快代理：https://www.kuaidaili.com每次30个19-04-13可用"""url = "https://www.kuaidaili.com/free/inha/{}/"items = [p for p in range(1, 3)]for page in items:html = requests(url.format(page))if html:doc = pyquery.PyQuery(html)for proxy in doc(".table-bordered tr").items():ip = proxy("[data-title=IP]").text()port = proxy("[data-title=PORT]").text()if ip and port:yield "http://{}:{}".format(ip, port)def crawl_ip3366():"""云代理：http://www.ip3366.net每页10个，验证较快19-04-30可用"""url = "http://www.ip3366.net/?stype=1&page={}"items = [p for p in range(1, 8)]for page in items:html = requests(url.format(page))if html:doc = pyquery.PyQuery(html)for proxy in doc(".table-bordered tr").items():ip = proxy("td:nth-child(1)").text()port = proxy("td:nth-child(2)").text()schema = proxy("td:nth-child(4)").text()if ip and port and schema:yield "{}://{}:{}".format(schema.lower(), ip, port)def crawl_data5u():"""无忧代理：http://www.data5u.com/每次14个，验证时间比较新19-04-30可用"""url = "http://www.data5u.com/free/index.html"html = requests(url)if html:doc = pyquery.PyQuery(html)for index, item in enumerate(doc(".wlist li .l2").items()):if index > 0:ip = item("span:nth-child(1)").text()port = item("span:nth-child(2)").text()schema = item("span:nth-child(4)").text()if ip and port and schema:yield "{}://{}:{}".format(schema, ip, port)def crawl_iphai():"""ip 海代理：http://www.iphai.com爬取国内高匿、国外高匿、国外普通各10个19-04-30可用"""url = "http://www.iphai.com/free/{}"items = ["ng", "np", "wg", "wp"]for proxy_type in items:html = requests(url.format(proxy_type))if html:doc = pyquery.PyQuery(html)for item in doc(".table-bordered tr").items():ip = item("td:nth-child(1)").text()port = item("td:nth-child(2)").text()schema = item("td:nth-child(4)").text().split(",")[0]if ip and port and schema:yield "{}://{}:{}".format(schema.lower(), ip, port)

解释一下代码，返回的代理一般都是(http/https)://ip:port格式的代理。这里的requests是使用 asyncio、aiohttp做了个方法，来实现异步爬取。对于asyncio的介绍可以查看廖雪峰的教程。这个是自己写的异步爬取，替换了之前的request.get方法，加快爬虫效率。

import asyncio
import aiohttpfrom settings import HEADERS, REQUEST_TIMEOUT, REQUEST_DELAYLOOP = asyncio.get_event_loop()async def _get_page(url, sleep):"""获取并返回网页内容"""async with aiohttp.ClientSession() as session:try:await asyncio.sleep(sleep)async with session.get(url, headers=HEADERS, timeout=REQUEST_TIMEOUT) as resp:return await resp.text()except:return ""def requests(url, sleep=REQUEST_DELAY):"""请求方法，用于获取网页内容:param url: 请求链接:param sleep: 延迟时间（秒）"""html = LOOP.run_until_complete(asyncio.gather(_get_page(url, sleep)))if html:return "".join(html)

做好异步爬取工作就应该完成代理池搭建的30%了，之后我们就要尝试保存进数据库了，这里推荐使用redis数据库，毕竟其中的有序集合类型（sorted set）非常适合代理池cookies池的搭建，因为其中是有score的，也就是我们存入一个代理的同时也要给它一个分数，这方便我们之后对其校验以及取代理IP的优先级。redis有序集合类型

# redis 地址
REDIS_HOST = "localhost"
# redis 端口
REDIS_PORT = 6379
# redis 密码
REDIS_PASSWORD = None
# redis set key
REDIS_KEY = "myproxies"
# redis 连接池最大连接量
REDIS_MAX_CONNECTION = 20
# REDIS SCORE 最大分数
MAX_SCORE = 10
# REDIS SCORE 最小分数
MIN_SCORE = 0
# REDIS SCORE 初始分数
INIT_SCORE = 5class RedisClient:"""代理池依赖了 Redis 数据库，使用了其`有序集合`的数据结构（可按分数排序，key 值不能重复）"""def __init__(self, host=REDIS_HOST, port=REDIS_PORT, password=REDIS_PASSWORD):conn_pool = redis.ConnectionPool(host=host,port=port,password=password,max_connections=REDIS_MAX_CONNECTION,)self.redis = redis.Redis(connection_pool=conn_pool)def add_proxy(self, proxy, score=INIT_SCORE):"""新增一个代理，初始化分数 INIT_SCORE < MAX_SCORE，确保在运行完收集器后还没运行校验器就获取代理，导致获取到分数虽为 MAX_SCORE,但实际上确是未经验证，不可用的代理:param proxy: 新增代理:param score: 初始化分数"""if not self.redis.zscore(REDIS_KEY, proxy):self.redis.zadd(REDIS_KEY, proxy, score)def reduce_proxy_score(self, proxy):"""验证未通过，分数减一:param proxy: 验证代理"""score = self.redis.zscore(REDIS_KEY, proxy)if score and score > MIN_SCORE:self.redis.zincrby(REDIS_KEY, proxy, -1)else:self.redis.zrem(REDIS_KEY, proxy)def increase_proxy_score(self, proxy):"""验证通过，分数加一:param proxy: 验证代理"""score = self.redis.zscore(REDIS_KEY, proxy)if score and score < MAX_SCORE:self.redis.zincrby(REDIS_KEY, proxy, 1)def pop_proxy(self):"""返回一个代理"""# 第一次尝试取分数最高，也就是最新可用的代理first_chance = self.redis.zrangebyscore(REDIS_KEY, MAX_SCORE, MAX_SCORE)if first_chance:return random.choice(first_chance)else:# 第二次尝试取 7-10 分数的任意一个代理second_chance = self.redis.zrangebyscore(REDIS_KEY, MAX_SCORE - 3, MAX_SCORE)if second_chance:return random.choice(second_chance)# 最后一次就随便取咯else:last_chance = self.redis.zrangebyscore(REDIS_KEY, MIN_SCORE, MAX_SCORE)if last_chance:return random.choice(last_chance)def get_proxies(self, count=1):"""返回指定数量代理，分数由高到低排序:param count: 代理数量"""proxies = self.redis.zrevrange(REDIS_KEY, 0, count - 1)for proxy in proxies:yield proxy.decode("utf-8")def count_all_proxies(self):"""返回所有代理总数"""return self.redis.zcard(REDIS_KEY)def count_score_proxies(self, score):"""返回指定分数代理总数:param score: 代理分数"""if 0 <= score <= 10:proxies = self.redis.zrangebyscore(REDIS_KEY, score, score)return len(proxies)return -1def clear_proxies(self, score):"""删除分数小于等于 score 的代理"""if 0 <= score <= 10:proxies = self.redis.zrangebyscore(REDIS_KEY, 0, score)for proxy in proxies:self.redis.zrem(REDIS_KEY, proxy)return Truereturn Falsedef all_proxies(self):"""返回全部代理"""return self.redis.zrangebyscore(REDIS_KEY, MIN_SCORE, MAX_SCORE)

这就是写的对redis数据库的操作，至于我们如何把爬取的IP放入，就需要改一下我们的爬取器了，封装成一个类会比我写这样的方法好得多。

redis_conn = RedisClient()
all_funcs = []def collect_funcs(func):"""装饰器，用于收集爬虫函数"""all_funcs.append(func)return funcclass Crawler:"""返回格式： http://host:port"""@staticmethoddef run():"""启动收集器"""for func in all_funcs:for proxy in func():redis_conn.add_proxy(proxy)#添加之前写的爬取ip方法，在每个方法之前声明装饰器@collect_funcs#然后实例个对象
crawl = Crawl()

完成了数据库和爬取器的搭建，就差不多完成七七八八了，现在就加验证器，验证器是为了验证代理IP是否可用，如何可以使用的话就给它加一分，如果不可以使用的话就减一分

import os
import asyncioimport aiohttpfrom db import RedisClient#验证url
VALIDATOR_BASE_URL = "http://baidu.com"
#批量测试数量
VALIDATOR_BATCH_COUNT = 250class Validator:def __init__(self):self.redis = RedisClient()async def test_proxy(self, proxy):"""测试代理:param proxy: 指定代理"""async with aiohttp.ClientSession() as session:try:if isinstance(proxy, bytes):proxy = proxy.decode("utf8")async with session.get(VALIDATOR_BASE_URL, proxy=proxy, timeout=REQUEST_TIMEOUT) as resp:if resp.status == 200:self.redis.increase_proxy_score(proxy)else:self.redis.reduce_proxy_score(proxy)except:self.redis.reduce_proxy_score(proxy)def run(self):"""启动校验器"""proxies = self.redis.all_proxies()loop = asyncio.get_event_loop()for i in range(0, len(proxies), VALIDATOR_BATCH_COUNT):_proxies = proxies[i : i + VALIDATOR_BATCH_COUNT]tasks = [self.test_proxy(proxy) for proxy in _proxies]if tasks:loop.run_until_complete(asyncio.wait(tasks))validator = Validator()

网上有许多检验IP的方法，诸如requests.get telnet之类的，这里利用的是aiohttp的session，其实这个检验都差不多的，只是在爬取IP那边利用的异步，干脆检验这里也用异步吧。

之后写个调度器，调度器就是运行项目后，在每隔一个时间段后运行某个方法，这里我们设置循环时间，爬取是30分钟，检验是15分钟，然后启动爬取器和校验器的run方法

import timeimport schedulefrom crawler import crawler
from validator import validator#爬取ip检查时间（分）
CRAWLER_RUN_CYCLE = 30#验证ip检查时间（分）
VALIDATOR_RUN_CYCLE = 15def run_schedule():"""启动客户端"""# 启动收集器schedule.every(CRAWLER_RUN_CYCLE).minutes.do(crawler.run).run()# 启动验证器schedule.every(VALIDATOR_RUN_CYCLE).minutes.do(validator.run).run()while True:try:schedule.run_pending()time.sleep(1)except KeyboardInterrupt:return

最后就是整个项目如何运行了，我们单纯跑调度器肯定是不合适的，因为这样redis中有代理了，但是我们要在爬虫项目中连接redis来获取代理，这一点是比较麻烦的，而且每次运行都要在本地跑一次代理池，这样肯定不符合程序员偷懒的初衷。正确的做法是做一个api接口，然后部署到云端，让云服务器爬取代理IP，这样我们就每次访问api接口就能得到我们的数据了，下面来写写接口，这里用的是Flask，因为简单

from flask import Flask, jsonify
from db import RedisClient
from scheduler import run_schedulemyapp = Flask(__name__)
redis_conn = RedisClient()@myapp.route("/")
def index():return jsonify({"Welcome": "This is a proxy pool system."},{"if there has problem": "Please communicate with QQ:976264593"})@myapp.route("/pop")
def pop_proxy():proxy = redis_conn.pop_proxy().decode("utf8")if proxy[:5] == "https":return jsonify({"https": proxy})else:return jsonify({"http": proxy})@myapp.route("/get/<int:count>")
def get_proxy(count):res = []for proxy in redis_conn.get_proxies(count):if proxy[:5] == "https":res.append({"https": proxy})else:res.append({"http": proxy})return jsonify(res)@myapp.route("/count")
def count_all_proxies():count = redis_conn.count_all_proxies()return jsonify({"count": str(count)})@myapp.route("/count/<int:score>")
def count_score_proxies(score):count = redis_conn.count_score_proxies(score)return jsonify({"count": str(count)})@myapp.route("/clear/<int:score>")
def clear_proxies(score):if redis_conn.clear_proxies(score):return jsonify({"Clear": "Successful"})return jsonify({"Clear": "Score should >= 0 and <= 10"})if __name__ == "__main__":# 启动服务端 Flask appmyapp.run(host='localhost', port=5000, debug=True)run_schedule()

这个接口比较简单，实现了查看代理池数量，获取代理IP（推荐pop因为使用后就可以将代理IP删除）当然有许多低分的代理我们也可以通过接口调用将其删除。出于安全性，其实这个api应该加个校验的，不然别人一直用你的代理池就白费功夫了。

好了，这就是python代理池搭建，如果有不懂可以提出来，博主也是小白一个

源码链接

python代理池搭建相关推荐

技术分享：Proxy-Pool代理池搭建IP代理
提示:本章内容仅供参考,不涉及实际使用. Proxy Pool代理池搭建前言一.环境准备二.搭建过程 1.Redis数据库搭建 2.Proxy-Pool代理池搭建总结前言本章内容仅供参考, ...
python代理池_用Python搭建一个简单的代理池
其实每次爬东西的时候,特怕IP被封,所以每次都要把时间延迟设置得长一点...这次用Python搭建一个简单的代理池.获取代理IP,然后验证其有效性.不过结果好像不是很理想,为什么西刺代理的高匿代理都能 ...
python代理池好难啊_新人不会自己搭建代理池?快来引用大佬的
新人不会自己搭建代理池?快来引用大佬的对于新人学习爬虫来说,虽然不会爬取太难的网站,但是有时候爬取的数据量过大的时候,也会遇到返回不了数据的问题,这时候打开网页一看.可能会看到"你的ip访 ...
python代理池_进击：用Python搭建高匿代理池
Email: WlRCME0zSjRRR2R0WVdsc0xtTnZiUT09 0x01 写在前面常听到很多人抱怨自己的IP因爬虫次数太多而被网站屏蔽,不得不频繁使用各种代理IP,却又因为网上的公开 ...
好用的免费代理池搭建
好用的免费代理池 **项目地址:https://github.com/lin423497786/ProxyPool.git** 效果图运行项目使用项目地址:https://github.com/ ...
动态可维护ip代理池搭建（定时更新模块）
动态可维护ip代理池(爬虫模块)继上一次的爬虫模块,我们先来优化一下并添加redis配置参数,参数配置以自身机器设定. import random import time import request ...
手把手教你用Python搭建IP代理池，轻松破解请求频率限制反爬虫~
我们所写的爬虫,它对服务器发出的网络请求频率要比正常用户的高的多,从而开发者可以将请求频率过高的用户视为爬虫程序,从而来限制爬虫程序. 今天志斌就来给大家分享一下,如何用Python搭建一个IP代理池 ...
手把手教你用Python搭建IP代理池
今天给大家分享一下,如何用Python搭建一个IP代理池,来破解服务器通过对用户请求频率进行限制的反爬虫. 01 原理因为客户端的IP地址是唯一的,所以开发者便将IP地址作为客户端的身份标识. 服务 ...
Python搭建代理池
由于爬虫工作往往有大量数据需要爬取,便需要大量的备用IP更换,这时就需要用到代理IP池.将大量可以用于更换的代理IP汇聚要一起,便于管理和调用,IP池就这样产生了.IP池有一下特征:它里面的IP是持续 ...
使用python+redis搭建代理池
前言: 搭建一套稳定的代理池服务,可以保证爬虫的快速稳定运行,应对一些反爬技术. 一.安装redis Redis是一个字典结构的存储服务器,主要用来存储代理ip. redis下载地址:Releases ...

python代理池搭建

python代理池搭建相关推荐

最新文章

热门文章