妈妈再也不用担心爬虫被封号了！手把手教你搭建Cookies池_崔庆才丨静觅的专栏文章_微信文章

很多时候，在爬取没有登录的情况下，我们也可以访问一部分页面或请求一些接口，因为毕竟网站本身需要做SEO，不会对所有页面都设置登录限制。

但是，不登录直接爬取会有一些弊端，弊端主要有以下两点。

设置了登录限制的页面无法爬取。如某论坛设置了登录才可查看资源，某博客设置了登录才可查看全文等，这些页面都需要登录账号才可以查看和爬取。
一些页面和接口虽然可以直接请求，但是请求一旦频繁，访问就容易被限制或者IP直接被封，但是登录之后就不会出现这样的问题，因此登录之后被反爬的可能性更低。

下面我们就第二种情况做一个简单的实验。以微博为例，我们先找到一个Ajax接口，例如新浪财经官方微博的信息接口https://m.weibo.cn/api/container/getIndex?uid=1638782947&luicode=20000174&type=uid&value=1638782947&containerid=1005051638782947，如果用浏览器直接访问，返回的数据是JSON格式，如下图所示，其中包含了新浪财经官方微博的一些信息，直接解析JSON即可提取信息。

但是，这个接口在没有登录的情况下会有请求频率检测。如果一段时间内访问太过频繁，比如打开这个链接，一直不断刷新，则会看到请求频率过高的提示，如下图所示。

如果重新打开一个浏览器窗口，打开https://passport.weibo.cn/signin/login?entry=mweibo&r=https://m.weibo.cn/，登录微博账号之后重新打开此链接，则页面正常显示接口的结果，而未登录的页面仍然显示请求过于频繁，如下图所示。

图中左侧是登录了账号之后请求接口的结果，右侧是未登录账号请求接口的结果，二者的接口链接是完全一样的。未登录状态无法正常访问，而登录状态可以正常显示。

因此，登录账号可以降低被封禁的概率。

我们可以尝试登录之后再做爬取，被封禁的几率会小很多，但是也不能完全排除被封禁的风险。如果一直用同一个账号频繁请求，那就有可能遇到请求过于频繁而封号的问题。

如果需要做大规模抓取，我们就需要拥有很多账号，每次请求随机选取一个账号，这样就降低了单个账号的访问频率，被封的概率又会大大降低。

那么如何维护多个账号的登录信息呢？这时就需要用到Cookies池了。接下来我们看看Cookies池的构建方法。

一、本节目标

我们以新浪微博为例来实现一个Cookies池的搭建过程。Cookies池中保存了许多新浪微博账号和登录后的Cookies信息，并且Cookies池还需要定时检测每个Cookies的有效性，如果某Cookies无效，那就删除该Cookies并模拟登录生成新的Cookies。同时Cookies池还需要一个非常重要的接口，即获取随机Cookies的接口，Cookies运行后，我们只需请求该接口，即可随机获得一个Cookies并用其爬取。

由此可见，Cookies池需要有自动生成Cookies、定时检测Cookies、提供随机Cookies等几大核心功能。

二、准备工作

搭建之前肯定需要一些微博的账号。需要安装好Redis数据库并使其正常运行。需要安装Python的RedisPy、requests、Selelnium、Flask库。另外，还需要安装Chrome浏览器并配置好ChromeDriver。

三、Cookies池架构

Cookies的架构和代理池类似，同样是4个核心模块，如下图所示。

Cookies池架构的基本模块分为4块：存储模块、生成模块、检测模块、接口模块。每个模块的功能如下。

存储模块负责存储每个账号的用户名密码以及每个账号对应的Cookies信息，同时还需要提供一些方法来实现方便的存取操作。
生成模块负责生成新的Cookies。此模块会从存储模块逐个拿取账号的用户名和密码，然后模拟登录目标页面，判断登录成功，就将Cookies返回并交给存储模块存储。
检测模块需要定时检测数据库中的Cookies。在这里我们需要设置一个检测链接，不同的站点检测链接不同，检测模块会逐个拿取账号对应的Cookies去请求链接，如果返回的状态是有效的，那么此Cookies没有失效，否则Cookies失效并移除。接下来等待生成模块重新生成即可。
接口模块需要用API来提供对外服务的接口。由于可用的Cookies可能有多个，我们可以随机返回Cookies的接口，这样保证每个Cookies都有可能被取到。Cookies越多，每个Cookies被取到的概率就会越小，从而减少被封号的风险。

