Python代理：requests, httpx, aiohttp 和 Scrapy

# pip install requests
import requests

USERNAME = "USERNAME"   # base proxy username from the client area
PASSWORD = "PASSWORD"   # proxy password (not your account password)
HOST = "gw.namelessproxy.com"
HTTP_PORT = 7777
SOCKS_PORT = 7778

proxy = f"http://{USERNAME}-country-ae:{PASSWORD}@{HOST}:{HTTP_PORT}"
proxies = {"http": proxy, "https": proxy}

r = requests.get("https://api.ipify.org?format=json", proxies=proxies, timeout=30)
print(r.json())   # {'ip': '...'}

import uuid

def proxy_url(country, city=None, session=None, ttl=30, scheme="http"):
    """Build a gateway URL. Targeting lives in the username, in this order:
    country -> city -> session -> ttl."""
    user = f"{USERNAME}-country-{country.lower()}"
    if city:
        user += f"-city-{city.lower().replace(' ', '_')}"
        if session:
            user += f"-session-{session}-ttl-{ttl}"
    port = SOCKS_PORT if scheme.startswith("socks") else HTTP_PORT
    return f"{scheme}://{user}:{PASSWORD}@{HOST}:{port}"

def new_session():
    """A fresh session ID = a fresh sticky IP."""
    return uuid.uuid4().hex[:10]

# Rotating IP in Egypt, and a sticky IP in Abu Dhabi for 30 minutes
rotating = proxy_url("eg")
sticky = proxy_url("ae", city="Abu Dhabi", session=new_session(), ttl=30)

# pip install "requests[socks]"
# socks5h:// = the proxy resolves DNS, so lookups also exit in the target country
proxy = proxy_url("ke", scheme="socks5h")
r = requests.get("https://api.ipify.org", proxies={"http": proxy, "https": proxy}, timeout=30)
print(r.text)

import random
import time

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

def fetch(url, country="ae", attempts=4):
    """Retry with exponential backoff. Each attempt opens a new connection,
    so a rotating username exits from a new IP."""
    for attempt in range(attempts):
        proxy = proxy_url(country)
        try:
            r = requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=30)
            if r.status_code not in RETRY_STATUS:
                return r
        except (requests.exceptions.ConnectionError, requests.exceptions.Timeout):
            pass  # ProxyError is a ConnectionError: retry on a new IP
        time.sleep(min(2 ** attempt + random.random(), 30))
    raise RuntimeError(f"Giving up on {url} after {attempts} attempts")

# pip install httpx   (0.26 or later: the argument is "proxy", not "proxies")
import asyncio
import httpx

# Sync: one sticky IP in Nairobi for 10 minutes
with httpx.Client(proxy=proxy_url("ke", city="nairobi", session=new_session(), ttl=10), timeout=30) as client:
    print(client.get("https://api.ipify.org").text)

# Async: 20 connections at most, rotating IPs in Nigeria
async def main(urls):
    limits = httpx.Limits(max_connections=20)
    async with httpx.AsyncClient(proxy=proxy_url("ng"), timeout=30, limits=limits) as client:
        results = await asyncio.gather(*(client.get(u) for u in urls), return_exceptions=True)
    for url, res in zip(urls, results):
        print(url, res.status_code if isinstance(res, httpx.Response) else repr(res))

asyncio.run(main(["https://api.ipify.org"] * 5))

# pip install aiohttp   (HTTP proxies only; the credentials can sit in the proxy URL)
import asyncio
import aiohttp

async def fetch(session, url, sem):
    async with sem:
        async with session.get(url, proxy=proxy_url("eg")) as r:
            return r.status, await r.text()

async def main(urls):
    sem = asyncio.Semaphore(20)   # concurrency cap
    timeout = aiohttp.ClientTimeout(total=30)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        results = await asyncio.gather(*(fetch(session, u, sem) for u in urls), return_exceptions=True)
    print(results)

asyncio.run(main(["https://api.ipify.org"] * 5))

# myproject/middlewares.py
USERNAME = "USERNAME"
PASSWORD = "PASSWORD"
GATEWAY = "gw.namelessproxy.com:7777"

class NamelessProxyMiddleware:
    """Set the proxy per request from request.meta:
    country (default 'ae'), optional city, optional session + ttl."""

    def process_request(self, request, spider):
        meta = request.meta
        user = f"{USERNAME}-country-{meta.get('country', 'ae')}"
        if meta.get("city"):
            user += f"-city-{meta['city']}"
            if meta.get("session"):
                user += f"-session-{meta['session']}-ttl-{meta.get('ttl', 10)}"
        # Scrapy's HttpProxyMiddleware turns the credentials into Proxy-Authorization
        meta["proxy"] = f"http://{user}:{PASSWORD}@{GATEWAY}"

# myproject/settings.py
DOWNLOADER_MIDDLEWARES = {
    # Below 750 so it runs before Scrapy's built-in HttpProxyMiddleware
    "myproject.middlewares.NamelessProxyMiddleware": 350,
}
CONCURRENT_REQUESTS = 16
DOWNLOAD_TIMEOUT = 30
RETRY_TIMES = 3
RETRY_HTTP_CODES = [429, 500, 502, 503, 504, 408]
AUTOTHROTTLE_ENABLED = True
ROBOTSTXT_OBEY = True

# In a spider:
# yield scrapy.Request(url, meta={"country": "sa", "city": "riyadh", "session": "cart01"})