使用Python編寫一個(gè)瀏覽器集群框架

更新時(shí)間：2023年10月27日 14:42:47 作者：qwertyuiop1822

這篇文章主要為大家詳細(xì)介紹了如何使用Python編寫一個(gè)瀏覽器集群框架,文中的示例代碼講解詳細(xì),具有一定的學(xué)習(xí)價(jià)值,感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下

這是做什么用的

框架用途

在采集大量新聞網(wǎng)站時(shí)，不可避免的遇到動(dòng)態(tài)加載的網(wǎng)站，這給配模版的人增加了很大難度。本來配靜態(tài)網(wǎng)站只需要兩個(gè)技能點(diǎn)：xpath和正則，如果是動(dòng)態(tài)網(wǎng)站的還得抓包，遇到加密的還得js逆向。

所以就需要用瀏覽器渲染這些動(dòng)態(tài)網(wǎng)站，來減少了配模板的工作難度和技能要求。動(dòng)態(tài)加載的網(wǎng)站在新聞網(wǎng)站里占比很低，需要的硬件資源相對于一個(gè)人工來說更便宜。

實(shí)現(xiàn)方式

采集框架使用瀏覽器渲染有兩種方式，一種是直接集成到框架，類似GerapyPyppeteer，這個(gè)項(xiàng)目你看下源代碼就會(huì)發(fā)現(xiàn)寫的很粗糙，它把瀏覽器放在_process_request方法里啟動(dòng)，然后采集完一個(gè)鏈接再關(guān)閉瀏覽器，大部分時(shí)間都浪費(fèi)在瀏覽器的啟動(dòng)和關(guān)閉上，而且采集多個(gè)鏈接會(huì)打開多個(gè)瀏覽器搶占資源。

另一種則是將瀏覽器渲染獨(dú)立成一個(gè)服務(wù)，類似scrapy-splash，這種方式比直接集成要好，本來就是兩個(gè)不同的功能，實(shí)際就應(yīng)該解耦成兩個(gè)單獨(dú)的模塊。不過聽前輩說這東西不太好用，會(huì)有內(nèi)存泄漏的情況，我就沒測試它。

自己實(shí)現(xiàn)

原理：在自動(dòng)化瀏覽器中嵌入http服務(wù)實(shí)現(xiàn)http控制瀏覽器。這里我選擇aiohttp+pyppeteer。之前看到有大佬使用go的rod來做，奈何自己不會(huì)go語言，還是用Python比較順手。

后面會(huì)考慮用playwright重寫一遍，pyppeteer的github說此倉庫不常維護(hù)了，建議使用playwright。

開始寫代碼

web服務(wù)

from aiohttp import web

app = web.Application()
app.router.add_view('/render.html', RenderHtmlView)
app.router.add_view('/render.png', RenderPngView)
app.router.add_view('/render.jpeg', RenderJpegView)
app.router.add_view('/render.json', RenderJsonView)

然后在RenderHtmlView類中寫/render.html請求的邏輯。/render.json是用于獲取網(wǎng)頁的某個(gè)ajax接口響應(yīng)內(nèi)容。有些情況網(wǎng)頁可能不方便解析，想拿到接口的json響應(yīng)數(shù)據(jù)。

初始化瀏覽器

瀏覽器只需要初始化一次，所以啟動(dòng)放到on_startup，關(guān)閉放到on_cleanup

c = LaunchChrome()
app.on_startup.append(c.on_startup_tasks)
app.on_cleanup.append(c.on_cleanup_tasks)

其中on_startup_tasks和on_cleanup_tasks方法如下：

async def on_startup_tasks(self, app: web.Application) -> None:
		page_count = 4
		await asyncio.create_task(self._launch())
		app["browser"] = self.browser
		tasks = [asyncio.create_task(self.launch_tab()) for _ in range(page_count-1)]
		await asyncio.gather(*tasks)
		queue = asyncio.Queue(maxsize=page_count+1)
		for i in await self.browser.pages():
				await queue.put(i)
		app["pages_queue"] = queue
		app["screenshot_lock"] = asyncio.Lock()

async def on_cleanup_tasks(self, app: web.Application) -> None:
		await self.browser.close()

page_count為初始化的標(biāo)簽頁數(shù)，這種常量一般定義到配置文件里，這里我圖方便就不寫配置文件了。

首先初始化所有的標(biāo)簽頁放到隊(duì)列里，然后存放在app這個(gè)對象里，這個(gè)對象可以在RenderHtmlView類里通過self.request.app訪問到, 到時(shí)候就能控制使用哪個(gè)標(biāo)簽頁來訪問鏈接

我還初始化了一個(gè)協(xié)程鎖，后面在RenderPngView類里截圖的時(shí)候會(huì)用到，因?yàn)槎鄻?biāo)簽不能同時(shí)截圖，需要加鎖。

