企业隧道代理怎么选？从本地 IP 池切换到统一入口，架构上要看什么？

import os
​
import requests
​
​
proxy_url = os.environ["TUNNEL_PROXY_URL"]
​
proxies = {
    "http": proxy_url,
    "https": proxy_url,
}
​
response = requests.get(
    "https://example.com/public-page",
    proxies=proxies,
    timeout=(5, 15),
)
​
print(response.status_code)

import requests
​
​
def create_session(proxies):
    session = requests.Session()
    session.proxies.update(proxies)
    session.headers.update({
        "User-Agent": "DataClient/1.0",
    })
​
    return session
​
​
def fetch_pages(session):
    list_page = session.get(
        "https://example.com/public-list",
        timeout=(5, 15),
    )
    list_page.raise_for_status()
​
    detail_page = session.get(
        "https://example.com/public-detail",
        timeout=(5, 15),
    )
    detail_page.raise_for_status()
​
    return detail_page.text

import time
from typing import Any
​
import requests
from requests.exceptions import (
    ConnectTimeout,
    ProxyError,
    ReadTimeout,
    RequestException,
)
​
​
def request_once(
    session: requests.Session,
    url: str,
) -> dict[str, Any]:
    started_at = time.perf_counter()
​
    try:
        response = session.get(
            url,
            timeout=(5, 15),
        )
​
        elapsed_ms = (
            time.perf_counter() - started_at
        ) * 1000
​
        return {
            "type": "response",
            "status_code": response.status_code,
            "elapsed_ms": round(elapsed_ms, 2),
            "response_bytes": len(response.content),
        }
​
    except ProxyError as exc:
        return {
            "type": "proxy_error",
            "message": str(exc),
        }
​
    except ConnectTimeout:
        return {
            "type": "connect_timeout",
        }
​
    except ReadTimeout:
        return {
            "type": "read_timeout",
        }
​
    except RequestException as exc:
        return {
            "type": "request_error",
            "message": str(exc),
        }

import time
​
​
RETRYABLE_ERRORS = {
    "proxy_error",
    "connect_timeout",
    "read_timeout",
}
​
​
def fetch_with_retry(
    session,
    url,
    max_attempts=3,
):
    last_result = None
​
    for attempt in range(max_attempts):
        result = request_once(session, url)
        last_result = result
​
        if result["type"] == "response":
            return result
​
        if result["type"] not in RETRYABLE_ERRORS:
            return result
​
        if attempt < max_attempts - 1:
            time.sleep(0.5 * (2 ** attempt))
​
    return last_result

from concurrent.futures import (
    ThreadPoolExecutor,
    as_completed,
)
from statistics import median
​
​
def percentile(values, ratio):
    if not values:
        return None
​
    index = int((len(values) - 1) * ratio)
    return values[index]
​
​
def run_probe(
    session_factory,
    url,
    total_requests=200,
    workers=5,
):
    def execute():
        session = session_factory()
        return request_once(session, url)
​
    results = []
​
    with ThreadPoolExecutor(
        max_workers=workers,
    ) as executor:
        futures = [
            executor.submit(execute)
            for _ in range(total_requests)
        ]
​
        for future in as_completed(futures):
            results.append(future.result())
​
    responses = [
        item
        for item in results
        if item["type"] == "response"
    ]
​
    elapsed = sorted(
        item["elapsed_ms"]
        for item in responses
    )
​
    return {
        "total": len(results),
        "responses": len(responses),
        "proxy_errors": sum(
            item["type"] == "proxy_error"
            for item in results
        ),
        "connect_timeouts": sum(
            item["type"] == "connect_timeout"
            for item in results
        ),
        "read_timeouts": sum(
            item["type"] == "read_timeout"
            for item in results
        ),
        "median_ms": (
            median(elapsed)
            if elapsed
            else None
        ),
        "p95_ms": percentile(elapsed, 0.95),
    }

import threading
import time
​
​
class RateLimiter:
    def __init__(self, requests_per_second):
        self.interval = 1 / requests_per_second
        self.lock = threading.Lock()
        self.next_time = time.perf_counter()
​
    def wait(self):
        with self.lock:
            now = time.perf_counter()
​
            if now < self.next_time:
                time.sleep(self.next_time - now)
​
            self.next_time = (
                max(now, self.next_time)
                + self.interval
            )

limiter = RateLimiter(
    requests_per_second=5,
)
​
​
def limited_request(session, url):
    limiter.wait()
    return request_once(session, url)