(
method: str,
url: str,
headers: Dict[str, str],
payload: Optional[Dict[str, Any]],
timeout: int,
)
| 332 | |
| 333 | |
| 334 | async def _sync_json_http_request( |
| 335 | method: str, |
| 336 | url: str, |
| 337 | headers: Dict[str, str], |
| 338 | payload: Optional[Dict[str, Any]], |
| 339 | timeout: int, |
| 340 | ) -> tuple[int, Optional[Any], str]: |
| 341 | req_headers = dict(headers or {}) |
| 342 | req_headers.setdefault("Accept", "application/json") |
| 343 | request_method = (method or "GET").upper() |
| 344 | request_kwargs: Dict[str, Any] = { |
| 345 | "headers": req_headers, |
| 346 | "timeout": _build_remote_browser_http_timeout(timeout), |
| 347 | } |
| 348 | |
| 349 | if payload is not None: |
| 350 | req_headers["Content-Type"] = "application/json; charset=utf-8" |
| 351 | if request_method != "GET": |
| 352 | request_kwargs["json"] = payload |
| 353 | |
| 354 | if httpx is None: |
| 355 | return await _stdlib_json_http_request( |
| 356 | method=method, |
| 357 | url=url, |
| 358 | headers=req_headers, |
| 359 | payload=payload, |
| 360 | timeout=timeout, |
| 361 | ) |
| 362 | |
| 363 | try: |
| 364 | # remote_browser 控制面是服务间 JSON API,使用 httpx 避免 curl_cffi 在当前 |
| 365 | # Windows + impersonate 场景下 POST body 丢失导致 FastAPI 直接判定 body 缺失。 |
| 366 | async with httpx.AsyncClient(follow_redirects=False, trust_env=False) as session: |
| 367 | response = await session.request( |
| 368 | method=request_method, |
| 369 | url=url, |
| 370 | **request_kwargs, |
| 371 | ) |
| 372 | except Exception as e: |
| 373 | raise RuntimeError(f"远程打码服务请求失败: {e}") from e |
| 374 | |
| 375 | status_code = int(getattr(response, "status_code", 0) or 0) |
| 376 | text = response.text or "" |
| 377 | parsed = _parse_json_response_text(text) |
| 378 | |
| 379 | return status_code, parsed, text |
| 380 | |
| 381 | |
| 382 | async def _resolve_score_test_verify_proxy( |
no test coverage detected