Python如何实现同时进行多个浏览器注入

pip install selenium 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}] 执行结果：{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())

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()

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())

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()