-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathhttp_utils.py
More file actions
50 lines (43 loc) · 1.73 KB
/
Copy pathhttp_utils.py
File metadata and controls
50 lines (43 loc) · 1.73 KB
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
"""HTTP 响应解析工具。"""
from __future__ import annotations
import json
import requests
def response_text(resp: requests.Response) -> str:
"""获取响应文本,兼容 UTF-8 BOM。"""
if not resp.encoding or resp.encoding.lower() in ("iso-8859-1", "ascii"):
resp.encoding = resp.apparent_encoding or "utf-8"
text = resp.text
if text.startswith("\ufeff"):
text = text[1:]
return text.strip()
def parse_json_response(resp: requests.Response, *, source: str = "平台") -> dict:
"""
将 HTTP 响应解析为 JSON 对象。
若返回 HTML / 空内容 / 非 JSON,抛出带说明的 RuntimeError。
"""
text = response_text(resp)
if not text:
raise RuntimeError(
f"{source} 返回空内容(HTTP {resp.status_code}),"
"请稍后重试或检查网络。"
)
lower = text.lower()
if lower.startswith("<!doctype") or lower.startswith("<html") or "<html" in lower[:200]:
if "频繁" in text or "访问过于频繁" in text:
raise RuntimeError(
f"{source} 提示访问过于频繁,请增大请求间隔后重试。"
)
raise RuntimeError(
f"{source} 返回了网页而非数据(HTTP {resp.status_code}),"
"可能被限流或拦截,请稍后重试。"
)
try:
data = json.loads(text)
except json.JSONDecodeError as exc:
preview = text[:200].replace("\n", " ")
raise RuntimeError(
f"{source} 响应不是合法 JSON:{exc};内容开头:{preview!r}"
) from exc
if not isinstance(data, dict):
raise RuntimeError(f"{source} 响应格式异常(期望 JSON 对象)。")
return data