Playwright 代理设置，旋转住宅 IP（Python 和 Node.js）

from playwright.sync_api import sync_playwright

GATEWAY = "http://gw.crawlrelay.com:7777"
LOCALES = {"us": "en-US", "de": "de-DE", "jp": "ja-JP"}

with sync_playwright() as p:
    browser = p.chromium.launch()
    for country, locale in LOCALES.items():
        context = browser.new_context(
            proxy={"server": GATEWAY, "username": "USERNAME-country-%s" % country, "password": "PASSWORD"},
            locale=locale,
        )
        page = context.new_page()
        page.goto("https://api.ipify.org", wait_until="domcontentloaded")
        print(country, page.inner_text("body"))
        context.close()
    browser.close()

import { chromium } from "playwright";

const browser = await chromium.launch({
  proxy: {
    server: "http://gw.crawlrelay.com:7777",
    username: "USERNAME-country-us-session-pw1-ttl-30",
    password: "PASSWORD",
  },
});
const context = await browser.newContext();
// Images, fonts and media are most of a page's bytes
await context.route("**/*", (route) =>
  ["image", "font", "media"].includes(route.request().resourceType()) ? route.abort() : route.continue()
);
const page = await context.newPage();
const response = await page.goto("https://example.com/", { waitUntil: "domcontentloaded" });
console.log(response.status(), await page.title());
await browser.close();

import asyncio
import uuid

from playwright.async_api import async_playwright

GATEWAY = "http://gw.crawlrelay.com:7777"
WORKERS = 4


async def worker(browser, queue):
    session = uuid.uuid4().hex[:10]
    context = await browser.new_context(proxy={
        "server": GATEWAY,
        "username": "USERNAME-country-us-session-%s-ttl-30" % session,
        "password": "PASSWORD",
    })
    page = await context.new_page()
    while not queue.empty():
        url = await queue.get()
        try:
            response = await page.goto(url, wait_until="domcontentloaded", timeout=45000)
            print(session, url, response.status if response else None)
        except Exception as exc:  # timeouts and network errors: log and move on
            print(session, url, type(exc).__name__)
        finally:
            queue.task_done()
    await context.close()


async def main(urls):
    queue = asyncio.Queue()
    for url in urls:
        queue.put_nowait(url)
    async with async_playwright() as p:
        browser = await p.chromium.launch()
        await asyncio.gather(*(worker(browser, queue) for _ in range(WORKERS)))
        await browser.close()


asyncio.run(main(["https://example.com/", "https://example.org/"]))