Method _build_headers

app/clients/fingerprint_http.py:178–280

Build a Chrome-compliant ordered header dict. Header order follows Chrome's HTTP/2 frame ordering as captured via packet inspection (Charles Proxy / Wireshark): 1. sec-ch-ua group (Chrome/Edge only – low-entropy Client Hints) 2. content-type (body requests only)

(
        self,
        *,
        method: str,
        headers: dict[str, str] | None,
        user_agent: str | None,
        browser_impersonate: str | None,
        has_body: bool = False,
        sec_fetch_site: str = "same-origin",
    )

                    "上游请求失败",
                    details={"transport": self.transport_name, "url": url, "reason": str(exc)},
                ) from exc
            finally:
                session.close()

        try:
            kwargs = {
                "method": method.upper(),
                "url": url,
                "headers": headers,
                "cookies": cookies,
                "params": params,
            }
            if json_body is not None:
                kwargs["json"] = json_body
            if form_body is not None:
                kwargs["data"] = form_body
            client_kwargs: dict[str, Any] = {
                "timeout": self.settings.request_timeout_seconds,
                "follow_redirects": True,
                "trust_env": False,
            }
            if proxy_url:
                client_kwargs["proxy"] = proxy_url
            with httpx.Client(**client_kwargs) as client:
                return client.request(**kwargs)
        except Exception as exc:
            raise UpstreamRequestError(
                "上游请求失败",
                details={"transport": self.transport_name, "url": url, "reason": str(exc)},
            ) from exc

    def _decode_response(self, *, response: Any, url: str):
        self._ensure_success(response=response, url=url)
        try:
            payload = response.json()
        except Exception as exc:
            raise UpstreamRequestError(
                "上游返回了非 JSON 数据",
                details={
                    "transport": self.transport_name,
                    "url": url,
                    "status_code": getattr(response, "status_code", "unknown"),
                    "body_preview": getattr(response, "text", "")[:500],
                },
            ) from exc

        return payload

    def _ensure_success(self, *, response: Any, url: str) -> None:
        status_code = int(getattr(response, "status_code", 0) or 0)
        if status_code < 400:
            return
        body_preview = str(getattr(response, "text", ""))[:500]
        raise UpstreamRequestError(
            "上游接口返回错误",
            details={"url": url, "status_code": status_code, "body_preview": body_preview},
        )