超時(shí)停止頁面繼續(xù)加載

async def _goto(self, page: Optional[Page], options: AjaxPostData) -> Dict:
		try:
				await page.goto(options.url, 
						waitUntil=options.wait_util, timeout=options.timeout*1000)
		except PPTimeoutError:
				#await page.evaluate('() => window.stop()')
				await page._client.send("Page.stopLoading")
		finally:
				page.remove_all_listeners("request")

有時(shí)間頁面明明加載出來了，但還在轉(zhuǎn)圈，因?yàn)槟硞€(gè)圖片或css等資源訪問不到，強(qiáng)制停止加載也不會(huì)影響到網(wǎng)頁的內(nèi)容。

Page.stopLoading和window.stop()都可以停止頁面繼續(xù)加載，忘了之前為什么選擇前者了

定義請求參數(shù)

class HtmlPostData(BaseModel):
    url: str
    timeout: float = 30
    wait_util: str = "domcontentloaded"
    wait: float = 0   
    js_name: str = "" 
    filters: List[str] = [] 
    images: bool = 0  
    forbidden_content_types: List[str] = ["image", "media"]
    cache: bool = 1 
    cookie: bool = 0 
    text: bool = 1 
		headers: bool = 1

url: 訪問的鏈接
timeout: 超時(shí)時(shí)間
wait_util: 頁面加載完成的標(biāo)識(shí)，一般都是domcontentloaded，只有截圖的時(shí)候會(huì)選擇networkidle2，讓網(wǎng)頁加載全一點(diǎn)。更多的選項(xiàng)的選項(xiàng)請看：Puppeteer waitUntil Options
wait: 頁面加載完成后等待的時(shí)間，有時(shí)候還得等頁面的某個(gè)元素加載完成
js_name: 預(yù)留的參數(shù)，用于在頁面訪問前加載js，目前就只有一個(gè)js(stealth.min.js)用于去瀏覽器特征
filters: 過濾的請求列表, 支持正則。比如有些css請求你不想讓他加載
images: 是否加載圖片
forbidden_content_types: 禁止加載的資源類型，默認(rèn)是圖片和視頻。所有的類型見: resourcetype
cache: 是否啟用緩存
cookie: 是否在返回結(jié)果里包含cookie
text: 是否在返回結(jié)果里包含html
headers: 是否在返回結(jié)果里包含headers

圖片的參數(shù)

class PngPostData(HtmlPostData):
    render_all: int = 0
    text: bool = 0
    images: bool = 1
    forbidden_content_types: List[str] = []
    wait_util: str = "networkidle2"

參數(shù)和html的基本一樣，增加了一個(gè)render_all用于是否截取整個(gè)頁面。截圖的時(shí)候一般是需要加載圖片的，所以就啟用了圖片加載

怎么使用

多個(gè)標(biāo)簽同時(shí)采集

默認(rèn)是啟動(dòng)了四個(gè)標(biāo)簽頁，這四個(gè)標(biāo)簽頁可以同時(shí)訪問不同鏈接。如果標(biāo)簽頁過多可能會(huì)影響性能，不過開了二三十個(gè)應(yīng)該沒什么問題

請求例子如下：

import sys
import asyncio
import aiohttp

if sys.platform == 'win32':
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

async def get_sign(session, delay):
    url = f"http://www.httpbin.org/delay/{delay}"
    api = f'http://127.0.0.1:8080/render.html?url={url}'
    async with session.get(api) as resp:
        data = await resp.json()
        print(url, data.get("status"))
        return data

async def main():
    headers = {
        "Content-Type": "application/json",
        'user-agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.198 Safari/537.36',
    }
    loop = asyncio.get_event_loop()
    t = loop.time()
    async with aiohttp.ClientSession(headers=headers) as session:
        tasks = [asyncio.create_task(get_sign(session, i)) for i in range(1, 5)]
        await asyncio.gather(*tasks)
    print("耗時(shí): ", loop.time()-t)

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

http://www.httpbin.org/delay后面跟的數(shù)字是多少，網(wǎng)站就會(huì)多少秒后返回。所以如果同步運(yùn)行的話至少需要1+2+3+4秒，而多標(biāo)簽頁異步運(yùn)行的話至少需要4秒

結(jié)果如圖，四個(gè)鏈接只用了4秒多點(diǎn)：

攔截指定ajax請求的響應(yīng)

import json
import sys
import asyncio
import aiohttp

if sys.platform == 'win32':
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

async def get_sign(session, url):
    api = f'http://127.0.0.1:8080/render.json'
    data = {
        "url": url,
        "xhr": "/api/", # 攔截接口包含/api/的響應(yīng)并返回
        "cache": 0,
        "filters": [".png", ".jpg"]
    }
    async with session.post(api, data=json.dumps(data)) as resp:
        data = await resp.json()
        print(url, data)
        return data

