你好，我是小焦。之前给大家分享了很多有趣的小爬虫。今天也是给大家分享一下爬虫的代理池如何搭建。作为一个业余编程爱好者，也是一边学习一边建立。摸爬滚打的也是建立起来的了。今天给大家分享一下吧，希望对大家有所帮助吧！

废话也就不多说了，进入正题吧！很多初学者可能不知道什么是cookie池？简单说下吧。平时上网逛过网站的朋友平时可能会发现，我们在一个网站登录过账号密码，当我们第二次打开浏览器进入改网站的时候，不需要登录会直接进入到登录状态，省去了每次登录的麻烦，这是因为我们的浏览器保留我们的cookie在本地。

    那么这个cookie对爬虫来说有什么意义呢？我们知道反爬机制在互联网应用是越来越多，有的网站我们没有登录的话连主页都看不见，其次一个账号的访问频率太高的话也会被封禁，所以不得不应用cookie池和代理池了，之前我们说过代理池得搭建，今天就来说说cookie池吧。

搭建思路：

Cookies池的架构和代理池类似，同样是4个核心模块（存储模块、生成模块、检测模块和接口模块）：

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

其实整体的流程就是：生成模块拿到cookie并保存到数据库中，检测模块调用cookie访问指定网站测试有效性，最后将有效的cookie通过api接口供爬虫使用。

存储模块：

    存储这块我们使用redis数据库，因为包含账号密码，和账号对应的cookie，所以使用hash非常方便。构建两组映射，由于我们后期还要扩展，所以在命名上面的标注，这里以我们本次案例穿越火线的官方网站为例。存放账号的hash命名为accounts:cf，存放cookies的则命名为cookies:cf。为了后期调用方便，我们也将后期使用的方法封装成一个个函数。

save.py代码如下：

# 此文件为储存模块，主要是将获取的cookie加载到redis中
# 将redis中hash的用法封装为一个个函数
# 2021.1.10


import redis
import random

# 先设置链接的IP和端口

REDIS_HOST = 'localhost'
REDIS_PORT = 6379
REDIS_PASSWORD = None

# 将redis的方法进行封装成类
class RedisClient(object):
    # 实例化类
    def __init__(self,type,website,host=REDIS_HOST,port=REDIS_PORT,password=REDIS_PASSWORD):
        self.db = redis.StrictRedis(host=host,port=port,password=password,decode_responses=True)
        self.type = type  # 这块为储存的类型
        self.website = website  # 这块是网站的名称，例如我们存新浪的，可以命名为xinlang

    # 后期为了相同用户名的密码和cookie等进行映射，这里我们使用redis中hash

    # 写一个名字函数
    def name(self):
        # 生成名字
        return "{type}:{website}".format(type=self.type,website=self.website)

    # 设置键值对，usename为用户名，value为密码或者cookie值
    def set(self,usename,value):
        # 给对应的key赋值用户名和值
        return self.db.hset(self.name(),usename,value)

    # 根据key中的键名获取值
    def get(self,usename):
        return self.db.hget(self.name(),usename)

    # 根据键名删除键值对
    def delete(self,usename):
        return self.db.hdel(self.name(),usename)

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

    # 随机得到键值，这里我们用来获取cookies
    def random(self):
        # 这里使用random的choice方法随机获取，里面的方法为hash获取所有的值
        return random.choice(self.db.hvals(self.name()))

    # 获取所有账户信息，获取所有的账户名,也就是所有的键
    def usernames(self):
        return self.db.hkeys(self.name())

    # 获取所有的键值对
    def all(self):
        # 获取所有的键值对，可以用来获取账号密码或者账号对应的cookies
        return self.db.hgetall(self.name())

注意的是，上面我们构建了一个name()函数来拼接成hash的名称方便后期区分。

然后就是导入账号了，这里小焦专门写了一个导入程序，账号和密码我们按照一定的格式存放在account.txt文件中即可。导入程序我们命名为addaccount.py。有新加账号时，我们体现吧账号批量写入到txt文件中，然后运行此程序导入。

# 此文件用来添加我们自己的账号到redis中
# 创建cookie池肯定会用到大量的账号，所以我们这边以txt文件为例，不用我们手动输入了
# 将事先申请好的账号密码保存到txt文件中，格式为：账号----密码，然后直接读取就行
# 默认放到同路径下的account.txt中，

from save import RedisClient

