爬蟲神器pyppeteer，對 js 加密降維打擊

參考文章：

• 一日一技：如何正確移除Selenium中window.navigator.webdriver的值
• pyppeteer使用總結
• pyppeteer登錄淘寶筆記

pyppeteer 是對無頭瀏覽器 puppeteer的 Python 封裝。無頭瀏覽器普遍用於自動化測試，同時也是一種很好地爬蟲思路。

使用 puppeteer（等其餘無頭瀏覽器）的最大優點固然是對 js 加密實行降維打擊，徹底無視 js 加密手段，對於一些須要登陸的應用，也能夠模擬點擊而後保存 cookie。而不少時候前端的加密是爬蟲最難攻克的一部分。固然puppeteer也有劣勢，最大的劣勢就是相比面向接口爬蟲效率很低，就算是無頭的chromium，那也會佔用至關一部份內存。另外額外維護一個瀏覽器的啓動、關閉也是一種負擔。

這篇文章咱們來寫一個簡單的 demo，爬取拼多多搜索頁面的數據，最終的效果以下：

咱們把全部 api 請求的原始數據保存下來：

示例 json 文件以下：

開發環境

• python3.6+

最好是 python3.7，由於asyncio在 py3.7中加入了很好用的asyncio.run()方法。

• 安裝pyppeteer

若是安裝有問題請去看官方文檔。

python3 -m pip install pyppeteer
複製代碼

• 安裝 chromium

你懂的，天朝網絡環境很複雜，若是要用pyppeteer本身綁定的chromium，半天都下載不下來，因此咱們要手動安裝，而後在程序裏面指定executablePath。

下載地址：www.chromium.org/getting-inv…

hello world

pyppeteer 的 hello world 程序是前往exmaple.com截個圖：

import asyncio
from pyppeteer import launch

async def main():
    browser = await launch({
        # Windows 和 Linux 的目錄不同，情換成本身對應的executable文件地址
        'executablePath': '你下載的Chromium.app/Contents/MacOS/Chromium',
    })
    page = await browser.newPage()
    await page.goto('http://example.com')
    await page.screenshot({'path': 'example.png'})
    await browser.close()

asyncio.get_event_loop().run_until_complete(main())
複製代碼

pyppeteer 重要接口介紹

pyppeteer.launch

launch 瀏覽器，能夠傳入一個字典來配置幾個options，好比：

browser = await pyppeteer.launch({
    'headless': False, # 關閉無頭模式
    'devtools': True, # 打開 chromium 的 devtools
    'executablePath': '你下載的Chromium.app/Contents/MacOS/Chromiu',
    'args': [ 
        '--disable-extensions',
        '--hide-scrollbars',
        '--disable-bundled-ppapi-flash',
        '--mute-audio',
        '--no-sandbox',
        '--disable-setuid-sandbox',
        '--disable-gpu',
    ],
    'dumpio': True,  
})
複製代碼

其中全部可選的args參數在這裏：peter.sh/experiments…

dumpio的做用：把無頭瀏覽器進程的 stderr 核 stdout pip 到主程序，也就是設置爲 True 的話，chromium console 的輸出就會在主程序中被打印出來。

注入 js 腳本

能夠經過page.evaluate形式，例如：

await page.evaluate(""" () =>{ Object.defineProperties(navigator,{ webdriver:{ get: () => false } }) } """)
複製代碼

咱們會看到這一步很是關鍵，由於puppeteer出於政策考慮（這個詞用的不是很好，就是那個意思）會設置window.navigator.webdriver爲true，告訴網站我是一個 webdriver 驅動的瀏覽器。有些網站比較聰明（反爬措施作得比較好），就會經過這個來判斷對方是否是爬蟲程序。詳情情參考這篇文章：一日一技：如何正確移除Selenium中window.navigator.webdriver的值

這等價於在 devtools 裏面輸入那一段 js 代碼。

還能夠加載一個 js 文件：

await page.addScriptTag(path=path_to_your_js_file)
複製代碼

經過注入 js 腳本能完成不少不少有用的操做，好比自動下拉頁面等。

截獲 request 和 response

await page.setRequestInterception(True)
page.on('request', intercept_request)
page.on('response', intercept_response)
複製代碼

intercept_request和intercept_response至關因而註冊的兩個回調函數，在瀏覽器發出請求和獲取到請求以前指向這兩個函數。

好比能夠這樣禁止獲取圖片、多媒體資源和發起 websocket 請求：

async def intercept_request(req):
    """請求過濾"""
    if req.resourceType in ['image', 'media', 'eventsource', 'websocket']:
        await req.abort()
    else:
        await req.continue_()
複製代碼

而後每次獲取到請求以後將內容打印出來（這裏只打印了fetch和xhr類型response 的內容）：

async def intercept_response(res):
    resourceType = res.request.resourceType
    if resourceType in ['xhr', 'fetch']:
        resp = await res.text()
        print(resp)
