Playwright 与 Selenium

from playwright.sync_api import sync_playwright

PROXY = {
    "server": "http://GATEWAY_HOST:58971",
    "username": "USERNAME-country-US-session-job1-time-30",
    "password": "PASSWORD",
}

with sync_playwright() as playwright:
    browser = playwright.chromium.launch(headless=True)
    context = browser.new_context(proxy=PROXY)
    page = context.new_page()

    page.goto("https://api.ipify.org?format=json")
    print(page.inner_text("body"))

    context.close()
    browser.close()

from playwright.sync_api import sync_playwright

GATEWAY = "http://GATEWAY_HOST:58971"
BASE_USER = "USERNAME"
PASSWORD = "PASSWORD"


def proxy_for(session_id: str, country: str = "US", minutes: int = 30) -> dict:
    return {
        "server": GATEWAY,
        # session 值只允许 [A-Za-z0-9.-]，别用下划线
        "username": f"{BASE_USER}-country-{country}-session-{session_id}-time-{minutes}",
        "password": PASSWORD,
    }


with sync_playwright() as playwright:
    browser = playwright.chromium.launch(headless=True)

    for index in range(4):
        context = browser.new_context(
            proxy=proxy_for(f"ctx.{index}"),
            locale="en-US",
            timezone_id="America/Los_Angeles",
            viewport={"width": 1440, "height": 900},
        )
        page = context.new_page()
        page.goto("https://api.ipify.org", wait_until="domcontentloaded")
        print(index, page.inner_text("body"))
        context.close()

    browser.close()

import asyncio

from playwright.async_api import async_playwright


async def scrape(session_id: str, url: str) -> str:
    async with async_playwright() as playwright:
        browser = await playwright.chromium.launch(headless=True)
        context = await browser.new_context(proxy={
            "server": "http://GATEWAY_HOST:58971",
            "username": f"USERNAME-country-US-session-{session_id}-time-30",
            "password": "PASSWORD",
        })
        page = await context.new_page()
        await page.goto(url, wait_until="domcontentloaded")
        text = await page.inner_text("body")
        await browser.close()
        return text


async def main() -> None:
    results = await asyncio.gather(*(
        scrape(f"job.{i}", "https://api.ipify.org") for i in range(4)
    ))
    for result in results:
        print(result)


asyncio.run(main())

import { chromium } from 'playwright';

const browser = await chromium.launch({ headless: true });

const context = await browser.newContext({
  proxy: {
    server: 'http://GATEWAY_HOST:58971',
    username: 'USERNAME-country-US-session-job1-time-30',
    password: 'PASSWORD',
  },
  locale: 'en-US',
  timezoneId: 'America/Los_Angeles',
});

const page = await context.newPage();
await page.goto('https://api.ipify.org?format=json');
console.log(await page.innerText('body'));

await browser.close();

import puppeteer from 'puppeteer';

const browser = await puppeteer.launch({
  headless: true,
  // 启动参数只放 host:port，凭据不能写在这里
  args: ['--proxy-server=http://GATEWAY_HOST:58971'],
});

const page = await browser.newPage();

// 凭据必须在 goto 之前设置
await page.authenticate({
  username: 'USERNAME-country-US-session-job1-time-30',
  password: 'PASSWORD',
});

await page.goto('https://api.ipify.org?format=json');
console.log(await page.evaluate(() => document.body.innerText));

await browser.close();

pip install selenium-wire

from seleniumwire import webdriver

PROXY = "http://USERNAME-country-US-session-job1-time-30:PASSWORD@GATEWAY_HOST:58971"

options = webdriver.ChromeOptions()
options.add_argument("--headless=new")

driver = webdriver.Chrome(
    options=options,
    seleniumwire_options={
        "proxy": {"http": PROXY, "https": PROXY, "no_proxy": "localhost,127.0.0.1"},
    },
)

try:
    driver.get("https://api.ipify.org?format=json")
    print(driver.find_element("tag name", "body").text)
finally:
    driver.quit()

import os
import tempfile
import zipfile

from selenium import webdriver

