Python3爬蟲中如何搭建Cookies池

小編給大家分享一下Python3爬蟲中如何搭建Cookies池，相信大部分人都還不怎么了解，因此分享這篇文章給大家參考一下，希望大家閱讀完這篇文章后大有收獲，下面讓我們一起去了解一下吧！

Cookies 池的搭建

很多時候，在爬取沒有登錄的情況下，我們也可以訪問一部分頁面或請求一些接口，因為畢竟網站本身需要做 SEO，不會對所有頁面都設置登錄限制。

但是，不登錄直接爬取會有一些弊端，弊端主要有以下兩點。

設置了登錄限制的頁面無法爬取。如某論壇設置了登錄才可查看資源，某博客設置了登錄才可查看全文等，這些頁面都需要登錄賬號才可以查看和爬取。

一些頁面和接口雖然可以直接請求，但是請求一旦頻繁，訪問就容易被限制或者 IP 直接被封，但是登錄之后就不會出現這樣的問題，因此登錄之后被反爬的可能性更低。

下面我們就第二種情況做一個簡單的實驗。以微博為例，我們先找到一個 Ajax 接口，例如新浪財經官方微博的信息接口 https://m.weibo.cn/api/container/getIndex?uid=1638782947&luicode=20000174

&type=uid&value=1638782947&containerid=1005051638782947，如果用瀏覽器直接訪問，返回的數據是 JSON 格式，如圖 10-7 所示，其中包含了新浪財經官方微博的一些信息，直接解析 JSON 即可提取信息。

                                                                圖 10-7 返回數據

但是，這個接口在沒有登錄的情況下會有請求頻率檢測。如果一段時間內訪問太過頻繁，比如打開這個鏈接，一直不斷刷新，則會看到請求頻率過高的提示，如圖 10-8 所示。

                                                          圖 10-8 提示頁面

如果重新打開一個瀏覽器窗口，打開 https://passport.weibo.cn/signin/login?entry=mweibo&r=

https://m.weibo.cn/，登錄微博賬號之后重新打開此鏈接，則頁面正常顯示接口的結果，而未登錄的頁面仍然顯示請求過于頻繁，如圖 10-9 所示。

                                                           圖 10-9 對比頁面

圖中左側是登錄了賬號之后請求接口的結果，右側是未登錄賬號請求接口的結果，二者的接口鏈接是完全一樣的。未登錄狀態無法正常訪問，而登錄狀態可以正常顯示。

因此，登錄賬號可以降低被封禁的概率。

我們可以嘗試登錄之后再做爬取，被封禁的幾率會小很多，但是也不能完全排除被封禁的風險。如果一直用同一個賬號頻繁請求，那就有可能遇到請求過于頻繁而封號的問題。

如果需要做大規模抓取，我們就需要擁有很多賬號，每次請求隨機選取一個賬號，這樣就降低了單個賬號的訪問頻率，被封的概率又會大大降低。

那么如何維護多個賬號的登錄信息呢？這時就需要用到 Cookies 池了。接下來我們看看 Cookies 池的構建方法。

1. 本節目標

我們以新浪微博為例來實現一個 Cookies 池的搭建過程。Cookies 池中保存了許多新浪微博賬號和登錄后的 Cookies 信息，并且 Cookies 池還需要定時檢測每個 Cookies 的有效性，如果某 Cookies 無效，那就刪除該 Cookies 并模擬登錄生成新的 Cookies。同時 Cookies 池還需要一個非常重要的接口，即獲取隨機 Cookies 的接口，Cookies 運行后，我們只需請求該接口，即可隨機獲得一個 Cookies 并用其爬取。

由此可見，Cookies 池需要有自動生成 Cookies、定時檢測 Cookies、提供隨機 Cookies 等幾大核心功能。

2. 準備工作

搭建之前肯定需要一些微博的賬號。需要安裝好 Redis 數據庫并使其正常運行。需要安裝 Python 的 redis-py、requests、Selelnium 和 Flask 庫。另外，還需要安裝 Chrome 瀏覽器并配置好 ChromeDriver，其流程可以參考第一章的安裝說明。

3. Cookies 池架構

