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