# 先实例化Readis类，参入key名字
conn = RedisClient('account','cf')

def readaccount(sp='----'):
    print('开始读取读取account.txt文件......')
    with open("account.txt","r") as f:
        datas = f.readlines()
        # readlinses方法是将txt文件中的内容以列表形式输出
        for data in datas:
            # 由于txt文件中自带\n，所以我们先去除换行符然后分割账号密码
            account,password=data.strip('\n').split(sp)
            print("正在导入账号:%s   密码:%s" %(account,password))
            # 调用我们实现写好的set方法将账号密码储存到redis中
            result = conn.set(account,password)
            print('导入成功\n' if result else '导入失败')

if __name__ == "__main__":
    readaccount()

生成模块：

    生成模块主要的作用就是模拟登录获取cookies，然后对比账号和cookie看看那些账号还没有生成，重新获取。由于我们做的是一个可扩展的cookie池，所以不同网站得单独写一个类来调用，这里是以我经常玩得一个游戏穿越火线得官网为例来写得，后期还有一个小商机，暂时还没来的及做哦，哈哈。主次原因，我们先看看CF官网的cookie获取代码，再来看父类吧。

cookies.py

# 此文件为我们对应网站的cookie获取
# 使用selenium来动态抓取网页

from selenium import webdriver
# 下面这个模块是判断页面加载完毕用的,是显示等待
from selenium.webdriver.support.wait import WebDriverWait
# 该模块用来判断元素是否存在或是否点击，这个和上面的模块一般都是组合使用
from selenium.webdriver.support import expected_conditions as EC
# By是内置的class，定义了很多方法
from selenium.webdriver.common.by import By
import selenium.common.exceptions as ex
import time

class CfCookies():
    def __init__(self,username,password,browser):
        self.url = 'https://cf.qq.com/cp/a20210111cattle/pc/index.shtml'
        # 这里将brower单独重新定义主要用于不同得浏览器调用
        self.browser = browser
        # 设置浏览器最长等待时间，这里我们设置为10秒
        self.wait = WebDriverWait(browser,10)
        self.username = username
        self.password = password

# 这里开始进行浏览器得操作
    def open(self):
        # 打开浏览器
        self.browser.get(self.url)
        # 用显示等待，定位并点击登录按钮
        self.wait.until(EC.presence_of_element_located((By.ID,'dologin'))).click()
        # 点击登录按钮后，弹出QQ得登录页面，这里iframe嵌套页面，切换到子页面
        self.browser.switch_to.frame('loginIframe')
        # 选择账号密码登录按钮，并点击
        self.wait.until(EC.presence_of_element_located((By.ID,'switcher_plogin'))).click()
        # 定位到账号和密码输入框，并输入账号密码
        self.wait.until(EC.presence_of_element_located((By.ID,'u'))).send_keys(self.username)
        self.wait.until(EC.presence_of_element_located((By.ID,'p'))).send_keys(self.password)
        time.sleep(2)
        # 点击登录
        self.wait.until(EC.presence_of_element_located((By.ID,"login_button"))).click()

        # 点击登录后，会弹出验证码，由于验证码更新换代太快，以往得破解方法不适用，这里我们手动完成

    # 该函数判断密码账号是否输入错误,如果错误或超时返回False
    def password_error(self):
        try:
            # 当密码错误时会弹出一个提示，我们只需捕获这个错误提示就知道是否输入错误
            return bool(self.wait.until(EC.presence_of_element_located((By.ID,'err_m'))))
        except ex.TimeoutException:
            return False

    # 获取cookies
    def get_cookies(self):
        return self.browser.get_cookies()

    # 判断是否登录成功。
    def login_successfully(self):
        # 当登录成功后会出现注销按钮，我们只需判断这个即可
        try:
            return bool(self.wait.until(EC.presence_of_element_located((By.ID,"dologout"))))
        except ex.TimeoutException:
            return False

    # 这里对整个获取流程运行
    def main(self):
        self.open()
        # 登录账号后，记得手动滑动验证码
        time.sleep(10)
        # 判断是否账号密码错误
        if self.password_error():
            return {
                'status':2,
                'content':'用户名或密码错误'
            }
        elif self.login_successfully():
            # 如果登录成功，则获取cookies
            cookies = self.get_cookies()
            return {
                'status':1,
                'content':cookies
            }
        else:
            return {
                'status':3,
                'content':'登录失败'
            }