以上设计Cookies池的的基本思路和前面讲的代理池有相似之处。接下来我们设计整体的架构，然后用代码实现该Cookies池。

四、Cookies池的实现

首先分别了解各个模块的实现过程。

1. 存储模块

其实，需要存储的内容无非就是账号信息和Cookies信息。账号由用户名和密码两部分组成，我们可以存成用户名和密码的映射。Cookies可以存成JSON字符串，但是我们后面得需要根据账号来生成Cookies。生成的时候我们需要知道哪些账号已经生成了Cookies，哪些没有生成，所以需要同时保存该Cookies对应的用户名信息，其实也是用户名和Cookies的映射。这里就是两组映射，我们自然而然想到Redis的Hash，于是就建立两个Hash，结构分别如下图所示。

Hash的Key就是账号，Value对应着密码或者Cookies。另外需要注意，由于Cookies池需要做到可扩展，存储的账号和Cookies不一定单单只有本例中的微博，其他站点同样可以对接此Cookies池，所以这里Hash的名称可以做二级分类，例如存账号的Hash名称可以为accounts:weibo，Cookies的Hash名称可以为cookies:weibo。如要扩展知乎的Cookies池，我们就可以使用accounts:zhihu和cookies:zhihu，这样比较方便。

接下来我们创建一个存储模块类，用以提供一些Hash的基本操作，代码如下：

import random
import redis

class RedisClient(object):
    def __init__(self, type, website, host=REDIS_HOST, port=REDIS_PORT, password=REDIS_PASSWORD):
        """
        初始化Redis连接
        :param host: 地址
        :param port: 端口
        :param password: 密码
        """
        self.db = redis.StrictRedis(host=host, port=port, password=password, decode_responses=True)
        self.type = type
        self.website = website

    def name(self):
        """
        获取Hash的名称
        :return: Hash名称
        """
        return "{type}:{website}".format(type=self.type, website=self.website)

    def set(self, username, value):
        """
        设置键值对
        :param username: 用户名
        :param value: 密码或Cookies
        :return:
        """
        return self.db.hset(self.name(), username, value)

    def get(self, username):
        """
        根据键名获取键值
        :param username: 用户名
        :return:
        """
        return self.db.hget(self.name(), username)

    def delete(self, username):
        """
        根据键名删除键值对
        :param username: 用户名
        :return: 删除结果
        """
        return self.db.hdel(self.name(), username)

    def count(self):
        """
        获取数目
        :return: 数目
        """
        return self.db.hlen(self.name())

    def random(self):
        """
        随机得到键值，用于随机Cookies获取
        :return: 随机Cookies
        """
        return random.choice(self.db.hvals(self.name()))

    def usernames(self):
        """
        获取所有账户信息
        :return: 所有用户名
        """
        return self.db.hkeys(self.name())

    def all(self):
        """
        获取所有键值对
        :return: 用户名和密码或Cookies的映射表
        """
        return self.db.hgetall(self.name())

这里我们新建了一个 RedisClien t类，初始化 __init__() 方法有两个关键参数 type 和 website ，分别代表类型和站点名称，它们就是用来拼接Hash名称的两个字段。如果这是存储账户的Hash，那么此处的 type 为 accounts 、 website 为 weibo ，如果是存储Cookies的Hash，那么此处的 type 为 cookies 、 website 为 weibo 。

接下来还有几个字段代表了Redis的连接信息，初始化时获得这些信息后初始化 StrictRedis 对象，建立Redis连接。

name() 方法拼接了 type 和 website ，组成Hash的名称。 set() 、 get() 、 delete() 方法分别代表设置、获取、删除Hash的某一个键值对， count() 获取Hash的长度。