Cookies 的架構和代理池類似，同樣是 4 個核心模塊，如圖 10-10 所示。

                                          圖 10-10 Cookies 池架構

Cookies 池架構的基本模塊分為 4 塊：存儲模塊、生成模塊、檢測模塊和接口模塊。每個模塊的功能如下。

存儲模塊負責存儲每個賬號的用戶名密碼以及每個賬號對應的 Cookies 信息，同時還需要提供一些方法來實現方便的存取操作。

生成模塊負責生成新的 Cookies。此模塊會從存儲模塊逐個拿取賬號的用戶名和密碼，然后模擬登錄目標頁面，判斷登錄成功，就將 Cookies 返回并交給存儲模塊存儲。

檢測模塊需要定時檢測數據庫中的 Cookies。在這里我們需要設置一個檢測鏈接，不同的站點檢測鏈接不同，檢測模塊會逐個拿取賬號對應的 Cookies 去請求鏈接，如果返回的狀態是有效的，那么此 Cookies 沒有失效，否則 Cookies 失效并移除。接下來等待生成模塊重新生成即可。

接口模塊需要用 API 來提供對外服務的接口。由于可用的 Cookies 可能有多個，我們可以隨機返回 Cookies 的接口，這樣保證每個 Cookies 都有可能被取到。Cookies 越多，每個 Cookies 被取到的概率就會越小，從而減少被封號的風險。

以上設計 Cookies 池的基本思路和前面講的代理池有相似之處。接下來我們設計整體的架構，然后用代碼實現該 Cookies 池。

4. Cookies 池的實現

首先分別了解各個模塊的實現過程。

存儲模塊

其實，需要存儲的內容無非就是賬號信息和 Cookies 信息。賬號由用戶名和密碼兩部分組成，我們可以存成用戶名和密碼的映射。Cookies 可以存成 JSON 字符串，但是我們后面得需要根據賬號來生成 Cookies。生成的時候我們需要知道哪些賬號已經生成了 Cookies，哪些沒有生成，所以需要同時保存該 Cookies 對應的用戶名信息，其實也是用戶名和 Cookies 的映射。這里就是兩組映射，我們自然而然想到 Redis 的 Hash，于是就建立兩個 Hash，結構分別如圖 10-11 和圖 10-12 所示。

                                            圖 10-11 用戶名密碼 Hash 結構

                                             圖 10-12 用戶名 Cookies 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())```
這里我們新建了一個 RedisClient 類，初始化__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。
#### 生成模塊
生成模塊負責獲取各個賬號信息并模擬登錄，隨后生成 Cookies 并保存。我們首先獲取兩個 Hash 的信息，看看賬戶的 Hash 比 
Cookies 的 Hash 多了哪些還沒有生成 Cookies 的賬號，然后將剩余的賬號遍歷，再去生成 Cookies 即可。
這里主要邏輯就是找出那些還沒有對應 Cookies 的賬號，然后再逐個獲取 Cookies，代碼如下：
```python
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。

檢測模塊

我們現在可以用生成模塊來生成 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 去請求目標站點，同時禁止重定向和設置超時時間，得到響應之后檢測其返回狀態碼。如果直接返回 200 狀態碼，則 Cookies 有效，否則可能遇到了 302 跳轉等情況，一般會跳轉到登錄頁面，則 Cookies 已失效。如果 Cookies 失效，我們將其從 Cookies 的 Hash 里移除即可。

接口模塊

生成模塊和檢測模塊如果定時運行就可以完成 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 '<h3>Welcome to Cookie Pool System</h3>'
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):
    """
    獲取隨機的 Cookie, 訪問地址如 /weibo/random
    :return: 隨機 Cookie
    """
    g = get_conn()
    cookies = getattr(g, website + '_cookies').random()
    return cookies

我們同樣需要實現通用的配置來對接不同的站點，所以接口鏈接的第一個字段定義為站點名稱，第二個字段定義為獲取的方法，例如，/weibo/random 是獲取微博的隨機 Cookies，/zhihu/random 是獲取知乎的隨機 Cookies。

調度模塊

最后，我們再加一個調度模塊讓這幾個模塊配合運行起來，主要的工作就是驅動幾個模塊定時運行，同時各個模塊需要在不同進程上運行，實現如下所示：