上面我们将CF网站的抓取流程写成一个类，后期我们只需每个网站写成一个单独的类，需要哪个调用哪个就成，一劳永逸。现在来看看生成模块的主程序吧。该程序主要写了一个父类，后期我们的子类使用全部继承父类的方法。

这里我们命名为getter.py

# 该文件为cookie获取文件
from selenium import webdriver
from save import RedisClient
from cookies import CfCookies
import json
import time

# 这里我们先写一个通用类，后期不同得网站重新调用父类
class CookiesGenerator(object):
    def __init__(self,website='default'):
        self.website = website
        self.cookies_db = RedisClient('cookies',self.website)
        self.account_db = RedisClient('account',self.website)
        self.init_browser()

    # 当不需要实例得时候我们手动销毁，释放内存
    def __del__(self):
        self.close()

    # 这里对浏览器进行设置，
    def init_browser(self):
        # 如果加上下面两行代码让浏览器在后台静默执行，由于我们要手动验证码所以忽略
        # self.option = webdriver.ChromeOptions()
        # self.option.add_argument('headless')
        self.browser = webdriver.Chrome()

    # 获取cookies，子类到时候自行重写，不然报错
    def new_cookies(self,username,password):
        raise NotImplementedError

    #提取cookies中得name和value重新生产新字典,其他字段无用
    def process_cookies(self,cookies):
        dict = {}
        for cookie in cookies:
            dict[cookie['name']] = cookie['value']
        return dict

    # 运行函数
    def run(self):
        # 导出所有得账号列表
        account_usernames = self.account_db.usernames()
        cookies_usernames = self.cookies_db.usernames()

        # 遍历所有账号，找出没有cookies得账号
        for username in account_usernames:
            if not username in cookies_usernames:
                password = self.account_db.get(username)
                print('正在生成cookies','账号:',username,'密码----')
                result = self.new_cookies(username,password)
                time.sleep(10)
                # 这快利用我们cookies文件生成得状态码判断登录状态,登录正常得获取并保存cookies，错误得则删除账号
                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.account_db.delete(username):
                        print('成功删除错误账号')

                else:
                    print(result.get('content'))

        else:
            print('所有账号都已成功获取cookies')


    def close(self):
        try:
            print('关闭浏览器')
            self.browser.close()
            del self.browser

        except TypeError:
            print('浏览器关闭失败')

# 继承上面得类，重新子类
class CfCookiesGenerator(CookiesGenerator):
    def __init__(self,website='cf'):
        CookiesGenerator.__init__(self,website)
        self.website = website

    def new_cookies(self,username,password):
        return CfCookies(username,password,self.browser).main()

# a = CfCookiesGenerator()
# a.run()

检测模块：

    改模块主要检测我们的cookies的有效性，像上面的CF官网cookies时间就非常短了，经常会失效，所以我们要定期检测。这里直接用request测试就行。失效的或者错误的直接从数据库中剔除。文件命名为tester.py

# 此文件为cookies得测试文件

from save import RedisClient
import json
import requests

# 测试URL
TEST_URL_MAP = {
    'cf':'https://cf.qq.com/cp/a20210111cattle/pc/index.shtml',
}

# 写一个测试模块父类
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)
        self.header = {
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 '
        }

    # 测试cookies，这块用子类重写
    def test(self,username,cookies):
        raise NotImplementedError

    # 获取所有得cookies进行测试
    def run(self):
        cookies_groups = self.cookies_db.all()
        for username,cookies in cookies_groups.items():
            self.test(username,cookies)

class CfValidTester(ValidTester):
    def __init__(self,website='cf'):
        ValidTester.__init__(self,website)

    def test(self,username,cookies):
        print('开始测试cookies','用户名：',username)
        try:
            cookies = json.loads(cookies)
        except TypeError:
            print('cookie不合法',username)
            self.cookies_db.delete(username)
            print('删除cookies',username)
            return

        try:
            # 调取对应得测试链接
            test_url = TEST_URL_MAP[self.website]
            response = requests.get(test_url,headers=self.header,cookies=cookies,timeout=5,allow_redirects=False)
            if response.status_code == 200:
                print('cookies有效',username)
            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)

a = CfValidTester()
a.run()

接口模块：

    接口还是使用flask框架构建web来调用，后期只需按照指定格式访问网站即可。命名为api.py

