如何在 Playwright 中设置代理进行网页抓取

npm init playwright@latest

pip install playwright

playwright install

dotnet new console -n PlaywrightDemo

cd PlaywrightDemo

dotnet add package Microsoft.Playwright

dotnet build

pwsh bin/Debug/netX/playwright.ps1 install 

<dependency>

  <groupId>com.microsoft.playwright</groupId>

  <artifactId>playwright</artifactId>

  <version>1.59.0</version>

</dependency> 

mvn compile exec:java -D exec.mainClass="org.example.App"

pip install playwright

playwright install

HTTP_PROXY=http://username:password@proxy-server-ip:host
SOCKS5_PROXY=socks5://username:password@proxy-server-ip:host

CYBERYOZH_API_KEY=Your_API_URL_Key

import asyncio
import os
import json
import requests
from dotenv import load_dotenv
from playwright.async_api import async_playwright

load_dotenv()

PROXY_URL = os.getenv("HTTP_PROXY")  # or SOCKS5, from the .env document
CYBERYOZH_API_KEY = os.getenv("CYBERYOZH_API_KEY") # API key from the .env document
TARGET_URL = "https://httpbin.org/ip" # the website to scrape; here is the test one
OUTPUT_FILE = "output/results.json" # output folder


def check_ip_quality(ip: str) -> bool:

    """Use CyberYozh IP Checker API to verify trust score before rotating."""

    try:

        response = requests.get(
            f"https://app.cyberyozh.com/api/v1/checkers/socks/",
            headers={"Authorization": f"Bearer {CYBERYOZH_API_KEY}"},
            timeout=5
        )

        score = response.json().get("fraud_score", 100)
        return score < 50  # Accept IPs with fraud score below 50

    except Exception:

        return True  # Allow on API error


async def scrape(proxy_url: str) -> dict:

    """Run a single Playwright scraping session with proxy."""

    # Parse proxy credentials for Playwright proxy authentication

    from urllib.parse import urlparse

    parsed = urlparse(proxy_url)

    proxy_config = {
        "server": f"{parsed.scheme}://{parsed.hostname}:{parsed.port}",
        "username": parsed.username,
        "password": parsed.password,
    }



    async with async_playwright() as p:

        # Browser entity (headless to save traffic): launch with proxy server

        browser = await p.chromium.launch(headless=True)
        # Context entity: proxy settings applied per context

        context = await browser.new_context(proxy=proxy_config)
        page = await context.new_page()




        try:

            await page.goto(TARGET_URL, timeout=30000)
            content = await page.inner_text("body")
            result = json.loads(content)
            return {"status": "ok", "ip": result.get("origin")}

        except Exception as e:

            return {"status": "error", "error": str(e)}

        finally:

            await context.close()
            await browser.close()


async def main():

    results = []

    for i in range(5):

        print(f"Request {i+1}: using proxy {PROXY_URL}")
        data = await scrape(PROXY_URL)
        print(f"  → IP: {data.get('ip')} | Status: {data.get('status')}")
        results.append(data)

    os.makedirs("output", exist_ok=True)

    with open(OUTPUT_FILE, "w") as f:

        json.dump(results, f, indent=2)

    print(f"Results saved to {OUTPUT_FILE}")


asyncio.run(main())

python scraper.py