strepsiades
628 lines
22 KiB
Python
628 lines
22 KiB
Python
"""
|
||
ApiClient - HTTP 通信层
|
||
|
||
职责:
|
||
- 封装所有 HTTP 通信逻辑
|
||
- 管理 httpx.AsyncClient，维护连接池和 session
|
||
- 生成请求签名（timestamp, nonce, sign）
|
||
- 统一请求入口，自动注入 uid
|
||
- 批量 Push Log 查询
|
||
"""
|
||
|
||
import hashlib
|
||
import hmac
|
||
import time
|
||
import uuid
|
||
import json as json_lib
|
||
import logging
|
||
import asyncio
|
||
from typing import Dict, List, Any, Optional
|
||
from collections import defaultdict, deque
|
||
from pathlib import Path
|
||
from datetime import datetime
|
||
import httpx
|
||
|
||
from schemas.common.push_system import PushIdsSchema
|
||
|
||
|
||
def _build_default_push_steps() -> List[Dict[str, str]]:
|
||
"""构造最小审批步骤字段，满足第三方 v2 push schema。"""
|
||
return []
|
||
|
||
|
||
class ApiClient:
|
||
"""API HTTP 客户端"""
|
||
_TRACE_TEXT_MAX_LEN = 1000
|
||
|
||
def __init__(
|
||
self,
|
||
base_url: str,
|
||
uid: str,
|
||
secret: str,
|
||
debug: bool = False,
|
||
log_file: Optional[Path] = None,
|
||
rate_limit_max_requests: int = 30,
|
||
rate_limit_window_seconds: float = 10.0,
|
||
):
|
||
"""
|
||
初始化 API 客户端
|
||
|
||
Args:
|
||
base_url: API 基础 URL（如 https://api.example.com）
|
||
uid: 用户标识
|
||
secret: 签名密钥
|
||
debug: 是否启用调试模式（打印请求详情）
|
||
log_file: 日志文件路径（可选）
|
||
rate_limit_max_requests: 限流窗口内最大请求数（<=0 表示禁用）
|
||
rate_limit_window_seconds: 限流窗口秒数（<=0 表示禁用）
|
||
"""
|
||
self._base_url = base_url.rstrip('/')
|
||
self._uid = uid
|
||
self._secret = secret
|
||
self._client: Optional[httpx.AsyncClient] = None
|
||
self._debug = debug
|
||
self._log_file = log_file
|
||
self._trace_seq = 0
|
||
self._trace_by_push_id: Dict[str, deque[Dict[str, Any]]] = defaultdict(deque)
|
||
self._trace_by_biz_id: Dict[str, deque[Dict[str, Any]]] = defaultdict(deque)
|
||
self._trace_by_op: Dict[str, deque[Dict[str, Any]]] = defaultdict(deque)
|
||
self._load_trace_by_item: Dict[tuple[str, str], deque[Dict[str, Any]]] = defaultdict(deque)
|
||
self._logger = logging.getLogger(__name__)
|
||
self._rate_limit_max_requests = int(rate_limit_max_requests)
|
||
self._rate_limit_window_seconds = float(rate_limit_window_seconds)
|
||
self._rate_limit_lock = asyncio.Lock()
|
||
self._rate_limit_timestamps: deque[float] = deque()
|
||
|
||
def _is_rate_limit_enabled(self) -> bool:
|
||
return self._rate_limit_max_requests > 0 and self._rate_limit_window_seconds > 0
|
||
|
||
async def _acquire_rate_limit_slot(self) -> None:
|
||
if not self._is_rate_limit_enabled():
|
||
return
|
||
|
||
while True:
|
||
wait_seconds = 0.0
|
||
async with self._rate_limit_lock:
|
||
now = time.monotonic()
|
||
window_start = now - self._rate_limit_window_seconds
|
||
while self._rate_limit_timestamps and self._rate_limit_timestamps[0] <= window_start:
|
||
self._rate_limit_timestamps.popleft()
|
||
|
||
if len(self._rate_limit_timestamps) < self._rate_limit_max_requests:
|
||
self._rate_limit_timestamps.append(now)
|
||
return
|
||
|
||
oldest = self._rate_limit_timestamps[0]
|
||
wait_seconds = max(0.001, self._rate_limit_window_seconds - (now - oldest))
|
||
|
||
await asyncio.sleep(wait_seconds)
|
||
|
||
async def initialize(self):
|
||
"""初始化 HTTP 客户端"""
|
||
if self._client is None:
|
||
# trust_env=False 阻止从环境变量读取代理设置（避免 localhost 走代理导致超时）
|
||
self._client = httpx.AsyncClient(timeout=300.0, trust_env=False)
|
||
|
||
async def close(self):
|
||
"""关闭 HTTP 客户端"""
|
||
if self._client:
|
||
await self._client.aclose()
|
||
self._client = None
|
||
|
||
def _log(self, message: str, level: str = "DEBUG"):
|
||
"""
|
||
统一日志输出
|
||
|
||
Args:
|
||
message: 日志消息
|
||
level: 日志级别（INFO/ERROR）
|
||
"""
|
||
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
log_line = f"[{timestamp}] [{level}] {message}"
|
||
|
||
level_num = logging.getLevelNamesMapping().get(level.upper(), logging.DEBUG)
|
||
self._logger.log(level_num, log_line)
|
||
|
||
# 文件输出
|
||
if self._log_file:
|
||
self._log_file.parent.mkdir(parents=True, exist_ok=True)
|
||
with open(self._log_file, 'a', encoding='utf-8') as f:
|
||
f.write(log_line + '\n')
|
||
|
||
def _get_client(self) -> httpx.AsyncClient:
|
||
"""获取 HTTP 客户端实例"""
|
||
if self._client is None:
|
||
raise RuntimeError("ApiClient not initialized. Call 'await client.initialize()' first.")
|
||
return self._client
|
||
|
||
def _derive_op_kind(self, method: str, endpoint: str) -> str:
|
||
endpoint_norm = endpoint.lower().strip()
|
||
if endpoint_norm == "/push/result":
|
||
return "poll"
|
||
method_upper = method.upper()
|
||
if method_upper == "GET":
|
||
if endpoint_norm.endswith("/list"):
|
||
return "load"
|
||
return "get"
|
||
if endpoint_norm.endswith("/create"):
|
||
return "create"
|
||
if endpoint_norm.endswith("/delete"):
|
||
return "delete"
|
||
if endpoint_norm.endswith("/update") or endpoint_norm.endswith("/plan"):
|
||
return "update"
|
||
if method_upper == "POST":
|
||
return "create"
|
||
if method_upper == "PUT":
|
||
return "update"
|
||
if method_upper == "DELETE":
|
||
return "delete"
|
||
return method.lower()
|
||
|
||
def _derive_node_type_from_list_endpoint(self, endpoint: str) -> Optional[str]:
|
||
endpoint_norm = endpoint.strip("/")
|
||
if not endpoint_norm.endswith("/list"):
|
||
return None
|
||
prefix = endpoint_norm[:-5]
|
||
if not prefix:
|
||
return None
|
||
return prefix.replace("/", "_")
|
||
|
||
def _record_trace(
|
||
self,
|
||
*,
|
||
method: str,
|
||
endpoint: str,
|
||
url: str,
|
||
request_params: Optional[Dict[str, Any]],
|
||
request_payload: Optional[Dict[str, Any]],
|
||
response: Optional[Dict[str, Any]],
|
||
elapsed_ms: int,
|
||
) -> None:
|
||
self._trace_seq += 1
|
||
op_kind = self._derive_op_kind(method, endpoint)
|
||
payload = request_payload or {}
|
||
response_view = self._build_trace_response_view(op_kind=op_kind, response=response)
|
||
params_text = self._to_trace_text(request_params or {}, max_len=self._TRACE_TEXT_MAX_LEN)
|
||
payload_text = self._to_trace_text(payload, max_len=self._TRACE_TEXT_MAX_LEN)
|
||
response_text = self._to_trace_text(response_view, max_len=self._TRACE_TEXT_MAX_LEN)
|
||
|
||
trace = {
|
||
"trace_id": self._trace_seq,
|
||
"op": op_kind,
|
||
"method": method.upper(),
|
||
"endpoint": endpoint,
|
||
"url": url,
|
||
"params": request_params or {},
|
||
"payload": payload,
|
||
"response": response_view,
|
||
"params_text": params_text,
|
||
"payload_text": payload_text,
|
||
"response_text": response_text,
|
||
"elapsed_ms": elapsed_ms,
|
||
"ts": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
}
|
||
|
||
push_id = payload.get("push_id")
|
||
biz_id = payload.get("biz_id")
|
||
if isinstance(push_id, str) and push_id:
|
||
self._trace_by_push_id[push_id].append(trace)
|
||
if isinstance(biz_id, str) and biz_id:
|
||
self._trace_by_biz_id[biz_id].append(trace)
|
||
self._trace_by_op[op_kind].append(trace)
|
||
|
||
if op_kind == "load":
|
||
node_type = self._derive_node_type_from_list_endpoint(endpoint)
|
||
data = (response or {}).get("data") if isinstance(response, dict) else None
|
||
if isinstance(data, list):
|
||
items = data
|
||
elif isinstance(data, dict):
|
||
items = data.get("list", [])
|
||
else:
|
||
items = []
|
||
if node_type and isinstance(items, list):
|
||
for item in items:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
item_id = item.get("id") or item.get("_id")
|
||
if not item_id:
|
||
continue
|
||
self._load_trace_by_item[(node_type, str(item_id))].append(trace)
|
||
|
||
def _to_trace_text(self, value: Any, *, max_len: int) -> str:
|
||
try:
|
||
text = json_lib.dumps(value, ensure_ascii=False, separators=(",", ":"))
|
||
except Exception:
|
||
text = str(value)
|
||
if len(text) > max_len:
|
||
return f"{text[:max_len]}...(已截断)"
|
||
return text
|
||
|
||
def _build_trace_response_view(self, *, op_kind: str, response: Optional[Dict[str, Any]]) -> Any:
|
||
return response or {}
|
||
|
||
def consume_traces_by_push_id(self, push_id: str) -> List[Dict[str, Any]]:
|
||
if not push_id:
|
||
return []
|
||
bucket = self._trace_by_push_id.get(push_id)
|
||
if not bucket:
|
||
return []
|
||
traces = list(bucket)
|
||
bucket.clear()
|
||
return traces
|
||
|
||
def consume_latest_trace_by_biz_id(self, biz_id: str, op_kind: Optional[str] = None) -> Optional[Dict[str, Any]]:
|
||
if not biz_id:
|
||
return None
|
||
bucket = self._trace_by_biz_id.get(biz_id)
|
||
if not bucket:
|
||
return None
|
||
if op_kind is None:
|
||
return bucket.pop() if bucket else None
|
||
for index in range(len(bucket) - 1, -1, -1):
|
||
trace = bucket[index]
|
||
if trace.get("op") == op_kind:
|
||
del bucket[index]
|
||
return trace
|
||
return None
|
||
|
||
def peek_latest_trace_by_biz_id(self, biz_id: str, op_kind: Optional[str] = None) -> Optional[Dict[str, Any]]:
|
||
if not biz_id:
|
||
return None
|
||
bucket = self._trace_by_biz_id.get(biz_id)
|
||
if not bucket:
|
||
return None
|
||
if op_kind is None:
|
||
return bucket[-1] if bucket else None
|
||
for index in range(len(bucket) - 1, -1, -1):
|
||
trace = bucket[index]
|
||
if trace.get("op") == op_kind:
|
||
return trace
|
||
return None
|
||
|
||
def consume_latest_trace_by_op(self, op_kind: str) -> Optional[Dict[str, Any]]:
|
||
bucket = self._trace_by_op.get(op_kind)
|
||
if not bucket:
|
||
return None
|
||
return bucket.pop() if bucket else None
|
||
|
||
def peek_latest_trace_by_op(self, op_kind: str) -> Optional[Dict[str, Any]]:
|
||
bucket = self._trace_by_op.get(op_kind)
|
||
if not bucket:
|
||
return None
|
||
return bucket[-1] if bucket else None
|
||
|
||
def consume_load_traces(self, node_type: str, data_id: str) -> List[Dict[str, Any]]:
|
||
key = (node_type, data_id)
|
||
bucket = self._load_trace_by_item.get(key)
|
||
if not bucket:
|
||
return []
|
||
traces = list(bucket)
|
||
bucket.clear()
|
||
return traces
|
||
|
||
def _generate_signature(self, params: Dict[str, Any]) -> Dict[str, str]:
|
||
"""
|
||
生成签名参数（兼容旧版签名算法）
|
||
|
||
Args:
|
||
params: 请求参数
|
||
|
||
Returns:
|
||
包含 timestamp, nonce, sign 的字典
|
||
"""
|
||
timestamp = int(time.time() * 1000) # 毫秒时间戳
|
||
nonce = uuid.uuid4().hex[:16] # 16位随机字符串
|
||
|
||
# 构建签名参数
|
||
sign_params = params.copy()
|
||
sign_params['timestamp'] = timestamp
|
||
sign_params['nonce'] = nonce
|
||
|
||
# 按key排序
|
||
sorted_items = sorted(sign_params.items())
|
||
|
||
# 构建签名字符串
|
||
items = []
|
||
for k, v in sorted_items:
|
||
if v is None:
|
||
v_str = "null"
|
||
elif isinstance(v, bool):
|
||
v_str = "true" if v else "false"
|
||
elif isinstance(v, str):
|
||
v_str = v
|
||
elif isinstance(v, (int, float)):
|
||
v_str = str(v)
|
||
elif isinstance(v, (list, dict)):
|
||
v_str = json_lib.dumps(v, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
|
||
else:
|
||
v_str = str(v)
|
||
items.append(f"{k}={v_str}")
|
||
|
||
sign_str = "&".join(items)
|
||
|
||
# 使用 HMAC-SHA256 生成签名
|
||
sign = hmac.new(self._secret.encode("utf-8"), sign_str.encode("utf-8"), hashlib.sha256).hexdigest()
|
||
|
||
return {
|
||
'timestamp': str(timestamp),
|
||
'nonce': nonce,
|
||
'sign': sign
|
||
}
|
||
|
||
async def request(
|
||
self,
|
||
method: str,
|
||
endpoint: str,
|
||
params: Optional[Dict[str, Any]] = None,
|
||
data: Optional[Dict[str, Any]] = None,
|
||
json_data: Optional[Dict[str, Any]] = None,
|
||
json: Optional[Dict[str, Any]] = None
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
统一请求入口，自动签名
|
||
|
||
Args:
|
||
method: HTTP 方法（GET/POST/PUT/DELETE）
|
||
endpoint: API 端点（如 /api/v2/project）
|
||
params: URL 查询参数
|
||
data: 请求体数据（form data）
|
||
json_data: JSON 请求体数据
|
||
json: JSON 请求体数据（兼容参数）
|
||
|
||
Returns:
|
||
响应 JSON 数据
|
||
|
||
Raises:
|
||
httpx.HTTPStatusError: HTTP 错误
|
||
"""
|
||
# 兼容json参数
|
||
if json is not None and json_data is None:
|
||
json_data = json
|
||
|
||
client = self._get_client()
|
||
url = f"{self._base_url}{endpoint}"
|
||
await self._acquire_rate_limit_slot()
|
||
|
||
# 记录请求信息
|
||
self._log(f"\n{'='*60}", level="DEBUG")
|
||
self._log(f"REQUEST: {method} {endpoint}", level="DEBUG")
|
||
self._log(f"Full URL: {url}", level="DEBUG")
|
||
if params:
|
||
self._log(f"Params: {json_lib.dumps(params, ensure_ascii=False, indent=2)}", level="DEBUG")
|
||
if data:
|
||
self._log(f"Data: {json_lib.dumps(data, ensure_ascii=False, indent=2)}", level="DEBUG")
|
||
if json_data:
|
||
self._log(f"JSON: {json_lib.dumps(json_data, ensure_ascii=False, indent=2)}", level="DEBUG")
|
||
|
||
# 准备签名参数
|
||
if method.upper() == 'GET':
|
||
# GET 请求：uid 放在 params 中
|
||
if params is None:
|
||
params = {}
|
||
params['uid'] = self._uid
|
||
|
||
# 转换所有参数值为字符串（除了 list/dict）
|
||
params = {k: str(v) if not isinstance(v, (list, dict)) else v for k, v in params.items()}
|
||
|
||
sign_params = params.copy()
|
||
# 生成签名并附加到 params
|
||
signature = self._generate_signature(sign_params)
|
||
params.update(signature)
|
||
else:
|
||
# POST/PUT/DELETE：uid 和签名放在 body 中（符合 v2 schema）
|
||
if data is None:
|
||
data = {}
|
||
if 'steps' not in data:
|
||
data['steps'] = _build_default_push_steps()
|
||
data['uid'] = self._uid
|
||
sign_params = data.copy()
|
||
if params:
|
||
sign_params.update(params)
|
||
signature = self._generate_signature(sign_params)
|
||
data.update(signature)
|
||
|
||
# 发送请求并增强异常处理
|
||
start_time = time.time()
|
||
try:
|
||
response = await client.request(
|
||
method=method,
|
||
url=url,
|
||
params=params,
|
||
json=data if method.upper() in ['POST', 'PUT'] else None
|
||
)
|
||
|
||
# 记录响应时间
|
||
elapsed = time.time() - start_time
|
||
self._log(f"Response Time: {elapsed:.3f}s", level="DEBUG")
|
||
|
||
# 检查 HTTP 状态
|
||
response.raise_for_status()
|
||
|
||
# 记录响应
|
||
result = response.json()
|
||
self._log(f"Status Code: {response.status_code}", level="DEBUG")
|
||
self._log(f"Response: {json_lib.dumps(result, ensure_ascii=False, indent=2)}", level="DEBUG")
|
||
self._log(f"{'='*60}\n", level="DEBUG")
|
||
|
||
self._record_trace(
|
||
method=method,
|
||
endpoint=endpoint,
|
||
url=url,
|
||
request_params=params,
|
||
request_payload=data if method.upper() in ['POST', 'PUT', 'DELETE'] else None,
|
||
response=result,
|
||
elapsed_ms=int(elapsed * 1000),
|
||
)
|
||
|
||
return result
|
||
|
||
except Exception as e:
|
||
import traceback
|
||
|
||
# 记录错误
|
||
elapsed = time.time() - start_time
|
||
error_msg = "\n========== [ApiClient HTTP Error] ==========\n"
|
||
error_msg += f"Request: {method} {url}\n"
|
||
error_msg += f"Params: {params}\n"
|
||
if method.upper() in ['POST', 'PUT']:
|
||
error_msg += f"JSON: {data}\n"
|
||
error_msg += f"Exception: {type(e).__name__}: {e}\n"
|
||
error_msg += f"Elapsed: {elapsed:.3f}s\n"
|
||
|
||
response_obj = e.response if isinstance(e, httpx.HTTPStatusError) else None
|
||
if response_obj is not None:
|
||
error_msg += f"Status Code: {response_obj.status_code}\n"
|
||
error_msg += f"Response Headers: {response_obj.headers}\n"
|
||
try:
|
||
error_msg += f"Response Text: {response_obj.text}\n"
|
||
except Exception:
|
||
pass
|
||
|
||
error_msg += traceback.format_exc()
|
||
error_msg += "============================================\n"
|
||
|
||
# 记录到日志文件
|
||
self._log(error_msg, level="ERROR")
|
||
|
||
raise
|
||
|
||
async def get(self, endpoint: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||
"""
|
||
GET 请求
|
||
|
||
Args:
|
||
endpoint: API 端点
|
||
params: URL 查询参数
|
||
|
||
Returns:
|
||
响应 JSON 数据
|
||
"""
|
||
return await self.request('GET', endpoint, params=params)
|
||
|
||
async def post(self, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""
|
||
POST 请求，data 中已包含 push_id/biz_id（由 Handler 生成）
|
||
|
||
Args:
|
||
endpoint: API 端点
|
||
data: 请求体数据（包含业务数据和 push_id/biz_id）
|
||
|
||
Returns:
|
||
响应 JSON 数据
|
||
"""
|
||
return await self.request('POST', endpoint, data=data)
|
||
|
||
async def put(self, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""
|
||
PUT 请求，data 中已包含 push_id/biz_id
|
||
|
||
Args:
|
||
endpoint: API 端点
|
||
data: 请求体数据
|
||
|
||
Returns:
|
||
响应 JSON 数据
|
||
"""
|
||
return await self.request('PUT', endpoint, data=data)
|
||
|
||
async def delete(self, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""
|
||
DELETE 请求，data 中已包含 push_id/biz_id
|
||
|
||
Args:
|
||
endpoint: API 端点
|
||
data: 请求体数据
|
||
|
||
Returns:
|
||
响应 JSON 数据
|
||
"""
|
||
return await self.request('DELETE', endpoint, data=data)
|
||
|
||
async def get_push_logs(self, push_ids: List[str]) -> Dict[str, Dict[str, Any]]:
|
||
"""
|
||
批量查询 Push Log
|
||
|
||
注意：单个查询也传单元素列表
|
||
当前实现：用 for 循环模拟，后续等后端实现批量接口
|
||
|
||
Args:
|
||
push_ids: Push ID 列表
|
||
|
||
Returns:
|
||
{push_id: push_log_data} 的映射
|
||
"""
|
||
result: Dict[str, Dict[str, Any]] = {}
|
||
|
||
if not push_ids:
|
||
return result
|
||
|
||
try:
|
||
payload = PushIdsSchema(push_ids=','.join(push_ids)).model_dump(exclude_none=True)
|
||
response = await self.request(
|
||
method='POST',
|
||
endpoint='/push/result',
|
||
data=payload
|
||
)
|
||
except httpx.HTTPStatusError as e:
|
||
return {push_id: {'error': str(e)} for push_id in push_ids}
|
||
|
||
data_list: List[Dict[str, Any]] = []
|
||
if isinstance(response, dict):
|
||
raw_data = response.get('data')
|
||
if isinstance(raw_data, list):
|
||
data_list = [item for item in raw_data if isinstance(item, dict)]
|
||
elif isinstance(raw_data, dict):
|
||
data_list = [raw_data]
|
||
elif 'push_id' in response and 'status' in response:
|
||
data_list = [response]
|
||
|
||
for item in data_list:
|
||
push_id = item.get('push_id')
|
||
if not push_id:
|
||
continue
|
||
|
||
poll_trace = {
|
||
"trace_id": f"poll-{push_id}-{int(time.time() * 1000)}",
|
||
"op": "poll",
|
||
"method": "POST",
|
||
"endpoint": "/push/result",
|
||
"url": f"{self._base_url}/push/result",
|
||
"params": {},
|
||
"payload": {"push_ids": push_id},
|
||
"response": item,
|
||
"elapsed_ms": 0,
|
||
"ts": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
}
|
||
self._trace_by_push_id[push_id].append(poll_trace)
|
||
|
||
status = (item.get('status') or '').lower()
|
||
error_msg = item.get('error') or ''
|
||
result_data = item.get('result') or {}
|
||
|
||
if status == 'success':
|
||
# result_data 的结构是 {"code": 200, "message": "OK", "data": {"biz_id": "..."}}
|
||
# 需要提取 result_data["data"] 作为最终的 data
|
||
inner_data = result_data.get('data', {}) if isinstance(result_data, dict) else {}
|
||
result[push_id] = {
|
||
'code': 200,
|
||
'status': 'success',
|
||
'data': inner_data
|
||
}
|
||
elif status in {'failed', 'not_find'}:
|
||
result[push_id] = {
|
||
'code': 500,
|
||
'status': 'failed',
|
||
'message': error_msg or status or 'Unknown error',
|
||
'error': error_msg or status or 'Unknown error'
|
||
}
|
||
else:
|
||
result[push_id] = {
|
||
'status': status or 'processing'
|
||
}
|
||
|
||
for push_id in push_ids:
|
||
if push_id not in result:
|
||
result[push_id] = {'status': 'processing'}
|
||
|
||
return result
|