66 lines
2.0 KiB
Python
66 lines
2.0 KiB
Python
"""带限速、重试、SSL 宽松回退的 HTTP 抓取。"""
|
||
|
||
import time
|
||
|
||
import requests
|
||
import urllib3
|
||
|
||
from config import MAX_RETRIES, REQUEST_DELAY, REQUEST_TIMEOUT, USER_AGENT
|
||
|
||
# 站点证书在部分环境下校验失败（curl 需 -k），先严格再回退宽松
|
||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||
|
||
_session = requests.Session()
|
||
_session.headers.update({"User-Agent": USER_AGENT})
|
||
|
||
_last_request_ts = 0.0
|
||
|
||
|
||
def _throttle() -> None:
|
||
"""简单限速：保证两次请求之间至少间隔 REQUEST_DELAY 秒。"""
|
||
global _last_request_ts
|
||
now = time.monotonic()
|
||
wait = REQUEST_DELAY - (now - _last_request_ts)
|
||
if wait > 0:
|
||
time.sleep(wait)
|
||
_last_request_ts = time.monotonic()
|
||
|
||
|
||
def _get(url: str, **kwargs) -> requests.Response:
|
||
"""GET，SSL 校验失败时自动回退到不校验证书。"""
|
||
try:
|
||
return _session.get(url, timeout=REQUEST_TIMEOUT, verify=True, **kwargs)
|
||
except requests.exceptions.SSLError:
|
||
return _session.get(url, timeout=REQUEST_TIMEOUT, verify=False, **kwargs)
|
||
|
||
|
||
def fetch_text(url: str) -> str:
|
||
"""抓取并返回 UTF-8 文本，带重试。"""
|
||
last_err: Exception | None = None
|
||
for attempt in range(1, MAX_RETRIES + 1):
|
||
_throttle()
|
||
try:
|
||
resp = _get(url)
|
||
resp.raise_for_status()
|
||
resp.encoding = resp.apparent_encoding or "utf-8"
|
||
return resp.text
|
||
except requests.RequestException as e:
|
||
last_err = e
|
||
time.sleep(0.8 * attempt)
|
||
raise RuntimeError(f"抓取失败 {url}: {last_err}")
|
||
|
||
|
||
def fetch_bytes(url: str) -> bytes:
|
||
"""抓取二进制（图片等），带重试。"""
|
||
last_err: Exception | None = None
|
||
for attempt in range(1, MAX_RETRIES + 1):
|
||
_throttle()
|
||
try:
|
||
resp = _get(url)
|
||
resp.raise_for_status()
|
||
return resp.content
|
||
except requests.RequestException as e:
|
||
last_err = e
|
||
time.sleep(0.8 * attempt)
|
||
raise RuntimeError(f"抓取失败 {url}: {last_err}")
|