async def main():
    urls = ["https://spa1.scrape.center/"]
    headers = {
        "Content-Type": "application/json",
        'user-agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.198 Safari/537.36',
    }
    loop = asyncio.get_event_loop()
    t = loop.time()
    async with aiohttp.ClientSession(headers=headers) as session:
        tasks = [asyncio.create_task(get_sign(session, url)) for url in urls]
        await asyncio.gather(*tasks)
    print(loop.time()-t)

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

請求https://spa1.scrape.center/這個(gè)網(wǎng)站并獲取ajax鏈接中包含/api/的接口響應(yīng)數(shù)據(jù)，結(jié)果如圖：

請求一個(gè)網(wǎng)站用時(shí)21秒，這是因?yàn)榫W(wǎng)站一直在轉(zhuǎn)圈，其實(shí)要的數(shù)據(jù)已經(jīng)加載完成了，可能是一些圖標(biāo)或者css還在請求。

超時(shí)強(qiáng)制返回

加上timeout參數(shù)后，即使頁面未加載完成也會(huì)強(qiáng)制停止并返回?cái)?shù)據(jù)。如果這個(gè)時(shí)候已經(jīng)攔截到了ajax請求會(huì)返回ajax響應(yīng)內(nèi)容，不然就是返回空

不過好像因?yàn)橛芯彺?，現(xiàn)在時(shí)間不到1秒就返回了

截圖

import json
import sys
import asyncio
import base64
import aiohttp

if sys.platform == 'win32':
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

async def get_sign(session, url, name):
    api = f'http://127.0.0.1:8080/render.png'
    data = {
        "url": url,
        #"render_all": 1,
        "images": 1,
        "cache": 1,
        "wait": 1 
    }
    async with session.post(api, data=json.dumps(data)) as resp:
        data = await resp.json()
        if data.get('image'):
            image_bytes = base64.b64decode(data["image"])
            with open(name, 'wb') as f:
                f.write(image_bytes)
            print(url, name, len(image_bytes))
        return data

async def main():
    urls = [
        "https://www.baidu.com/s?ie=utf-8&f=8&rsv_bp=1&tn=44004473_102_oem_dg&wd=%E5%9B%BE%E7%89%87&rn=50",
        "https://www.toutiao.com/article/7145668657396564518/",
        "https://new.qq.com/rain/a/NEW2022092100053400",
        "https://new.qq.com/rain/a/DSG2022092100053300"
    ]
    headers = {
        "Content-Type": "application/json",
        'user-agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.198 Safari/537.36',
    }
    loop = asyncio.get_event_loop()
    t = loop.time()
    async with aiohttp.ClientSession(headers=headers) as session:
        tasks = [asyncio.create_task(get_sign(session, url, f"{n}.png")) for n,url in enumerate(urls)]
        await asyncio.gather(*tasks)
    print(loop.time()-t)


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

集成到scrapy

import json
import logging
from scrapy.exceptions import NotConfigured

logger = logging.getLogger(__name__)

class BrowserMiddleware(object):
    def __init__(self, browser_base_url: str):
        self.browser_base_url = browser_base_url
        self.logger = logger
        
    @classmethod
    def from_crawler(cls, crawler):
        s = crawler.settings
        browser_base_url = s.get('PYPPETEER_CLUSTER_URL')
        if not browser_base_url:
            raise NotConfigured
        o = cls(browser_base_url)
        return o
    
    def process_request(self, request, spider):
        if "browser_options" not in request.meta or request.method != "GET":
            return
        browser_options = request.meta["browser_options"]
        url = request.url
        browser_options["url"] = url
        uri = browser_options.get('browser_uri', "/render.html")
        browser_url = self.browser_base_url.rstrip('/') + '/' + uri.lstrip('/')
        new_request = request.replace(
            url=browser_url,
            method='POST',
            body=json.dumps(browser_options)
        )
        new_request.meta["ori_url"] = url
        return new_request

    def process_response(self, request, response, spider):
        if "browser_options" not in request.meta or "ori_url" not in request.meta:
            return response
        try:
            datas = json.loads(response.text)
        except json.decoder.JSONDecodeError:
            return response.replace(url=url, status=500)
        datas = self.deal_datas(datas)
        url = request.meta["ori_url"]
        new_response = response.replace(url=url, **datas)
        return new_response
    
    def deal_datas(self, datas: dict) -> dict:
        status = datas["status"]
        text: str = datas.get('text') or datas.get('content')
        headers = datas.get('headers')
        response = {
            "status": status,
            "headers": headers,
            "body": text.encode()
        }
        return response

開始想用aiohttp來請求，后面想了下，其實(shí)都要替換請求和響應(yīng)，為什么不直接用scrapy的下載器