执行原始文本请求（如 SSE），返回响应文本。
(
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,
respect_fingerprint_proxy: bool = True,
force_no_proxy: bool = False,
apply_default_client_headers: bool = True,
impersonate: str = "chrome124",
)
| 572 | raise Exception(f"Flow API request failed: {error_msg}") |
| 573 | |
| 574 | async def _make_text_request( |
| 575 | self, |
| 576 | method: str, |
| 577 | url: str, |
| 578 | headers: Optional[Dict] = None, |
| 579 | json_data: Optional[Dict] = None, |
| 580 | raw_body: Optional[Union[str, bytes]] = None, |
| 581 | use_st: bool = False, |
| 582 | st_token: Optional[str] = None, |
| 583 | use_at: bool = False, |
| 584 | at_token: Optional[str] = None, |
| 585 | timeout: Optional[int] = None, |
| 586 | respect_fingerprint_proxy: bool = True, |
| 587 | force_no_proxy: bool = False, |
| 588 | apply_default_client_headers: bool = True, |
| 589 | impersonate: str = "chrome124", |
| 590 | ) -> str: |
| 591 | """执行原始文本请求（如 SSE），返回响应文本。""" |
| 592 | fingerprint = self._request_fingerprint_ctx.get() |
| 593 | |
| 594 | proxy_url = None |
| 595 | if not force_no_proxy: |
| 596 | if self.proxy_manager: |
| 597 | if hasattr(self.proxy_manager, "get_request_proxy_url"): |
| 598 | proxy_url = await self.proxy_manager.get_request_proxy_url() |
| 599 | else: |
| 600 | proxy_url = await self.proxy_manager.get_proxy_url() |
| 601 | |
| 602 | if respect_fingerprint_proxy and isinstance(fingerprint, dict) and "proxy_url" in fingerprint: |
| 603 | proxy_url = fingerprint.get("proxy_url") |
| 604 | if proxy_url == "": |
| 605 | proxy_url = None |
| 606 | |
| 607 | request_timeout = timeout or self.timeout |
| 608 | |
| 609 | if headers is None: |
| 610 | headers = {} |
| 611 | else: |
| 612 | headers = dict(headers) |
| 613 | |
| 614 | if use_st and st_token: |
| 615 | headers["Cookie"] = f"__Secure-next-auth.session-token={st_token}" |
| 616 | |
| 617 | if use_at and at_token: |
| 618 | headers["authorization"] = f"Bearer {at_token}" |
| 619 | |
| 620 | account_id = None |
| 621 | if st_token: |
| 622 | account_id = st_token[:16] |
| 623 | elif at_token: |
| 624 | account_id = at_token[:16] |
| 625 | |
| 626 | fingerprint_user_agent = None |
| 627 | if isinstance(fingerprint, dict): |
| 628 | fingerprint_user_agent = fingerprint.get("user_agent") |
| 629 | |
| 630 | effective_user_agent = str(fingerprint_user_agent or self._generate_user_agent(account_id)).strip() |
| 631 | headers.setdefault("Content-Type", "application/json") |
no test coverage detected