Function _invoke_third_party_bind_api

src/web/routes/payment.py:1416–1483

(
    *,
    api_url: str,
    api_key: Optional[str],
    payload: dict,
    proxy: Optional[str] = None,
)

def _invoke_third_party_bind_api(
    *,
    api_url: str,
    api_key: Optional[str],
    payload: dict,
    proxy: Optional[str] = None,
) -> tuple[dict, str]:
    headers = {
        "Accept": "*/*",
        "Content-Type": "application/json",
        "User-Agent": "codex-console2/third-party-bind",
    }
    key = str(api_key or "").strip()
    if key:
        headers["X-API-Key"] = key
        headers["Authorization"] = f"Bearer {key}"

    url_candidates = _build_third_party_bind_api_candidates(api_url)
    if not url_candidates:
        raise RuntimeError("第三方绑卡 API 地址无效")

    proxy_candidates: List[Optional[str]] = []
    for value in (proxy, None):
        if value not in proxy_candidates:
            proxy_candidates.append(value)

    errors: List[str] = []
    for candidate_url in url_candidates:
        for proxy_item in proxy_candidates:
            proxies = {"http": proxy_item, "https": proxy_item} if proxy_item else None
            for attempt in range(1, 3):
                try:
                    resp = cffi_requests.post(
                        candidate_url,
                        headers=headers,
                        json=payload,
                        proxies=proxies,
                        timeout=120,
                        impersonate="chrome110",
                    )

                    if resp.status_code >= 400:
                        body = (resp.text or "")[:500]
                        err = f"{candidate_url} status={resp.status_code} proxy={'on' if proxy_item else 'off'} body={body}"
                        errors.append(err)
                        retryable = resp.status_code in (408, 409, 425, 429, 500, 502, 503, 504)
                        endpoint_maybe_wrong = resp.status_code in (404, 405)
                        if attempt < 2 and retryable:
                            time.sleep(0.6 * attempt)
                            continue
                        if endpoint_maybe_wrong:
                            break
                        raise RuntimeError(f"第三方绑卡请求失败: HTTP {resp.status_code} - {body}")

                    parsed = _parse_third_party_response(resp)
                    if isinstance(parsed, dict):
                        parsed["_meta_endpoint"] = candidate_url
                        parsed["_meta_proxy"] = "on" if proxy_item else "off"