MANIFEST = """
{
  "version": "1.0.0",
  "manifest_version": 3,
  "name": "NextProxy Auth",
  "permissions": ["proxy", "webRequest", "webRequestAuthProvider"],
  "host_permissions": ["<all_urls>"],
  "background": {"service_worker": "background.js"}
}
"""

BACKGROUND = """
chrome.proxy.settings.set({
  value: {
    mode: 'fixed_servers',
    rules: {
      singleProxy: { scheme: 'http', host: '%(host)s', port: %(port)d },
      bypassList: ['localhost', '127.0.0.1']
    }
  },
  scope: 'regular'
});

chrome.webRequest.onAuthRequired.addListener(
  () => ({ authCredentials: { username: '%(user)s', password: '%(password)s' } }),
  { urls: ['<all_urls>'] },
  ['blocking']
);
"""


def build_auth_extension(host: str, port: int, user: str, password: str) -> str:
    """生成一个把代理凭据带上的临时扩展，返回 crx 路径。"""
    path = os.path.join(tempfile.mkdtemp(), "np-auth.zip")
    with zipfile.ZipFile(path, "w") as archive:
        archive.writestr("manifest.json", MANIFEST)
        archive.writestr("background.js", BACKGROUND % {
            "host": host, "port": port, "user": user, "password": password,
        })
    return path


options = webdriver.ChromeOptions()
options.add_extension(build_auth_extension(
    "GATEWAY_HOST", 58971,
    "USERNAME-country-US-session-job1-time-30", "PASSWORD",
))

driver = webdriver.Chrome(options=options)
try:
    driver.get("https://api.ipify.org?format=json")
    print(driver.find_element("tag name", "body").text)
finally:
    driver.quit()

import requests
from selenium import webdriver


def fetch_one_port(api_key: str) -> str:
    response = requests.get(
        "https://api.example.com/api/v1/proxy/extract",
        params={"apikey": api_key, "num": 1, "type": "json",
                "country": "US", "session": "sticky", "time": 30},
        timeout=15,
    )
    response.raise_for_status()
    item = response.json()["data"][0]
    return f"{item['ip']}:{item['port']}"


options = webdriver.ChromeOptions()
options.add_argument("--headless=new")
# 免密端口，没有凭据问题
options.add_argument(f"--proxy-server=http://{fetch_one_port('YOUR_API_KEY')}")

driver = webdriver.Chrome(options=options)
try:
    driver.get("https://api.ipify.org?format=json")
    print(driver.find_element("tag name", "body").text)
finally:
    driver.quit()

import java.util.HashMap;
import java.util.Map;

import org.openqa.selenium.chrome.ChromeDriver;
import org.openqa.selenium.chrome.ChromeOptions;

public class SeleniumProxyExample {

    public static void main(String[] args) {
        ChromeOptions options = new ChromeOptions();
        options.addArguments("--headless=new");
        // 只放 host:port；Chromium 不接受启动参数里的凭据
        options.addArguments("--proxy-server=http://GATEWAY_HOST:58971");

        ChromeDriver driver = new ChromeDriver(options);
        try {
            // 用 CDP 提供凭据，避免弹认证框
            Map<String, Object> headers = new HashMap<>();
            String credentials = java.util.Base64.getEncoder().encodeToString(
                    "USERNAME-country-US:PASSWORD".getBytes());
            headers.put("Proxy-Authorization", "Basic " + credentials);

            driver.executeCdpCommand("Network.enable", new HashMap<>());
            driver.executeCdpCommand("Network.setExtraHTTPHeaders",
                    Map.of("headers", headers));

            driver.get("https://api.ipify.org?format=json");
            System.out.println(driver.findElement(
                    org.openqa.selenium.By.tagName("body")).getText());
        } finally {
            driver.quit();
        }
    }
}

# Playwright：拦掉图片、字体、媒体
def block_heavy_resources(route):
    if route.request.resource_type in {"image", "media", "font"}:
        route.abort()
    else:
        route.continue_()

context.route("**/*", block_heavy_resources)