统一HTTP请求处理
(
self,
method: str,
url: str,
headers: Optional[Dict] = None,
json_data: Optional[Dict] = None,
raw_body: Optional[Union[str, bytes]] = None,
use_st: bool = False,
st_token: Optional[str] = None,
use_at: bool = False,
at_token: Optional[str] = None,
timeout: Optional[int] = None,
use_media_proxy: bool = False,
respect_fingerprint_proxy: bool = True,
force_no_proxy: bool = False,
allow_urllib_fallback: bool = True,
apply_default_client_headers: bool = True,
impersonate: str = "chrome124",
)
| 350 | return None |
| 351 | |
| 352 | async def _make_request( |
| 353 | self, |
| 354 | method: str, |
| 355 | url: str, |
| 356 | headers: Optional[Dict] = None, |
| 357 | json_data: Optional[Dict] = None, |
| 358 | raw_body: Optional[Union[str, bytes]] = None, |
| 359 | use_st: bool = False, |
| 360 | st_token: Optional[str] = None, |
| 361 | use_at: bool = False, |
| 362 | at_token: Optional[str] = None, |
| 363 | timeout: Optional[int] = None, |
| 364 | use_media_proxy: bool = False, |
| 365 | respect_fingerprint_proxy: bool = True, |
| 366 | force_no_proxy: bool = False, |
| 367 | allow_urllib_fallback: bool = True, |
| 368 | apply_default_client_headers: bool = True, |
| 369 | impersonate: str = "chrome124", |
| 370 | ) -> Dict[str, Any]: |
| 371 | """统一HTTP请求处理""" |
| 372 | fingerprint = self._request_fingerprint_ctx.get() |
| 373 | |
| 374 | proxy_url = None |
| 375 | if not force_no_proxy: |
| 376 | if self.proxy_manager: |
| 377 | if use_media_proxy and hasattr(self.proxy_manager, "get_media_proxy_url"): |
| 378 | proxy_url = await self.proxy_manager.get_media_proxy_url() |
| 379 | elif hasattr(self.proxy_manager, "get_request_proxy_url"): |
| 380 | proxy_url = await self.proxy_manager.get_request_proxy_url() |
| 381 | else: |
| 382 | proxy_url = await self.proxy_manager.get_proxy_url() |
| 383 | |
| 384 | if respect_fingerprint_proxy and isinstance(fingerprint, dict) and "proxy_url" in fingerprint: |
| 385 | proxy_url = fingerprint.get("proxy_url") |
| 386 | if proxy_url == "": |
| 387 | proxy_url = None |
| 388 | request_timeout = timeout or self.timeout |
| 389 | |
| 390 | if headers is None: |
| 391 | headers = {} |
| 392 | else: |
| 393 | headers = dict(headers) |
| 394 | |
| 395 | if use_st and st_token: |
| 396 | headers["Cookie"] = f"__Secure-next-auth.session-token={st_token}" |
| 397 | |
| 398 | if use_at and at_token: |
| 399 | headers["authorization"] = f"Bearer {at_token}" |
| 400 | |
| 401 | account_id = None |
| 402 | if st_token: |
| 403 | account_id = st_token[:16] |
| 404 | elif at_token: |
| 405 | account_id = at_token[:16] |
| 406 | |
| 407 | fingerprint_user_agent = None |
| 408 | if isinstance(fingerprint, dict): |
| 409 | fingerprint_user_agent = fingerprint.get("user_agent") |
no test coverage detected