Proxy API 集成：将代理连接到您的自动化流水线

Source: Dev.to

Source:

现代代理 API

现代代理提供商提供的 API 远超简单的 HTTP/SOCKS 连接。您可以以编程方式管理 IP、监控使用情况、控制会话，并自动化整个代理生命周期。以下是将代理 API 集成到您的流水线中的指南。

代理 API 能提供的功能

大多数高级提供商会公开以下 API：

• 会话管理 – 创建、延长和终止粘性会话
• IP 管理 – 请求新 IP、检查当前 IP、按需轮换
• 使用监控 – 跟踪带宽、请求次数和费用
• 池配置 – 设置地理定位、代理类型、轮换规则
• 健康检查 – 监控代理状态和可用性

常见的 API 集成模式

模式 1：网关模型

最简单的集成方式。所有请求都通过单一网关端点，提供商负责轮换和选择。

import requests

# 单一网关端点处理所有事务
proxy = {
    "http": "http://user:pass@gateway.provider.com:8080",
    "https": "http://user:pass@gateway.provider.com:8080"
}

response = requests.get("https://target.com", proxies=proxy)

优点： 极其简单，无需管理
缺点： 对 IP 选择的控制有限

模式 2：通过 API 控制会话

使用 API 显式创建和管理会话。

import requests

class ProxySession:
    def __init__(self, api_key, country="US"):
        self.api_key = api_key
        self.base_url = "https://api.provider.com/v1"
        self.country = country
        self.session_id = None

    def create_session(self, duration_minutes=30):
        response = requests.post(
            f"{self.base_url}/sessions",
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={
                "country": self.country,
                "duration": duration_minutes,
                "type": "residential"
            }
        )
        data = response.json()
        self.session_id = data["session_id"]
        return data["proxy_url"]

    def extend_session(self, additional_minutes=15):
        requests.patch(
            f"{self.base_url}/sessions/{self.session_id}",
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={"extend": additional_minutes}
        )

    def terminate_session(self):
        requests.delete(
            f"{self.base_url}/sessions/{self.session_id}",
            headers={"Authorization": f"Bearer {self.api_key}"}
        )
        self.session_id = None

模式 3：使用监控集成

以编程方式跟踪费用和使用情况。

import requests

class ProxyUsageMonitor:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.provider.com/v1"

    def get_usage(self, period="today"):
        response = requests.get(
            f"{self.base_url}/usage",
            headers={"Authorization": f"Bearer {self.api_key}"},
            params={"period": period}
        )
        return response.json()

    def check_budget(self, daily_limit_gb=10):
        usage = self.get_usage("today")
        used_gb = usage["bandwidth_bytes"] / (1024 ** 3)

        if used_gb > daily_limit_gb * 0.8:
            send_alert(f"Warning: {used_gb:.1f}GB used of {daily_limit_gb}GB daily limit")

        if used_gb > daily_limit_gb:
            send_alert("Daily proxy budget exceeded. Pausing operations.")
            return False

        return True

模式 4：多提供商故障转移

使用多个提供商并实现自动故障转移。

import requests

class MultiProviderProxy:
    def __init__(self, providers):
        self.providers = providers          # List of provider configs
        self.primary = 0

    def get_proxy(self):
        for i, provider in enumerate(self.providers):
            try:
                proxy = provider.get_proxy()
                if self.test_proxy(proxy):

                 return proxy
            except Exception:
                continue
        raise Exception("All proxy providers failed")

    def test_proxy(self, proxy):
        try:
            r = requests.get(
                "https://httpbin.org/ip",
                proxies={"http": proxy, "https": proxy},
                timeout=10
            )
            return r.status_code == 200
        except Exception:
            return False

Webhook 集成

一些供应商支持 webhook，用于实时事件，例如：

• 会话已过期
• 带宽阈值已达
• IP 被标记或列入黑名单
• 账户余额不足

from flask import Flask, request

app = Flask(__name__)

@app.route("/proxy-webhook", methods=["POST"])
def handle_webhook():
    event = request.json

    if event["type"] == "session_expired":
        renew_session(event["session_id"])
    elif event["type"] == "bandwidth_alert":
        send_alert(f"Bandwidth at {event['usage_pct']}%")
    elif event["type"] == "ip_flagged":
        rotate_proxy(event["account_id"])

    return {"status": "ok"}

最佳实践

• 抽象供应商细节 – 构建统一的代理接口，以便轻松切换供应商。
• 实现重试逻辑 – API 调用可能失败；始终使用指数退避进行重试。
• 缓存响应 – 不要在每次请求时调用使用情况 API；在合理的时间间隔内缓存结果。
• 设置预算警报 – 通过自动监控防止费用失控。
• 记录所有日志 – 记录会话创建、轮换事件和错误，以便审计。

g** — API 调用、创建的会话、使用的 IP，用于调试

如需代理 API 集成指南和自动化教程，请访问 DataResearchTools。