比较重要的方法是 random() ，它主要用于从Hash里随机选取一个Cookies并返回。每调用一次 random() 方法，就会获得随机的Cookies，此方法与接口模块对接即可实现请求接口获取随机Cookies。

2. 生成模块

生成模块负责获取各个账号信息并模拟登录，随后生成Cookies并保存。我们首先获取两个Hash的信息，看看账户的Hash比Cookies的Hash多了哪些还没有生成Cookies的账号，然后将剩余的账号遍历，再去生成Cookies即可。

这里主要逻辑就是找出那些还没有对应Cookies的账号，然后再逐个获取Cookies，代码如下：

for username in accounts_usernames:
    if not username in cookies_usernames:
        password = self.accounts_db.get(username)
        print('正在生成Cookies', '账号', username, '密码', password)
        result = self.new_cookies(username, password)

因为我们对接的是新浪微博，前面我们已经破解了新浪微博的四宫格验证码，在这里我们直接对接过来即可，不过现在需要加一个获取Cookies的方法，并针对不同的情况返回不同的结果，逻辑如下所示：

def get_cookies(self):
    return self.browser.get_cookies()

def main(self):
    self.open()
    if self.password_error():
        return {
            'status': 2,
            'content': '用户名或密码错误'
        }
    # 如果不需要验证码直接登录成功
    if self.login_successfully():
        cookies = self.get_cookies()
        return {
            'status': 1,
            'content': cookies
        }
    # 获取验证码图片
    image = self.get_image('captcha.png')
    numbers = self.detect_image(image)
    self.move(numbers)
    if self.login_successfully():
        cookies = self.get_cookies()
        return {
            'status': 1,
            'content': cookies
        }
    else:
        return {
            'status': 3,
            'content': '登录失败'
        }

这里返回结果的类型是字典，并且附有状态码 status ，在生成模块里我们可以根据不同的状态码做不同的处理。例如状态码为1的情况，表示成功获取Cookies，我们只需要将Cookies保存到数据库即可。如状态码为2的情况，代表用户名或密码错误，那么我们就应该把当前数据库中存储的账号信息删除。如状态码为3的情况，则代表登录失败的一些错误，此时不能判断是否用户名或密码错误，也不能成功获取Cookies，那么简单提示再进行下一个处理即可，类似代码实现如下所示：

result = self.new_cookies(username, password)
# 成功获取
if result.get('status') == 1:
    cookies = self.process_cookies(result.get('content'))
    print('成功获取到Cookies', cookies)
    if self.cookies_db.set(username, json.dumps(cookies)):
        print('成功保存Cookies')
# 密码错误，移除账号
elif result.get('status') == 2:
    print(result.get('content'))
    if self.accounts_db.delete(username):
        print('成功删除账号')
else:
    print(result.get('content'))

如果要扩展其他站点，只需要实现 new_cookies() 方法即可，然后按此处理规则返回对应的模拟登录结果，比如1代表获取成功，2代表用户名或密码错误。

代码运行之后就会遍历一次尚未生成Cookies的账号，模拟登录生成新的Cookies。

3. 检测模块

我们现在可以用生成模块来生成Cookies，但还是免不了Cookies失效的问题，例如时间太长导致Cookies失效，或者Cookies使用太频繁导致无法正常请求网页。如果遇到这样的Cookies，我们肯定不能让它继续保存在数据库里。

所以我们还需要增加一个定时检测模块，它负责遍历池中的所有Cookies，同时设置好对应的检测链接，我们用一个个Cookies去请求这个链接。如果请求成功，或者状态码合法，那么该Cookies有效；如果请求失败，或者无法获取正常的数据，比如直接跳回登录页面或者跳到验证页面，那么此Cookies无效，我们需要将该Cookies从数据库中移除。

此Cookies移除之后，刚才所说的生成模块就会检测到Cookies的Hash和账号的Hash相比少了此账号的Cookies，生成模块就会认为这个账号还没生成Cookies，那么就会用此账号重新登录，此账号的Cookies又被重新更新。

