基于 Python 实现多浏览器并行注入的实战方案

浏览器注入允许我们在运行中的浏览器进程中执行特定代码。在自动化测试、数据抓取或安全研究中，往往需要同时对多个浏览器实例进行操作以提升效率。本方案基于 Python，结合 Pyppeteer、Selenium 和 CDP 协议，实现多浏览器的并行注入。

环境配置

首先安装必要的依赖项：

pip install selenium pyppeteer

确保已安装 Chrome 或 Chromium，并开启远程调试模式支持。

核心实现

使用 Pyppeteer 进行注入

Pyppeteer 是 Puppeteer 的 Python 移植版，适合以编程方式控制 Chromium。

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}] 执行结果：{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)]
    await asyncio.gather(*tasks)

asyncio.run(main())

注意：launch 方法用于启动浏览器，evaluate 则是执行 JS 的关键。实际运行时需注意沙箱参数 --no-sandbox 在某些 Linux 环境下的兼容性。

使用 Selenium 进行注入

Selenium 是经典的自动化工具，配合 multiprocessing 可实现多进程并行。

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):
        p = multiprocessing.Process(target=inject_js_selenium, args=(i, url, script))
        p.start()
        processes.append(p)
    for p in processes:
        p.join()

这里使用 execute_script 执行 JS，并通过 multiprocessing 管理进程。记得将 /path/to/chromedriver 替换为实际路径。

使用 CDP 进行远程调试注入

CDP 提供了更底层的控制能力。

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)]
    await asyncio.gather(*tasks)

asyncio.run(main())

优化与扩展

若需进一步提升并发效率，可考虑使用 ThreadPoolExecutor 或 asyncio.Queue 来调度任务。例如：

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('执行 JS')") for i in range(5)}
        for future in concurrent.futures.as_completed(futures):
            print(future.result())

parallel_execution()

此外，执行结果可持久化存储至 MongoDB 或 Redis，便于后续分析。

总结

本方案通过 Pyppeteer、Selenium 和 CDP 三种方式，实现了多浏览器的 JavaScript 并行注入。核心在于利用 Python 的多进程与异步特性，有效隔离浏览器 Session，避免相互干扰。无论是自动化测试还是数据采集，这套流程都能显著提升执行效率。