複製代碼

一共有哪些resourceType，pyppeteer文檔裏面有：

拼多多搜索爬蟲

頁面自動下拉

拼多多的搜索界面是一個無限下拉的頁面，咱們但願可以實現無限下拉頁面，而且可以控制程序提早退出，否則一直下拉也很差，咱們可能並不須要那麼多數據。

js 腳本

async () => {
    await new Promise((resolve, reject) => {

        // 容許下滑的最大高度，防止那種能夠無限下拉的頁面沒法結束
        const maxScrollHeight = null;

        // 控制下拉次數
        const maxScrollTimes = null;
        let currentScrollTimes = 0;

        // 記錄上一次scrollHeight，便於判斷這次下拉操做有沒有成功，從而提早結束下拉
        let scrollHeight = 0;

        // maxTries : 有時候沒法下拉多是網速的緣由
        let maxTries = 5;
        let tried = 0;

        const timer = setInterval(() => {

            // 下拉失敗，提早退出
            // BUG : 若是網速慢的話，這一步會成立~
            // 因此設置一個 maxTried 變量
            if (document.body.scrollHeight === scrollHeight) {
                tried += 1;
                if (tried >= maxTries) {
                    console.log("reached the end, now finished!");
                    clearInterval(timer);
                    resolve();
                }
            }

            scrollHeight = document.body.scrollHeight;
            window.scrollTo(0, scrollHeight);
            window.scrollBy(0, -10);

            // 判斷是否設置了maxScrollTimes
            if (maxScrollTimes) {
                if (currentScrollTimes >= maxScrollTimes) {
                    clearInterval(timer);
                    resolve();
                }
            }

            // 判斷是否設置了maxScrollHeight
            if (maxScrollHeight) {
                if (scrollHeight >= maxScrollHeight) {
                    if (currentScrollTimes >= maxScrollTimes) {
                        clearInterval(timer);
                        resolve();
                    }
                }
            }

            currentScrollTimes += 1;
            // 還原 tried
            tried = 0;
        }, 1000);

    });
};
複製代碼

這裏面有幾個重要的參數：

• interval : 下拉間隔時間，以毫秒爲單位
• maxScrollHeight : 運行頁面下拉最大高度
• maxScrollTimes : 最多下拉多少次（推薦使用，能夠更好控制爬取多少數據）
• maxTries : 下拉不成功時最多重試幾回，好比有時候會由於網絡緣由致使沒能在 interval ms 內成功下拉

把這些替換成你須要的。同時你能夠打開 chrome 的開發者工具運行一下這段 js 腳本。

完整代碼

這段代碼一共也就只有70多行，比較簡陋，情根據本身的實際需求更改。

import os
import time
import json
from urllib.parse import urlsplit
import asyncio
import pyppeteer
from scripts import scripts

BASE_DIR = os.path.dirname(__file__)


async def intercept_request(req):
    """請求過濾"""
    if req.resourceType in ['image', 'media', 'eventsource', 'websocket']:
        await req.abort()
    else:
        await req.continue_()


async def intercept_response(res):
    resourceType = res.request.resourceType
    if resourceType in ['xhr', 'fetch']:
        resp = await res.text()

        url = res.url
        tokens = urlsplit(url)

        folder = BASE_DIR + '/' + 'data/' + tokens.netloc + tokens.path + "/"
        if not os.path.exists(folder):
            os.makedirs(folder, exist_ok=True)
        filename = os.path.join(folder, str(int(time.time())) + '.json')
        with open(filename, 'w', encoding='utf-8') as f:
            f.write(resp)


async def main():
    browser = await pyppeteer.launch({
        # 'headless': False,
        # 'devtools': True
        'executablePath': '/Users/changjiang/apps/Chromium.app/Contents/MacOS/Chromium',
        'args': [
            '--disable-extensions',
            '--hide-scrollbars',
            '--disable-bundled-ppapi-flash',
            '--mute-audio',
            '--no-sandbox',
            '--disable-setuid-sandbox',
            '--disable-gpu',
        ],
        'dumpio': True,
    })
    page = await browser.newPage()

    await page.setRequestInterception(True)
    page.on('request', intercept_request)
    page.on('response', intercept_response)

    await page.setUserAgent('Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 '
                            '(KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36 Edge/16.16299')
    await page.setViewport({'width': 1080, 'height': 960})
    await page.goto('http://yangkeduo.com')
    await page.evaluate(""" () =>{ Object.defineProperties(navigator,{ webdriver:{ get: () => false } }) } """)
    await page.evaluate("你的那一段頁面自動下拉 js 腳本")
    await browser.close()


if __name__ == '__main__':
    asyncio.run(main())

複製代碼

若是你像我同樣真正熱愛計算機科學，喜歡研究底層邏輯，歡迎關注個人微信公衆號：