1. 程式人生 > >Python爬蟲神器pyppeteer,對 js 加密降維打擊

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

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

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())
大家在學python的時候肯定會遇到很多難題,以及對於新技術的追求,這裡推薦一下我們的Python學習扣qun:784,758,214,這裡是python學習者聚集地!

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.webdrivertrue,告訴網站我是一個 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_requestintercept_response相當於是註冊的兩個回撥函式,在瀏覽器發出請求和獲取到請求之前指向這兩個函式。

比如可以這樣禁止獲取圖片、多媒體資源和發起 websocket 請求:

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

然後每次獲取到請求之後將內容打印出來(這裡只打印了fetchxhr型別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);

    });
};
大家在學python的時候肯定會遇到很多難題,以及對於新技術的追求,這裡推薦一下我們的Python學習扣qun:784758214,這裡是python學習者聚集地!

這裡面有幾個重要的引數:

  • 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.r