File size: 4,140 Bytes
b4f2238 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | """HTTP 客户端 - 封装 httpx，自动处理 JWT 认证和 token 刷新"""
from __future__ import annotations
import re
import threading
import time
from typing import Any
import httpx
from . import config
class FachuanClient:
def __init__(self) -> None:
self._access_token: str = ""
self._refresh_token: str = ""
self._expires_at: float = 0.0
self._lock = threading.Lock()
self._http = httpx.Client(base_url=config.BASE_URL, timeout=60)
def _obtain_token(self) -> None:
resp = self._http.post(
"/token/pair",
json={"username": config.USERNAME, "password": config.PASSWORD},
)
resp.raise_for_status()
data = resp.json()
self._access_token = data["access"]
self._refresh_token = data["refresh"]
self._expires_at = time.time() + 270 # 提前 30s 刷新
def _refresh(self) -> None:
try:
resp = self._http.post("/token/refresh", json={"refresh": self._refresh_token})
resp.raise_for_status()
self._access_token = resp.json()["access"]
self._expires_at = time.time() + 270
except Exception:
self._obtain_token()
def _ensure_token(self) -> None:
with self._lock:
if not self._access_token:
self._obtain_token()
elif time.time() >= self._expires_at:
self._refresh()
def _headers(self) -> dict[str, str]:
self._ensure_token()
return {"Authorization": f"Bearer {self._access_token}"}
def get(self, path: str, **kwargs: Any) -> Any:
resp = self._http.get(path, headers=self._headers(), **kwargs)
return self._handle(resp)
def post(self, path: str, **kwargs: Any) -> Any:
resp = self._http.post(path, headers=self._headers(), **kwargs)
return self._handle(resp)
def put(self, path: str, **kwargs: Any) -> Any:
resp = self._http.put(path, headers=self._headers(), **kwargs)
return self._handle(resp)
def delete(self, path: str, **kwargs: Any) -> Any:
resp = self._http.delete(path, headers=self._headers(), **kwargs)
return self._handle(resp)
def upload(
self,
path: str,
files: dict[str, tuple[str, bytes, str]],
data: dict[str, Any] | None = None,
) -> Any:
"""multipart/form-data 上传。files: {field: (filename, content, content_type)}"""
resp = self._http.post(path, headers=self._headers(), files=files, data=data or {})
return self._handle(resp)
def download(self, path: str, **kwargs: Any) -> tuple[bytes, str, str]:
"""下载二进制内容，返回 (content_bytes, filename, content_type)。"""
resp = self._http.get(path, headers=self._headers(), **kwargs)
if not resp.is_success:
try:
detail = resp.json()
except Exception:
detail = resp.text
raise RuntimeError(f"HTTP {resp.status_code}: {detail}")
content_type = resp.headers.get("content-type", "application/octet-stream")
disposition = resp.headers.get("content-disposition", "")
filename = "download"
if disposition:
# 尝试解析 filename*=UTF-8''... 或 filename="..."
m = re.search(r"filename\*=UTF-8''([^;]+)", disposition)
if m:
from urllib.parse import unquote
filename = unquote(m.group(1))
else:
m2 = re.search(r'filename="?([^";]+)"?', disposition)
if m2:
filename = m2.group(1).strip()
return resp.content, filename, content_type
@staticmethod
def _handle(resp: httpx.Response) -> Any:
if not resp.is_success:
try:
detail = resp.json()
except Exception:
detail = resp.text
raise RuntimeError(f"HTTP {resp.status_code}: {detail}")
if resp.status_code == 204:
return None
return resp.json()
# 全局单例
client = FachuanClient()
|