Pyppeteer
引言
Selenium 在被使用的時候有個麻煩事,就是環境的相關配置,得安裝好相關瀏覽器,比如 Chrome、Firefox 等等,然后還要到官方網站去下載對應的驅動
最重要的還需要安裝對應的 Python Selenium 庫,確實是不是很方便,另外如果要做大規模部署的話,環境配置的一些問題也是個頭疼的事情
Pyppeteer簡介
注意,講解的模塊叫做 Pyppeteer,不是 Puppeteer。
Puppeteer 是 Google 基於 Node.js 開發的一個工具,有了它我們可以通過 JavaScript 來控制 Chrome 瀏覽器的一些操作,當然也可以用作網絡爬蟲上,其 API 極其完善,功能非常強大。
而 Pyppeteer 又是什么呢?它實際上是 Puppeteer 的 Python 版本的實現,但他不是 Google 開發的,是一位來自於日本的工程師依據 Puppeteer 的一些功能開發出來的非官方版本。
在 Pyppetter 中,實際上它背后也是有一個類似 Chrome 瀏覽器的 Chromium 瀏覽器在執行一些動作進行網頁渲染,首先說下 Chrome 瀏覽器和 Chromium 瀏覽器的淵源
Chromium 是谷歌為了研發 Chrome 而啟動的項目,是完全開源的。二者基於相同的源代碼構建,Chrome 所有的新功能都會先在 Chromium 上實現,待驗證穩定后才會移植,因此 Chromium 的版本更新頻率更高,也會包含很多新的功能,但作為一款獨立的瀏覽器,Chromium 的用戶群體要小眾得多。兩款瀏覽器“同根同源”,它們有着同樣的 Logo,但配色不同,Chrome 由藍紅綠黃四種顏色組成,而 Chromium 由不同深度的藍色構成。
Pyppeteer 就是依賴於 Chromium 這個瀏覽器來運行的。
那么有了 Pyppeteer 之后,我們就可以免去那些繁瑣的環境配置等問題。
另外 Pyppeteer 是基於 Python 的新特性 async 實現的,所以它的一些執行也支持異步操作,效率相對於 Selenium 來說也提高了。
環境安裝
- 由於 Pyppeteer 采用了 Python 的 async 機制,所以其運行要求的 Python 版本為 3.5 及以上
- pip install pyppeteer
如果第一次運行的時候,Chromium 瀏覽器沒有安裝,那么程序會幫我們自動安裝和配置,就免去了繁瑣的環境配置等工作。當然也可以使用 pyppeteer-install 命令就會自動下載對應的最新版本 chromium 瀏覽器到 pyppeteer 的默認位置
chromium 下載完后會在此目錄 C:\Users\Administrator\AppData\Local\pyppeteer\pyppeteer\local-chromium\575458
因為是國外網站,所以比較慢 先去國內源下載自己需要的版本
地址:https://npm.taobao.org/mirrors/chromium-browser-snapshots/
Pyppeteer 是一款非常高效的 web 自動化測試工具,由於 Pyppeteer 是基於 asyncio 構建的,它的所有 屬性 和方法 幾乎都是 coroutine (協程) 對象,因此在構建異步程序的時候非常方便,天生就支持異步運行。
程序構建的基本思路是新建 一個 browser 瀏覽器 和 一個 頁面 page。
看下面這段代碼,在 main 函數中,先是建立一個瀏覽器對象,然后打開新的標簽頁,訪問百度主頁,對當前頁面截圖並保存為“example.png”,最后關閉瀏覽器。前文也提到過,pyppeteer 是基於 asyncio 構建的,所以在使用的時候需要用到 async/await 結構
示例
爬取 http://quotes.toscrape.com/js/ 全部頁面數據
import asyncio from pyppeteer import launch from lxml import etree async def main(): browser = await launch(headless =False) page = await browser.newPage() await page.goto('http://quotes.toscrape.com/js/') page_text = await page.content() tree = etree.HTML(page_text) div_list = tree.xpath('//div[@class="quote"]/span/text()') # print(len(div_list)) for _ in div_list: print(_) await browser.close() asyncio.get_event_loop().run_until_complete(main())
運行上面這段代碼會發現並沒有瀏覽器彈出運行,這是因為 Pyppeteer 默認使用的是無頭瀏覽器,如果想要瀏覽器顯示,需要在launch 函數中設置參數 “headless =False”
解釋:launch 方法會新建一個 Browser 對象,然后賦值給 browser,然后調用 newPage 方法相當於瀏覽器中新建了一個選項卡,同時新建了一個 page 對象。
然后 page 對象調用了 goto 方法就相當於在瀏覽器中輸入了這個 URL,瀏覽器跳轉到了對應的頁面進行加載,加載完成之后再調用 content 方法,返回當前瀏覽器頁面的源代碼。
然后進一步地,我們用 pyquery 進行同樣地解析,就可以得到 JavaScript 渲染的結果了。
在這個過程中,我們沒有配置 Chrome 瀏覽器,沒有配置瀏覽器驅動,免去了一些繁瑣的步驟,同樣達到了 Selenium 的效果,還實現了異步抓取
詳情用法
開啟瀏覽器
- 調用 launch 方法即可,相關參數介紹:
- ignoreHTTPSErrors (bool): 是否要忽略 HTTPS 的錯誤,默認是 False。
- headless (bool): 是否啟用 Headless 模式,即無界面模式,如果 devtools 這個參數是 True 的話,那么該參數就會被設置為 False,否則為 True,即默認是開啟無界面模式的。
- executablePath (str): 可執行文件的路徑,如果指定之后就不需要使用默認的 Chromium 了,可以指定為已有的 Chrome 或 Chromium。
- args (List[str]): 在執行過程中可以傳入的額外參數。
- devtools (bool): 是否為每一個頁面自動開啟調試工具,默認是 False。如果這個參數設置為 True,那么 headless 參數就會無效,會被強制設置為 False。
關閉提示條:”Chrome 正受到自動測試軟件的控制”,這個提示條有點煩,那咋關閉呢?這時候就需要用到 args 參數了,禁用操作如下:
browser = await launch(headless=False, args=['--disable-infobars'])
處理頁面顯示問題:訪問淘寶首頁
import asyncio from pyppeteer import launch async def main(): browser = await launch(headless =False) page = await browser.newPage() await page.goto('https://www.taobao.com') await asyncio.sleep(3) asyncio.get_event_loop().run_until_complete(main())
發現頁面顯示出現了問題,需要手動調用setViewport方法設置顯示頁面的長寬像素。設置如下:
import asyncio from pyppeteer import launch async def main(): browser = await launch(headless =False) page = await browser.newPage() # 設置頁面視圖大小 await page.setViewport({'width':1280, 'height': 800}) await page.goto('https://www.taobao.com') await asyncio.sleep(3) asyncio.get_event_loop().run_until_complete(main())
執行js程序:拖動滾輪。調用evaluate方法。
import asyncio from pyppeteer import launch async def main(): browser = await launch(headless =False) page = await browser.newPage() # 設置頁面視圖大小 await page.setViewport({'width':1280, 'height': 800}) await page.setJavaScriptEnabled(enabled=True) await page.goto('https://movie.douban.com/typerank?type_name=%E5%8A%A8%E4%BD%9C&type=5&interval_id=100:90&action=') await asyncio.sleep(3) # evaluate 可以返回 js 程序的返回值 dimensions = await page.evaluate('window.scrollTo(0,document.body.scrollHeight)') await asyncio.sleep(3) print(dimensions) await browser.close() asyncio.get_event_loop().run_until_complete(main())
規避webdriver檢測:
import asyncio from pyppeteer import launch async def main(): browser = await launch(headless=False, args=['--disable-infobars']) page = await browser.newPage() await page.setViewport({'width': 1280, 'height': 800}) await page.goto('https://login.taobao.com/member/login.jhtml?redirectURL=https://www.taobao.com/') await asyncio.sleep(3) await page.evaluate('''() =>{ Object.defineProperties(navigator,{ webdriver:{ get: () => false } }) }''') await asyncio.sleep(3) await browser.close() asyncio.get_event_loop().run_until_complete(main())
UA 偽裝
await self.page.setUserAgent('xxx')
節點交互
import asyncio from pyppeteer import launch async def main(): browser = await launch(headless=False, args=['--disable-infobars']) page = await browser.newPage() await page.setViewport({'width': 1280, 'height': 800}) await page.goto('https://www.baidu.com') # 節點交互 await page.type('#kw', '鄧紫棋', {'delay': 100}) # delay 延遲 await asyncio.sleep(3) await page.click('#su') # 使用選擇器選中標簽進行點擊 alist = await page.querySelectorAll('.s_tab_inner > a') a = alist[3] await a.click() await asyncio.sleep(3) await browser.close() asyncio.get_event_loop().run_until_complete(main())
# 示例 import asyncio from pyppeteer import launch from lxml import etree async def main(): browser = await launch(headless=False) page = await browser.newPage() await page.setViewport(viewport={'width': 1280, 'height': 800}) await page.goto('https://www.toutiao.com') await asyncio.sleep(2) page_text = await page.content() page2 = await browser.newPage() await page2.setViewport(viewport={'width': 1280, 'height': 800}) await page2.goto('https://news.163.com/domestic/') await page2.evaluate('window.scrollTo(0,document.body.scrollHeight)') page_text1 = await page2.content() await browser.close() return {'wangyi': page_text1, 'toutiao': page_text} def parse(task): content_dic = task.result() wangyi = content_dic['wangyi'] toutiao = content_dic['toutiao'] tree = etree.HTML(toutiao) a_list = tree.xpath('//div[@class="title-box"]/a') for a in a_list: title = a.xpath('./text()')[0] print(f'toutiao: {title}') tree = etree.HTML(wangyi) div_list = tree.xpath('//div[@class="data_row news_article clearfix "]') print(len(div_list)) for div in div_list: title = div.xpath('.//div[@class="news_title"]/h3/a/text()')[0] print(f'wangyi: {title}') tasks = [] task1 = asyncio.ensure_future(main()) task1.add_done_callback(parse) tasks.append(task1) asyncio.get_event_loop().run_until_complete(asyncio.wait(tasks))