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

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

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

使用 puppeteer（等其他無頭瀏覽器）的最大優勢當然是 對 js 加密實行降維打擊 ，完全無視 js 加密手段，對於一些需要登錄的應用，也可以模擬點擊然後保存 cookie。 而很多時候前端的加密是爬蟲最難攻克的一部分 。當然puppeteer也有劣勢，最大的劣勢就是相比面向接口爬蟲效率很低，就算是無頭的chromium，那也會佔用相當一部分內存。另外額外維護一個瀏覽器的啟動、關閉也是一種負擔。

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

我們把所有 api 請求的原始數據保存下來：

Python學習交流群：1004391443，這裡是python學習者聚集地，有大牛答疑，有資源共享！小編也準備了一份python學習資料，有想學習python編程的，或是轉行，或是大學生，還有工作中想提升自己能力的，正在學習的小夥伴歡迎加入學習。

示例 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 驅動的瀏覽器。有些網站比較聰明（反爬措施做得比較好），就會通過這個來判斷對方是不是爬蟲程序。

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

一共有哪些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())