在 Python 中使用美国代理：requests、httpx、Scrapy 和 Playwright

import requests

HOST = "gw.statesideip.com"
USER = "USERNAME-country-us-state-ny"  # rotating New York State IP
PASSWORD = "PASSWORD"

proxy = f"http://{USER}:{PASSWORD}@{HOST}:7777"
r = requests.get("https://api.ipify.org?format=json",
                 proxies={"http": proxy, "https": proxy}, timeout=30)
print(r.json())

# SOCKS5 (pip install "requests[socks]"); socks5h resolves DNS through the proxy
socks = f"socks5h://{USER}:{PASSWORD}@{HOST}:7778"
r = requests.get("https://api.ipify.org?format=json",
                 proxies={"http": socks, "https": socks}, timeout=30)
print(r.json())

import uuid

def proxy_url(state=None, city=None, zip_code=None, session=None, ttl=10):
    parts = ["USERNAME", "country-us"]
    if state:
        parts.append(f"state-{state}")
    if city:
        parts.append(f"city-{city}")
    if zip_code:
        parts.append(f"zip-{zip_code}")
    if session:
        parts += [f"session-{session}", f"ttl-{ttl}"]
    user = "-".join(parts)
    return f"http://{user}:[email protected]:7777"

rotating = proxy_url(state="tx", city="houston")  # new IP per request
sticky = proxy_url(zip_code="10001", session=uuid.uuid4().hex[:8], ttl=30)  # one IP, 30 min

import asyncio
import httpx

PROXY = "http://USERNAME-country-us-state-ca:[email protected]:7777"
URLS = ["https://api.ipify.org?format=json"] * 20

async def fetch(client, sem, url):
    async with sem:  # caps requests in flight
        r = await client.get(url)
        r.raise_for_status()
        return r.json()

async def main():
    sem = asyncio.Semaphore(5)
    limits = httpx.Limits(max_keepalive_connections=0)  # new connection, new IP
    async with httpx.AsyncClient(proxy=PROXY, limits=limits, timeout=30) as client:
        results = await asyncio.gather(*(fetch(client, sem, u) for u in URLS),
                                       return_exceptions=True)
    for res in results:
        print(res)

asyncio.run(main())

import random
import time
import requests

RETRY_STATUS = {429, 500, 502, 503, 504}

def get_with_retries(url, proxies, attempts=5):
    for attempt in range(attempts):
        try:
            r = requests.get(url, proxies=proxies, timeout=30)
            if r.status_code == 407:
                raise RuntimeError("407: check USERNAME and PASSWORD")  # do not retry
            if r.status_code not in RETRY_STATUS:
                return r
            ra = r.headers.get("Retry-After", "")
            wait = int(ra) if ra.isdigit() else 2 ** attempt
        except (requests.ConnectionError, requests.Timeout):
            wait = 2 ** attempt
        time.sleep(wait + random.uniform(0, 1))  # backoff plus jitter
    raise RuntimeError(f"gave up on {url}")

# settings.py
ROBOTSTXT_OBEY = True
CONCURRENT_REQUESTS_PER_DOMAIN = 4
AUTOTHROTTLE_ENABLED = True
RETRY_TIMES = 3
RETRY_HTTP_CODES = [429, 500, 502, 503, 504]

# spiders/ip_check.py
import scrapy

PROXY = "http://USERNAME-country-us-state-fl-city-miami:[email protected]:7777"

class IpCheckSpider(scrapy.Spider):
    name = "ip_check"
    start_urls = ["https://api.ipify.org?format=json"]

    def start_requests(self):
        for url in self.start_urls:
            yield scrapy.Request(url, meta={"proxy": PROXY}, dont_filter=True)

    def parse(self, response):
        yield {"ip": response.json()["ip"]}

import asyncio
from playwright.async_api import async_playwright

async def main():
    async with async_playwright() as p:
        browser = await p.chromium.launch(proxy={
            "server": "http://gw.statesideip.com:7777",
            "username": "USERNAME-country-us-state-il-city-chicago-session-pw1-ttl-15",
            "password": "PASSWORD",
        })
        page = await browser.new_page()
        await page.goto("https://api.ipify.org?format=json")
        print(await page.text_content("body"))
        await browser.close()

asyncio.run(main())