Python Playwright代理采集

export PROXY_USER="替换为控制台代理用户名"
export PROXY_PASS="替换为控制台代理密码"
# 隧道代理固定网关：proxy.123proxy.cn:36923

$env:PROXY_USER="替换为控制台代理用户名"
$env:PROXY_PASS="替换为控制台代理密码"
# 隧道代理固定网关：proxy.123proxy.cn:36923

# requirements:
#   python -m pip install playwright
#   playwright install chromium
import asyncio
import json
import os
from urllib.parse import urljoin

from playwright.async_api import async_playwright

PROXY_HOST = "proxy.123proxy.cn"
PROXY_PORT = 36923
PROXY_USER = os.environ["PROXY_USER"]
PROXY_PASS = os.environ["PROXY_PASS"]
START_URL = "https://quotes.toscrape.com/js/"
MAX_PAGES = 3


async def main():
    async with async_playwright() as playwright:
        browser = await playwright.chromium.launch(
            headless=True,
            proxy={
                "server": f"http://{PROXY_HOST}:{PROXY_PORT}",
                "username": PROXY_USER,
                "password": PROXY_PASS,
            },
        )
        context = await browser.new_context(
            user_agent="123Proxy-Developer-Example/1.0",
            locale="en-US",
        )
        page = await context.new_page()
        page.set_default_navigation_timeout(45_000)

        records = []
        url = START_URL
        for _ in range(MAX_PAGES):
            await page.goto(url, wait_until="networkidle")
            await page.locator("div.quote").first.wait_for()

            records.extend(
                await page.locator("div.quote").evaluate_all(
                    """nodes => nodes.map(node => ({
                      text: node.querySelector("span.text")?.textContent.trim(),
                      author: node.querySelector("small.author")?.textContent.trim(),
                      tags: [...node.querySelectorAll("a.tag")]
                        .map(tag => tag.textContent.trim()),
                      source_url: location.href
                    }))"""
                )
            )

            next_link = page.locator("li.next > a")
            if await next_link.count() == 0:
                break
            href = await next_link.get_attribute("href")
            url = urljoin(page.url, href)

        with open("quotes-playwright.json", "w", encoding="utf-8") as output:
            json.dump(records, output, ensure_ascii=False, indent=2)

        print(
            f"saved {len(records)} records to quotes-playwright.json"
        )
        await context.close()
        await browser.close()


if __name__ == "__main__":
    asyncio.run(main())