检测模块需要做的就是检测Cookies失效，然后将其从数据中移除。

为了实现通用可扩展性，我们首先定义一个检测器的父类，声明一些通用组件，实现如下所示：

class ValidTester(object):
    def __init__(self, website='default'):
        self.website = website
        self.cookies_db = RedisClient('cookies', self.website)
        self.accounts_db = RedisClient('accounts', self.website)

    def test(self, username, cookies):
        raise NotImplementedError

    def run(self):
        cookies_groups = self.cookies_db.all()
        for username, cookies in cookies_groups.items():
            self.test(username, cookies)

在这里定义了一个父类叫作 ValidTester ，在 __init__() 方法里指定好站点的名称 website ，另外建立两个存储模块连接对象 cookies_db 和 accounts_db ，分别负责操作Cookies和账号的Hash， run() 方法是入口，在这里是遍历了所有的Cookies，然后调用 test() 方法进行测试，在这里 test() 方法是没有实现的，也就是说我们需要写一个子类来重写这个 test() 方法，每个子类负责各自不同网站的检测，如检测微博的就可以定义为 WeiboValidTester ，实现其独有的 test() 方法来检测微博的Cookies是否合法，然后做相应的处理，所以在这里我们还需要再加一个子类来继承这个 ValidTester ，重写其 test() 方法，实现如下：

import json
import requests
from requests.exceptions import ConnectionError

class WeiboValidTester(ValidTester):
    def __init__(self, website='weibo'):
        ValidTester.__init__(self, website)

    def test(self, username, cookies):
        print('正在测试Cookies', '用户名', username)
        try:
            cookies = json.loads(cookies)
        except TypeError:
            print('Cookies不合法', username)
            self.cookies_db.delete(username)
            print('删除Cookies', username)
            return
        try:
            test_url = TEST_URL_MAP[self.website]
            response = requests.get(test_url, cookies=cookies, timeout=5, allow_redirects=False)
            if response.status_code == 200:
                print('Cookies有效', username)
                print('部分测试结果', response.text[0:50])
            else:
                print(response.status_code, response.headers)
                print('Cookies失效', username)
                self.cookies_db.delete(username)
                print('删除Cookies', username)
        except ConnectionError as e:
            print('发生异常', e.args)

test() 方法首先将Cookies转化为字典，检测Cookies的格式，如果格式不正确，直接将其删除，如果格式没问题，那么就拿此Cookies请求被检测的URL。 test() 方法在这里检测微博，检测的URL可以是某个Ajax接口，为了实现可配置化，我们将测试URL也定义成字典，如下所示：

TEST_URL_MAP = {
    'weibo': 'https://m.weibo.cn/'
}

如果要扩展其他站点，我们可以统一在字典里添加。对微博来说，我们用Cookies去请求目标站点，同时禁止重定向和设置超时时间，得到Response之后检测其返回状态码。如果直接返回200状态码，则Cookies有效，否则可能遇到了302跳转等情况，一般会跳转到登录页面，则Cookies已失效。如果Cookies失效，我们将其从Cookies的Hash里移除即可。

4. 接口模块

生成模块和检测模块如果定时运行就可以完成Cookies实时检测和更新。但是Cookies最终还是需要给爬虫来用，同时一个Cookies池可供多个爬虫使用，所以我们还需要定义一个Web接口，爬虫访问此接口便可以取到随机的Cookies。我们采用Flask来实现接口的搭建，代码如下所示：

import json
from flask import Flask, g
app = Flask(__name__)
# 生成模块的配置字典
GENERATOR_MAP = {
    'weibo': 'WeiboCookiesGenerator'
}
@app.route('/')
def index():
    return '<h2>Welcome to Cookie Pool System</h2>'

def get_conn():
    for website in GENERATOR_MAP:
        if not hasattr(g, website):
            setattr(g, website + '_cookies', eval('RedisClient' + '("cookies", "' + website + '")'))
    return g

@app.route('/<website>/random')
def random(website):
    ""

妈妈再也不用担心爬虫被封号了！手把手教你搭建Cookies池

正文