import time
from multiprocessing import Process
from cookiespool.api import app
from cookiespool.config import *
from cookiespool.generator import *
from cookiespool.tester import *
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 GENERATOR_MAP.items():
                    generator = eval(cls + '(website="' + website + '")')
                    generator.run()
                    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()

這里用到了兩個重要的配置，即產生模塊類和測試模塊類的字典配置，如下所示：

# 產生模塊類，如擴展其他站點，請在此配置
GENERATOR_MAP = {'weibo': 'WeiboCookiesGenerator'}
# 測試模塊類，如擴展其他站點，請在此配置
TESTER_MAP = {'weibo': 'WeiboValidTester'}

這樣的配置是為了方便動態擴展使用的，鍵名為站點名稱，鍵值為類名。如需要配置其他站點可以在字典中添加，如擴展知乎站點的產生模塊，則可以配置成：

GENERATOR_MAP = {
    'weibo': 'WeiboCookiesGenerator',
    'zhihu': 'ZhihuCookiesGenerator',
}

Scheduler 里將字典進行遍歷，同時利用 eval() 動態新建各個類的對象，調用其入口 run() 方法運行各個模塊。同時，各個模塊的多進程使用了 multiprocessing 中的 Process 類，調用其 start() 方法即可啟動各個進程。

另外，各個模塊還設有模塊開關，我們可以在配置文件中自由設置開關的開啟和關閉，如下所示：

# 產生模塊開關
GENERATOR_PROCESS = True
# 驗證模塊開關
VALID_PROCESS = False
# 接口模塊開關
API_PROCESS = True

定義為 True 即可開啟該模塊，定義為 False 即關閉此模塊。

至此，我們的 Cookies 就全部完成了。接下來我們將模塊同時開啟，啟動調度器，控制臺類似輸出如下所示：

API 接口開始運行
 * Running on http://0.0.0.0:5000/ (Press CTRL+C to quit)
Cookies 生成進程開始運行
Cookies 檢測進程開始運行
正在生成 Cookies 賬號 14747223314 密碼 asdf1129
正在測試 Cookies 用戶名 14747219309
Cookies 有效 14747219309
正在測試 Cookies 用戶名 14740626332
Cookies 有效 14740626332
正在測試 Cookies 用戶名 14740691419
Cookies 有效 14740691419
正在測試 Cookies 用戶名 14740618009
Cookies 有效 14740618009
正在測試 Cookies 用戶名 14740636046
Cookies 有效 14740636046
正在測試 Cookies 用戶名 14747222472
Cookies 有效 14747222472
Cookies 檢測完成
驗證碼位置 420 580 384 544
成功匹配
拖動順序 [1, 4, 2, 3]
成功獲取到 Cookies {'SUHB': '08J77UIj4w5n_T', 'SCF': 'AimcUCUVvHjswSBmTswKh0g4kNj4K7_U9k57YzxbqFt4SFBhXq3Lx4YSN
O9VuBV841BMHFIaH4ipnfqZnK7W6Qs.', 'SSOLoginState': '1501439488', '_T_WM': '99b7d656220aeb9207b5db97743adc02', 
'M_WEIBOCN_PARAMS': 'uicode%3D20000174', 'SUB': '_2A250elZQDeRhGeBM6VAR8ifEzTuIHXVXhXoYrDV6PUJbkdBeLXTxkW17Zo
YhhJ92N_RGCjmHpfv9TB8OJQ..'}
成功保存 Cookies

以上所示是程序運行的控制臺輸出內容，我們從中可以看到各個模塊都正常啟動，測試模塊逐個測試 Cookies，生成模塊獲取尚未生成 Cookies 的賬號的 Cookies，各個模塊并行運行，互不干擾。

我們可以訪問接口獲取隨機的 Cookies，如圖 10-13 所示。

                                                                     圖 10-13 接口頁面

爬蟲只需要請求該接口就可以實現隨機 Cookies 的獲取。

以上是Python3爬蟲中如何搭建Cookies池的所有內容，感謝各位的閱讀！相信大家都有了一定的了解，希望分享的內容對大家有所幫助，如果還想學習更多知識，歡迎關注億速云行業資訊頻道！