# 方便后期爬虫调用，这是接口文件

import json
from flask import Flask,g
from save import *

GENERATOR_MAP = {
    'cf':'https://cf.qq.com/cp/a20210111cattle/pc/index.shtml'
}

__all__ = ['app']

app = Flask(__name__)

@app.route('/')
def index():
    return '<h2>欢迎进入cookie池系统</h2>'
def get_conn():
    for website in GENERATOR_MAP:
        print(website)
        #g是个全局对象,利用setattr给该对象设置方法
        if not hasattr(g,website):
            setattr(g,website+'_cookies',eval('RedisClient'+'("cookies","'+website+'")'))
            setattr(g, website + '_accounts', eval('RedisClient' + '("accounts","' + website + '")'))
        return g

@app.route('/<website>/random')
def random(website):
    '''
    获取随机的cookie,访问地址如/zhihu/random
    :param website:站点
    :return:随机cookie
    '''
    g = get_conn()
    # 调用设置号得方法
    cookies = getattr(g,website+'_cookies').random()
    return cookies

def add(website,username,password):
    '''
    添加用户，访问地址如/mafeng/add/user/password
    :param website: 站点
    :param username: 用户名
    :param password: 密码
    :return:
    '''
    g = get_conn()
    print(username,password)
    getattr(g,website+'_accounts').set(username,password)
    return json.dumps({'status':'1'})

@app.route('/<website>/count')
def count(website):
    '''
    获取cookies总数
    :param website:
    :return:
    '''
    g = get_conn()
    count = getattr(g,website+'_cookies').count()
    return json.dumps({'status': '1', 'count': count})

# app.run(host='127.0.0.1',port=5000)

调度模块：

    调度模块主要是为了让cookies池自己运行。或者我们自己手动一个一个模块运行都行，这里主要是为了方便集中化管理。命名为scheduler.py

#  该文件为总体得调度模块

import time
from tester import *
from save import *
from getter import *
from api import *
from multiprocessing import Process

API_HOST = '127.0.0.1'
API_PORT = 5000
# 产生器和验证器循环周期
CYCLE = 120

# 测试模块，后期将新增拓展可加到里面
TESTER_MAP = {
    'cf':'CfValidTester'
}
# 为了实现扩展性，这里后期将其他cookie池添加进来调用
GENERATE_MAP = {
    'cf':'CfCookiesGenerator'
}
# 产生器开关，模拟登录添加Cookies
GENERATOR_PROCESS = False
# 验证器开关，循环检测数据库中Cookies是否可用，不可用删除
VALID_PROCESS = False
# API接口服务
API_PROCESS = False


class Scheduler(object):
    @staticmethod
    def valid_cookie(cycle=CYCLE):
        while True:
            print('Cookies检测进程开始运行')
            try:
                for website,cls in TESTER_MAP.items():
                    tester = eval(cls + '(website="'+ website +'")')
                    tester.run()
                    print('Cookies检测完成')
                    del tester
                    time.sleep(cycle)
            except Exception as e:
                print(e.args)

    @staticmethod
    def generate_cookie(cycle=CYCLE):
        while True:
            print('Cookies生成进程开始运行')
            try:
                for website, cls in GENERATE_MAP.items():
                    # 这里使用eval函数根据不同得字符组成不用得调用函数。例如CfCookies(website="cf")
                    generator = eval(cls + '(website="' + website + '")')
                    generator.run()
                    time.sleep(20)
                    print('Cookies生成完成')
                    generator.close()
                    time.sleep(cycle)
            except Exception as e:
                print(e.args)

    @staticmethod
    def api():
        print('API接口开始运行')
        app.run(host=API_HOST,port=API_PORT)

    def run(self):
        if API_PROCESS:
            api_process = Process(target=Scheduler.api)
            api_process.start()
        if GENERATOR_PROCESS:
            generate_process = Process(target=Scheduler.generate_cookie)
            generate_process.start()
        if VALID_PROCESS:
            valid_process = Process(target=Scheduler.valid_cookie)
            valid_process.start()

今天cookie池也就介绍到这，大体框架也是参考崔大神的模式构建，其中也就很多的小知识点之前没有学习到。感谢阅读。技术探讨关注小焦的公众号进群交流哦

源码地址：https://github.com/zhanjiuyou/cookiechi

我的个人博客：https://jiaokangyang.com