使用Python編寫一個(gè)瀏覽器集群框架
這是做什么用的
框架用途
在采集大量新聞網(wǎng)站時(shí),不可避免的遇到動(dòng)態(tài)加載的網(wǎng)站,這給配模版的人增加了很大難度。本來(lái)配靜態(tài)網(wǎng)站只需要兩個(gè)技能點(diǎn):xpath和正則,如果是動(dòng)態(tài)網(wǎng)站的還得抓包,遇到加密的還得js逆向。
所以就需要用瀏覽器渲染這些動(dòng)態(tài)網(wǎng)站,來(lái)減少了配模板的工作難度和技能要求。動(dòng)態(tài)加載的網(wǎng)站在新聞網(wǎng)站里占比很低,需要的硬件資源相對(duì)于一個(gè)人工來(lái)說(shuō)更便宜。
實(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,這種方式比直接集成要好,本來(lái)就是兩個(gè)不同的功能,實(shí)際就應(yīng)該解耦成兩個(gè)單獨(dú)的模塊。不過(guò)聽前輩說(shuō)這東西不太好用,會(huì)有內(nèi)存泄漏的情況,我就沒(méi)測(cè)試它。
自己實(shí)現(xiàn)
原理:在自動(dòng)化瀏覽器中嵌入http服務(wù)實(shí)現(xiàn)http控制瀏覽器。這里我選擇aiohttp+pyppeteer。之前看到有大佬使用go的rod來(lái)做,奈何自己不會(huì)go語(yǔ)言,還是用Python比較順手。
后面會(huì)考慮用playwright重寫一遍,pyppeteer的github說(shuō)此倉(cāng)庫(kù)不常維護(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請(qǐng)求的邏輯。/render.json是用于獲取網(wǎng)頁(yè)的某個(gè)ajax接口響應(yīng)內(nèi)容。有些情況網(wǎng)頁(yè)可能不方便解析,想拿到接口的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)簽頁(yè)數(shù),這種常量一般定義到配置文件里,這里我圖方便就不寫配置文件了。
首先初始化所有的標(biāo)簽頁(yè)放到隊(duì)列里,然后存放在app這個(gè)對(duì)象里,這個(gè)對(duì)象可以在RenderHtmlView類里通過(guò)self.request.app訪問(wèn)到, 到時(shí)候就能控制使用哪個(gè)標(biāo)簽頁(yè)來(lái)訪問(wèn)鏈接
我還初始化了一個(gè)協(xié)程鎖,后面在RenderPngView類里截圖的時(shí)候會(huì)用到,因?yàn)槎鄻?biāo)簽不能同時(shí)截圖,需要加鎖。
超時(shí)停止頁(yè)面繼續(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í)間頁(yè)面明明加載出來(lái)了,但還在轉(zhuǎn)圈,因?yàn)槟硞€(gè)圖片或css等資源訪問(wèn)不到,強(qiáng)制停止加載也不會(huì)影響到網(wǎng)頁(yè)的內(nèi)容。
Page.stopLoading和window.stop()都可以停止頁(yè)面繼續(xù)加載,忘了之前為什么選擇前者了
定義請(qǐng)求參數(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: 訪問(wèn)的鏈接timeout: 超時(shí)時(shí)間wait_util: 頁(yè)面加載完成的標(biāo)識(shí),一般都是domcontentloaded,只有截圖的時(shí)候會(huì)選擇networkidle2,讓網(wǎng)頁(yè)加載全一點(diǎn)。更多的選項(xiàng)的選項(xiàng)請(qǐng)看:Puppeteer waitUntil Optionswait: 頁(yè)面加載完成后等待的時(shí)間,有時(shí)候還得等頁(yè)面的某個(gè)元素加載完成js_name: 預(yù)留的參數(shù),用于在頁(yè)面訪問(wèn)前加載js,目前就只有一個(gè)js(stealth.min.js)用于去瀏覽器特征filters: 過(guò)濾的請(qǐng)求列表, 支持正則。比如有些css請(qǐng)求你不想讓他加載images: 是否加載圖片forbidden_content_types: 禁止加載的資源類型,默認(rèn)是圖片和視頻。所有的類型見(jiàn): resourcetypecache: 是否啟用緩存cookie: 是否在返回結(jié)果里包含cookietext: 是否在返回結(jié)果里包含htmlheaders: 是否在返回結(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è)頁(yè)面。截圖的時(shí)候一般是需要加載圖片的,所以就啟用了圖片加載
怎么使用
多個(gè)標(biāo)簽同時(shí)采集
默認(rèn)是啟動(dòng)了四個(gè)標(biāo)簽頁(yè),這四個(gè)標(biāo)簽頁(yè)可以同時(shí)訪問(wèn)不同鏈接。如果標(biāo)簽頁(yè)過(guò)多可能會(huì)影響性能,不過(guò)開了二三十個(gè)應(yīng)該沒(méi)什么問(wèn)題
請(qǐ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è)異步運(yùn)行的話至少需要4秒
結(jié)果如圖,四個(gè)鏈接只用了4秒多點(diǎn):

攔截指定ajax請(qǐng)求的響應(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())
請(qǐng)求https://spa1.scrape.center/這個(gè)網(wǎng)站并獲取ajax鏈接中包含/api/的接口響應(yīng)數(shù)據(jù),結(jié)果如圖:

請(qǐng)求一個(gè)網(wǎng)站用時(shí)21秒,這是因?yàn)榫W(wǎng)站一直在轉(zhuǎn)圈,其實(shí)要的數(shù)據(jù)已經(jīng)加載完成了,可能是一些圖標(biāo)或者css還在請(qǐng)求。
超時(shí)強(qiáng)制返回
加上timeout參數(shù)后,即使頁(yè)面未加載完成也會(huì)強(qiáng)制停止并返回?cái)?shù)據(jù)。如果這個(gè)時(shí)候已經(jīng)攔截到了ajax請(qǐng)求會(huì)返回ajax響應(yīng)內(nèi)容,不然就是返回空
不過(guò)好像因?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來(lái)請(qǐng)求,后面想了下,其實(shí)都要替換請(qǐng)求和響應(yīng),為什么不直接用scrapy的下載器
完整源代碼
現(xiàn)在還只是個(gè)半成品玩具,還沒(méi)有用于實(shí)際生產(chǎn)中,集群打包也沒(méi)做。有興趣的話可以自己完善一下
如果感興趣的人比較多,后面也會(huì)系統(tǒng)的完善一下,打包成docker和發(fā)布第三方庫(kù)到pypi
github:https://github.com/kanadeblisst00/browser_cluster
以上就是使用Python編寫一個(gè)瀏覽器集群框架的詳細(xì)內(nèi)容,更多關(guān)于Python瀏覽器集群框架的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
Pandas Series如何轉(zhuǎn)換為DataFrame
這篇文章主要介紹了Pandas Series如何轉(zhuǎn)換為DataFrame問(wèn)題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2023-08-08
關(guān)于Python中進(jìn)度條的六個(gè)實(shí)用技巧分享
在項(xiàng)目開發(fā)過(guò)程中加載、啟動(dòng)、下載項(xiàng)目難免會(huì)用到進(jìn)度條,下面這篇文章主要給大家介紹了關(guān)于Python中進(jìn)度條的六個(gè)實(shí)用技巧,文中通過(guò)示例代碼介紹的非常詳細(xì),需要的朋友可以參考下2022-04-04
Django models.py應(yīng)用實(shí)現(xiàn)過(guò)程詳解
這篇文章主要介紹了Django models.py應(yīng)用實(shí)現(xiàn)過(guò)程詳解,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2019-07-07
python如何求數(shù)組連續(xù)最大和的示例代碼
這篇文章主要介紹了python如何求數(shù)組連續(xù)最大和的示例代碼,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2020-02-02
超詳細(xì)注釋之OpenCV實(shí)現(xiàn)視頻實(shí)時(shí)人臉模糊和人臉馬賽克
這篇文章主要介紹了OpenCV實(shí)現(xiàn)視頻實(shí)時(shí)人臉模糊和人臉馬賽克,本文給大家介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2021-09-09

