摘要：爬蟲(chóng)神器，對(duì)加密降維打擊是對(duì)無(wú)頭瀏覽器的封裝。使用等其他無(wú)頭瀏覽器的最大優(yōu)勢(shì)當(dāng)然是對(duì)加密實(shí)行降維打擊，完全無(wú)視加密手段，對(duì)于一些需要登錄的應(yīng)用，也可以模擬點(diǎn)擊然后保存。請(qǐng)求過(guò)濾你的那一段頁(yè)面自動(dòng)下拉腳本

pyppeteer?是對(duì)無(wú)頭瀏覽器?puppeteer的 Python 封裝。無(wú)頭瀏覽器廣泛用于自動(dòng)化測(cè)試，同時(shí)也是一種很好地爬蟲(chóng)思路。

使用 puppeteer（等其他無(wú)頭瀏覽器）的最大優(yōu)勢(shì)當(dāng)然是對(duì) js 加密實(shí)行降維打擊，完全無(wú)視 js 加密手段，對(duì)于一些需要登錄的應(yīng)用，也可以模擬點(diǎn)擊然后保存 cookie。而很多時(shí)候前端的加密是爬蟲(chóng)最難攻克的一部分。當(dāng)然puppeteer也有劣勢(shì)，最大的劣勢(shì)就是相比面向接口爬蟲(chóng)效率很低，就算是無(wú)頭的chromium，那也會(huì)占用相當(dāng)一部分內(nèi)存。另外額外維護(hù)一個(gè)瀏覽器的啟動(dòng)、關(guān)閉也是一種負(fù)擔(dān)。

這篇文章我們來(lái)寫(xiě)一個(gè)簡(jiǎn)單的 demo，爬取拼多多搜索頁(yè)面的數(shù)據(jù)，最終的效果如下：

我們把所有 api 請(qǐng)求的原始數(shù)據(jù)保存下來(lái)：

示例 json 文件如下：

python3.6+

最好是 python3.7，因?yàn)?b>asyncio在 py3.7中加入了很好用的asyncio.run()方法。

安裝pyppeteer

如果安裝有問(wèn)題請(qǐng)去看官方文檔。

python3 -m pip install pyppeteer

安裝 chromium

你懂的，天朝網(wǎng)絡(luò)環(huán)境很復(fù)雜，如果要用pyppeteer自己綁定的chromium，半天都下載不下來(lái)，所以我們要手動(dòng)安裝，然后在程序里面指定executablePath。

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

pyppeteer?的 hello world 程序是前往exmaple.com截個(gè)圖：

import asyncio
from pyppeteer import launch

async def main():
    browser = await launch({
        # Windows 和 Linux 的目錄不一樣，情換成自己對(duì)應(yīng)的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())

launch 瀏覽器，可以傳入一個(gè)字典來(lái)配置幾個(gè)options，比如：

browser = await pyppeteer.launch({
    "headless": False, # 關(guān)閉無(wú)頭模式
    "devtools": True, # 打開(kāi) 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參數(shù)在這里：peter.sh/experiments…

dumpio的作用：把無(wú)頭瀏覽器進(jìn)程的 stderr 核 stdout pip 到主程序，也就是設(shè)置為 True 的話(huà)，chromium console 的輸出就會(huì)在主程序中被打印出來(lái)。

可以通過(guò)page.evaluate形式，例如：

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

我們會(huì)看到這一步非常關(guān)鍵，因?yàn)?b>puppeteer出于政策考慮（這個(gè)詞用的不是很好，就是那個(gè)意思）會(huì)設(shè)置window.navigator.webdriver為true，告訴網(wǎng)站我是一個(gè) webdriver 驅(qū)動(dòng)的瀏覽器。有些網(wǎng)站比較聰明（反爬措施做得比較好），就會(huì)通過(guò)這個(gè)來(lái)判斷對(duì)方是不是爬蟲(chóng)程序。

這等價(jià)于在 devtools 里面輸入那一段 js 代碼。

還可以加載一個(gè) js 文件：

await page.addScriptTag(path=path_to_your_js_file)

通過(guò)注入 js 腳本能完成很多很多有用的操作，比如自動(dòng)下拉頁(yè)面等。

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

intercept_request和intercept_response相當(dāng)于是注冊(cè)的兩個(gè)回調(diào)函數(shù)，在瀏覽器發(fā)出請(qǐng)求和獲取到請(qǐng)求之前指向這兩個(gè)函數(shù)。

比如可以這樣禁止獲取圖片、多媒體資源和發(fā)起 websocket 請(qǐng)求：

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

然后每次獲取到請(qǐng)求之后將內(nèi)容打印出來(lái)（這里只打印了fetch和xhr類(lèi)型response 的內(nèi)容）：

async def intercept_response(res):
    resourceType = res.request.resourceType
    if resourceType in ["xhr", "fetch"]:
        resp = await res.text()
        print(resp)
大家在學(xué)python的時(shí)候肯定會(huì)遇到很多難題，以及對(duì)于新技術(shù)的追求，這里推薦一下我們的Python學(xué)習(xí)扣qun：784758214，這里是python學(xué)習(xí)者聚集地

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

拼多多的搜索界面是一個(gè)無(wú)限下拉的頁(yè)面，我們希望能夠?qū)崿F(xiàn)無(wú)限下拉頁(yè)面，并且能夠控制程序提前退出，不然一直下拉也不好，我們可能并不需要那么多數(shù)據(jù)。

js 腳本

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

        // 允許下滑的最大高度，防止那種可以無(wú)限下拉的頁(yè)面無(wú)法結(jié)束
        const maxScrollHeight = null;

        // 控制下拉次數(shù)
        const maxScrollTimes = null;
        let currentScrollTimes = 0;

        // 記錄上一次scrollHeight，便于判斷此次下拉操作有沒(méi)有成功，從而提前結(jié)束下拉
        let scrollHeight = 0;

        // maxTries : 有時(shí)候無(wú)法下拉可能是網(wǎng)速的原因
        let maxTries = 5;
        let tried = 0;

        const timer = setInterval(() => {

            // 下拉失敗，提前退出
            // BUG : 如果網(wǎng)速慢的話(huà)，這一步會(huì)成立~
            // 所以設(shè)置一個(gè) 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);

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

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

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

    });
};

這里面有幾個(gè)重要的參數(shù)：

interval : 下拉間隔時(shí)間，以毫秒為單位

maxScrollHeight : 運(yùn)行頁(yè)面下拉最大高度

maxScrollTimes : 最多下拉多少次（推薦使用，可以更好控制爬取多少數(shù)據(jù)）

maxTries : 下拉不成功時(shí)最多重試幾次，比如有時(shí)候會(huì)因?yàn)榫W(wǎng)絡(luò)原因?qū)е聸](méi)能在 interval ms 內(nèi)成功下拉

把這些替換成你需要的。同時(shí)你可以打開(kāi) chrome 的開(kāi)發(fā)者工具運(yùn)行一下這段 js 腳本。

這段代碼一共也就只有70多行，比較簡(jiǎn)陋，情根據(jù)自己的實(shí)際需求更改。

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):
    """請(qǐng)求過(guò)濾"""
    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("你的那一段頁(yè)面自動(dòng)下拉 js 腳本")
    await browser.close()

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