100 lines
3.7 KiB
Python
100 lines
3.7 KiB
Python
"""余额获取：模板渲染 → HTTP 请求（带重试）→ JSON 路径提取。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
from dataclasses import dataclass
|
|
|
|
import requests
|
|
|
|
from app.expr import extract_path, to_number
|
|
from app.providers.base import BalanceProvider
|
|
|
|
logger = logging.getLogger("monitor.fetcher")
|
|
|
|
PLACEHOLDER = "{{apiKey}}"
|
|
|
|
|
|
@dataclass
|
|
class FetchResult:
|
|
"""一次拉取的结果。"""
|
|
|
|
ok: bool
|
|
balance: float | None = None
|
|
auth_error: bool = False
|
|
status_code: int | None = None
|
|
error: str = ""
|
|
|
|
|
|
def render_template(text: str, api_key: str) -> str:
|
|
"""把模板中的 {{apiKey}} 替换为账号密钥。"""
|
|
return text.replace(PLACEHOLDER, api_key)
|
|
|
|
|
|
def extract_balance(data: object, path: str) -> float:
|
|
"""按点路径/数组索引提取余额（兼容旧写法，返回 float）。
|
|
|
|
新写法请用 evaluate_balance（支持运算符与函数）。
|
|
"""
|
|
return to_number(extract_path(data, path), path)
|
|
|
|
|
|
def fetch_balance(
|
|
provider: BalanceProvider,
|
|
api_key: str,
|
|
retry_count: int,
|
|
timeout: int,
|
|
) -> FetchResult:
|
|
"""执行一次余额拉取。
|
|
|
|
策略：
|
|
- 401/403 → 先按重试次数确认（可能瞬时），仍失败 → auth_error（调用方禁用账号并通知）
|
|
- 其他 4xx → 直接失败（不重试）
|
|
- 网络异常 / 5xx → 重试 retry_count 次，间隔 2s
|
|
- JSON 解析/余额提取失败 → 直接失败
|
|
"""
|
|
url, headers, body = provider.build_request(api_key)
|
|
method = provider.method
|
|
if method == "POST" and body and not any(k.lower() == "content-type" for k in headers):
|
|
headers["Content-Type"] = "application/json"
|
|
attempts = retry_count + 1
|
|
for attempt in range(attempts):
|
|
try:
|
|
if method == "GET":
|
|
resp = requests.get(url, headers=headers, timeout=timeout)
|
|
else:
|
|
resp = requests.post(url, headers=headers, data=body, timeout=timeout)
|
|
except requests.RequestException as exc:
|
|
logger.warning("请求异常(%s/%s) %s: %s", attempt + 1, attempts, url, exc)
|
|
if attempt < attempts - 1:
|
|
time.sleep(2)
|
|
continue
|
|
if resp.status_code in (401, 403):
|
|
# 认证失败可能是瞬时的（key 刚生效/风控），先按重试次数再确认，仍失败才判定
|
|
if attempt < attempts - 1:
|
|
logger.warning("认证失败(%s/%s) %s status=%s，重试确认", attempt + 1, attempts, url, resp.status_code)
|
|
time.sleep(1)
|
|
continue
|
|
return FetchResult(ok=False, auth_error=True, status_code=resp.status_code,
|
|
error=f"HTTP {resp.status_code}（认证失败）")
|
|
if resp.status_code >= 500:
|
|
logger.warning("服务端错误(%s/%s) %s status=%s", attempt + 1, attempts, url, resp.status_code)
|
|
if attempt < attempts - 1:
|
|
time.sleep(2)
|
|
continue
|
|
if resp.status_code >= 400:
|
|
return FetchResult(ok=False, status_code=resp.status_code,
|
|
error=f"HTTP {resp.status_code}: {resp.text[:120]}")
|
|
try:
|
|
data = resp.json()
|
|
except ValueError:
|
|
return FetchResult(ok=False, status_code=resp.status_code,
|
|
error="响应不是合法 JSON")
|
|
try:
|
|
balance = provider.extract_balance(data)
|
|
except Exception as exc:
|
|
return FetchResult(ok=False, status_code=resp.status_code, error=str(exc))
|
|
return FetchResult(ok=True, balance=balance, status_code=resp.status_code)
|
|
return FetchResult(ok=False, error=f"网络错误，重试 {retry_count} 次后仍失败")
|