Python Selenium代理采集

export PROXY_USER="替换为控制台代理用户名"
export PROXY_PASS="替换为控制台代理密码"
# 不限量动态住宅固定网关：unlimit.residential.123proxy.cn:10253

$env:PROXY_USER="替换为控制台代理用户名"
$env:PROXY_PASS="替换为控制台代理密码"
# 不限量动态住宅固定网关：unlimit.residential.123proxy.cn:10253

# requirements:
#   python -m pip install selenium
# run:
#   python selenium_proxy.py
import json
import os
import shutil
import tempfile
import time
from pathlib import Path
from urllib.parse import urljoin

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

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


def create_proxy_extension():
    extension_dir = Path(tempfile.mkdtemp(prefix="123proxy-selenium-"))
    manifest = {
        "manifest_version": 3,
        "name": "123Proxy Selenium Auth",
        "version": "1.0.0",
        "minimum_chrome_version": "108",
        "permissions": [
            "proxy",
            "webRequest",
            "webRequestAuthProvider",
        ],
        "host_permissions": ["<all_urls>"],
        "background": {"service_worker": "service_worker.js"},
    }
    worker = """
chrome.proxy.settings.set({
  value: {
    mode: "fixed_servers",
    rules: {
      singleProxy: {
        scheme: "http",
        host: "__HOST__",
        port: __PORT__
      },
      bypassList: ["localhost", "127.0.0.1"]
    }
  },
  scope: "regular"
});

chrome.webRequest.onAuthRequired.addListener(
  (details, callback) => {
    callback({
      authCredentials: {
        username: __USER__,
        password: __PASS__
      }
    });
  },
  {urls: ["<all_urls>"]},
  ["asyncBlocking"]
);
"""
    worker = (
        worker.replace("__HOST__", PROXY_HOST)
        .replace("__PORT__", str(PROXY_PORT))
        .replace("__USER__", json.dumps(PROXY_USER))
        .replace("__PASS__", json.dumps(PROXY_PASS))
    )
    (extension_dir / "manifest.json").write_text(
        json.dumps(manifest),
        encoding="utf-8",
    )
    (extension_dir / "service_worker.js").write_text(
        worker,
        encoding="utf-8",
    )
    return extension_dir


def text(node, selector):
    return node.find_element(By.CSS_SELECTOR, selector).text.strip()


def main():
    extension_dir = create_proxy_extension()
    options = webdriver.ChromeOptions()
    options.add_argument("--window-size=1440,1000")
    options.add_argument("--lang=en-US")
    options.add_argument("--load-extension=" + str(extension_dir))
    if os.environ.get("HEADLESS") == "1":
        options.add_argument("--headless=new")

    driver = webdriver.Chrome(options=options)
    driver.set_page_load_timeout(45)
    wait = WebDriverWait(driver, 30)
    records = []

    try:
        time.sleep(1)
        url = START_URL
        for _ in range(MAX_PAGES):
            driver.get(url)
            wait.until(
                EC.presence_of_element_located((By.CSS_SELECTOR, "div.quote"))
            )

            for quote in driver.find_elements(By.CSS_SELECTOR, "div.quote"):
                records.append({
                    "text": text(quote, "span.text"),
                    "author": text(quote, "small.author"),
                    "tags": [
                        tag.text.strip()
                        for tag in quote.find_elements(By.CSS_SELECTOR, "a.tag")
                    ],
                    "source_url": driver.current_url,
                })

            next_links = driver.find_elements(By.CSS_SELECTOR, "li.next > a")
            if not next_links:
                break
            url = urljoin(driver.current_url, next_links[0].get_attribute("href"))

        Path("quotes-selenium.json").write_text(
            json.dumps(records, ensure_ascii=False, indent=2),
            encoding="utf-8",
        )
        print(
            "saved "
            + str(len(records))
            + " records to quotes-selenium.json"
        )
    finally:
        driver.quit()
        shutil.rmtree(extension_dir, ignore_errors=True)


if __name__ == "__main__":
    main()