快捷導(dǎo)航

Python如何實現(xiàn)同時進行多個瀏覽器注入

更新時間：2025年03月04日 09:19:23 作者：百錦再@新空間代碼工作室

瀏覽器注入是一種技術(shù),允許我們在運行中的瀏覽器進程中注入代碼,以便執(zhí)行特定的任務(wù),下面我們就來看看Python如何實現(xiàn)同時進行多個瀏覽器注入吧

瀏覽器注入是一種技術(shù)，允許我們在運行中的瀏覽器進程中注入代碼，以便執(zhí)行特定的任務(wù)。在某些場景下，如自動化測試、Web數(shù)據(jù)抓取、安全性研究或擴展瀏覽器功能，我們可能需要同時對多個瀏覽器進行注入，以提高效率和并行能力。

方案概述

本方案基于 Python 語言，結(jié)合 Pyppeteer、Selenium 和 CDP（Chrome DevTools Protocol）來實現(xiàn)同時進行多個瀏覽器的注入。該方案支持：

多進程并行注入：利用 Python 的 multiprocessing 庫并行控制多個瀏覽器實例。
JavaScript 代碼注入：通過 Chrome 的 evaluate 方法動態(tài)執(zhí)行 JavaScript 代碼。
瀏覽器進程隔離：為每個瀏覽器分配獨立的 Session，避免相互干擾。
數(shù)據(jù)通信與監(jiān)控：通過 WebSocket 監(jiān)聽瀏覽器的執(zhí)行狀態(tài)，以便管理多個瀏覽器實例。

第一部分：環(huán)境配置

為了實現(xiàn)本方案，我們需要安裝必要的依賴項：

pip install selenium pyppeteer

此外，我們需要 Chrome 或 Chromium 作為瀏覽器執(zhí)行環(huán)境，并且確保 Chrome 允許遠程調(diào)試模式。

第二部分：瀏覽器注入的核心邏輯

1. 使用 Pyppeteer 進行瀏覽器注入

Pyppeteer 是 Chrome Puppeteer 的 Python 版本，它允許我們以編程方式控制 Chromium，并執(zhí)行 JavaScript 代碼。

(1) 啟動多個瀏覽器實例

使用 Pyppeteer，我們可以同時啟動多個瀏覽器實例：

import asyncio
from pyppeteer import launch

async def inject_js(browser_id, url, script):
    print(f"[{browser_id}] 啟動瀏覽器...")
    browser = await launch(headless=False, args=['--no-sandbox'])
    page = await browser.newPage()
    await page.goto(url)
    
    print(f"[{browser_id}] 注入 JavaScript 代碼...")
    result = await page.evaluate(script)
    
    print(f"[{browser_id}] 執(zhí)行結(jié)果：{result}")
    await browser.close()

async def main():
    url = "https://www.example.com"
    script = "document.body.style.backgroundColor = 'red'; '注入成功';"
    
    tasks = [inject_js(i, url, script) for i in range(5)]  # 5 個瀏覽器實例
    await asyncio.gather(*tasks)

asyncio.run(main())

說明：

launch 用于啟動瀏覽器。
newPage 創(chuàng)建新頁面并訪問指定 URL。
evaluate 方法執(zhí)行 JavaScript 代碼，實現(xiàn)注入。

2. 使用 Selenium 進行瀏覽器注入

Selenium 是另一種常見的瀏覽器自動化工具，我們可以結(jié)合它進行 JavaScript 注入。

(1) 啟動多個 Selenium 瀏覽器實例

from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
import multiprocessing

def inject_js_selenium(browser_id, url, script):
    print(f"[{browser_id}] 啟動瀏覽器...")
    
    chrome_options = Options()
    chrome_options.add_argument("--headless")
    
    driver = webdriver.Chrome(service=Service("/path/to/chromedriver"), options=chrome_options)
    driver.get(url)
    
    print(f"[{browser_id}] 注入 JavaScript 代碼...")
    driver.execute_script(script)
    
    print(f"[{browser_id}] 注入完成")
    driver.quit()

if __name__ == "__main__":
    url = "https://www.example.com"
    script = "document.body.style.backgroundColor = 'blue';"
    
    processes = []
    for i in range(5):  # 5個瀏覽器實例
        p = multiprocessing.Process(target=inject_js_selenium, args=(i, url, script))
        p.start()
        processes.append(p)
    
    for p in processes:
        p.join()

說明：

webdriver.Chrome 用于創(chuàng)建 Chrome 實例。
execute_script 方法執(zhí)行 JavaScript 代碼，實現(xiàn)注入。
multiprocessing 用于并行運行多個瀏覽器實例。

3. 使用 CDP 進行 DevTools 遠程調(diào)試注入

CDP（Chrome DevTools Protocol）提供了一種更底層的方式來控制瀏覽器。

import asyncio
from pyppeteer import launch

async def inject_cdp(browser_id, url, script):
    browser = await launch(headless=False, args=['--remote-debugging-port=9222'])
    pages = await browser.pages()
    page = pages[0] if pages else await browser.newPage()
    
    await page.goto(url)
    print(f"[{browser_id}] 通過 CDP 注入 JavaScript 代碼...")
    
    cdp_session = await page.target.createCDPSession()
    await cdp_session.send("Runtime.evaluate", {"expression": script})
    
    print(f"[{browser_id}] CDP 注入完成")
    await browser.close()

async def main():
    url = "https://www.example.com"
    script = "document.title = 'CDP 注入成功';"
    
    tasks = [inject_cdp(i, url, script) for i in range(3)]  # 3 個瀏覽器實例
    await asyncio.gather(*tasks)

asyncio.run(main())

第三部分：優(yōu)化與擴展

1. 增強并發(fā)性

可以使用 ThreadPoolExecutor 或 asyncio.Queue 來提高任務(wù)調(diào)度的效率。

import concurrent.futures

def parallel_execution():
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        futures = {executor.submit(inject_js_selenium, i, "https://www.example.com", "console.log('執(zhí)行JS')") for i in range(5)}
        for future in concurrent.futures.as_completed(futures):
            print(future.result())

parallel_execution()