认证与调用约定

curl -X POST 'https://api.example.com/api/v1/auth/login' \
  -H 'Origin: https://console.example.com' \
  -H 'Content-Type: application/json' \
  -c cookies.txt \
  -d '{"email":"you@example.com","password":"YOUR_PASSWORD","remember":true}'

curl -X POST 'https://api.example.com/api/v1/auth/login/two-factor' \
  -H 'Origin: https://console.example.com' \
  -H 'Content-Type: application/json' \
  -c cookies.txt \
  -d '{"ticket":"MFA_TICKET","code":"123456"}'

curl -X POST 'https://api.example.com/api/v1/auth/refresh' \
  -H 'Origin: https://console.example.com' \
  -b cookies.txt -c cookies.txt

{ "data": { } }

{
  "error": {
    "code": "machine_readable_code",
    "message": "给人看的说明",
    "fields": { "FieldName": "validationTag" }
  }
}

{
  "data": {
    "items": [],
    "page": 1,
    "pageSize": 20,
    "totalItems": 137,
    "totalPages": 7
  }
}

import threading
import time

import requests


class NextProxyClient:
    """带自动刷新的客户 API 客户端。

    access token 只有 15 分钟，所以在过期前主动换新，
    而不是等 401 再重试——后者会让每次过期都白费一个请求。
    """

    # 提前这么多秒刷新，留出网络往返的余量
    REFRESH_MARGIN = 120

    def __init__(self, base_url: str, origin: str) -> None:
        self._base = base_url.rstrip("/")
        self._origin = origin
        self._session = requests.Session()
        self._token: str | None = None
        self._expires_at = 0.0
        self._lock = threading.Lock()

    def login(self, email: str, password: str, *, remember: bool = True) -> None:
        payload = {"email": email, "password": password, "remember": remember}
        data = self._post("/api/v1/auth/login", payload)

        if "mfaTicket" in data:
            raise RuntimeError("账号启用了两步验证，请改用 login_with_totp")

        self._store_token(data)

    def login_with_totp(self, email: str, password: str, code: str) -> None:
        data = self._post(
            "/api/v1/auth/login",
            {"email": email, "password": password, "remember": True},
        )
        ticket = data.get("mfaTicket")
        if not ticket:
            self._store_token(data)
            return

        # MFA ticket 只有 5 分钟有效
        data = self._post(
            "/api/v1/auth/login/two-factor",
            {"ticket": ticket, "code": code},
        )
        self._store_token(data)

    def get(self, path: str, **params) -> dict:
        response = self._session.get(
            self._base + path,
            params=params or None,
            headers=self._headers(),
            timeout=30,
        )
        return self._unwrap(response)

    def post(self, path: str, payload: dict) -> dict:
        return self._post(path, payload, authenticated=True)

    def _post(self, path: str, payload: dict, *, authenticated: bool = False) -> dict:
        headers = self._headers() if authenticated else {"Origin": self._origin}
        response = self._session.post(
            self._base + path, json=payload, headers=headers, timeout=30,
        )
        return self._unwrap(response)

    def _headers(self) -> dict[str, str]:
        self._ensure_token()
        return {
            "Authorization": f"Bearer {self._token}",
            # 写操作必须带可信 Origin，否则 403 origin_rejected
            "Origin": self._origin,
        }

    def _ensure_token(self) -> None:
        with self._lock:
            if self._token and time.time() < self._expires_at - self.REFRESH_MARGIN:
                return
            response = self._session.post(
                self._base + "/api/v1/auth/refresh",
                headers={"Origin": self._origin},
                timeout=30,
            )
            self._store_token(self._unwrap(response))

    def _store_token(self, data: dict) -> None:
        self._token = data["accessToken"]
        # 服务端给的是秒数；没给就按默认 15 分钟算
        self._expires_at = time.time() + data.get("expiresIn", 900)

    @staticmethod
    def _unwrap(response: requests.Response) -> dict:
        payload = response.json()
        if response.ok:
            return payload.get("data", payload)

        error = payload.get("error", {})
        raise RuntimeError(
            f"HTTP {response.status_code} {error.get('code', 'unknown')}: "
            f"{error.get('message', response.text[:200])}"
        )


client = NextProxyClient("https://api.example.com", "https://console.example.com")
client.login("you@example.com", "YOUR_PASSWORD")
print(client.get("/api/v1/wallet"))