186 lines
6.8 KiB
Python
186 lines
6.8 KiB
Python
|
# =============================================================================
|
|||
|
|
# 企微IT智能服务台 — 阶段5 自动化 外部客户端基类
|
|||
|
|
# =============================================================================
|
|||
|
|
# 说明：提供异步 HTTP 客户端通用能力：
|
|||
|
|
# 1. 超时控制（连接/读取分开）
|
|||
|
|
# 2. 重试（tenacity：指数退避 + 抖动，仅重试超时/5xx，4xx 不重试）
|
|||
|
|
# 3. 审计钩子（出入参记录到 ActionLog，由调用方注入回调）
|
|||
|
|
#
|
|||
|
|
# 所有外部客户端（Dify / EHR / 未来 aTrust）均继承此类，
|
|||
|
|
# 保证超时、重试、审计行为在全项目一致，且方法签名完整、异常可捕获、
|
|||
|
|
# 出入参可被 ActionLog 记录（满足架构审计要求）。
|
|||
|
|
# =============================================================================
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import logging
|
|||
|
|
import time
|
|||
|
|
from typing import Any, Awaitable, Callable, Dict, Optional
|
|||
|
|
|
|||
|
|
import httpx
|
|||
|
|
from tenacity import (
|
|||
|
|
AsyncRetrying,
|
|||
|
|
retry_if_exception_type,
|
|||
|
|
stop_after_attempt,
|
|||
|
|
wait_exponential_jitter,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
logger = logging.getLogger(__name__)
|
|||
|
|
|
|||
|
|
# 默认超时（秒）
|
|||
|
|
DEFAULT_CONNECT_TIMEOUT = 5.0
|
|||
|
|
DEFAULT_READ_TIMEOUT = 20.0
|
|||
|
|
# 默认重试次数
|
|||
|
|
DEFAULT_MAX_ATTEMPTS = 3
|
|||
|
|
|
|||
|
|
|
|||
|
|
class BaseClientError(Exception):
|
|||
|
|
"""外部客户端通用异常（可被自动化引擎捕获并转人工）。"""
|
|||
|
|
|
|||
|
|
def __init__(self, message: str, code: int = -1, detail: str = ""):
|
|||
|
|
super().__init__(message)
|
|||
|
|
self.message = message
|
|||
|
|
self.code = code
|
|||
|
|
self.detail = detail
|
|||
|
|
|
|||
|
|
|
|||
|
|
class BaseClient:
|
|||
|
|
"""异步外部客户端基类。
|
|||
|
|
|
|||
|
|
Attributes:
|
|||
|
|
base_url: 外部系统基址（不含尾部斜杠）
|
|||
|
|
timeout: httpx 超时配置
|
|||
|
|
_audit: 审计回调（可选），签名
|
|||
|
|
async (event, direction, system, request, response, status, latency_ms, error)
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
# 子类声明系统标识，用于审计与日志
|
|||
|
|
system_name: str = "external"
|
|||
|
|
|
|||
|
|
def __init__(
|
|||
|
|
self,
|
|||
|
|
base_url: str,
|
|||
|
|
timeout: Optional[float] = None,
|
|||
|
|
audit: Optional[Callable[..., Awaitable[None]]] = None,
|
|||
|
|
):
|
|||
|
|
self.base_url = (base_url or "").rstrip("/")
|
|||
|
|
read = timeout or DEFAULT_READ_TIMEOUT
|
|||
|
|
self.timeout = httpx.Timeout(connect=DEFAULT_CONNECT_TIMEOUT, read=read)
|
|||
|
|
self._audit = audit
|
|||
|
|
|
|||
|
|
async def _emit_audit(
|
|||
|
|
self,
|
|||
|
|
event: str,
|
|||
|
|
direction: str,
|
|||
|
|
request: Any = None,
|
|||
|
|
response: Any = None,
|
|||
|
|
status: str = "",
|
|||
|
|
latency_ms: Optional[int] = None,
|
|||
|
|
error: str = "",
|
|||
|
|
) -> None:
|
|||
|
|
"""审计钩子：记录出入参（落 ActionLog）。
|
|||
|
|
|
|||
|
|
为什么单独成方法：审计失败绝不影响主流程（仅记 warning），
|
|||
|
|
避免外部审计系统抖动拖累自动化处置。
|
|||
|
|
"""
|
|||
|
|
if self._audit is None:
|
|||
|
|
return
|
|||
|
|
try:
|
|||
|
|
await self._audit(
|
|||
|
|
event=event,
|
|||
|
|
direction=direction,
|
|||
|
|
system=self.system_name,
|
|||
|
|
request=request,
|
|||
|
|
response=response,
|
|||
|
|
status=status,
|
|||
|
|
latency_ms=latency_ms,
|
|||
|
|
error=error,
|
|||
|
|
)
|
|||
|
|
except Exception as e: # 审计失败不影响主流程
|
|||
|
|
logger.warning(f"[{self.system_name}] 审计钩子执行失败: {e}")
|
|||
|
|
|
|||
|
|
async def request(
|
|||
|
|
self,
|
|||
|
|
method: str,
|
|||
|
|
path: str,
|
|||
|
|
*,
|
|||
|
|
json_data: Optional[Dict] = None,
|
|||
|
|
params: Optional[Dict] = None,
|
|||
|
|
headers: Optional[Dict] = None,
|
|||
|
|
event: str = "",
|
|||
|
|
retry: bool = True,
|
|||
|
|
) -> Dict[str, Any]:
|
|||
|
|
"""统一请求封装（带超时 + 重试 + 审计）。
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
method: HTTP 方法
|
|||
|
|
path: 路径（自动拼接 base_url）
|
|||
|
|
json_data/params/headers: 请求参数
|
|||
|
|
event: 审计事件名（如 "dify.intent"）
|
|||
|
|
retry: 是否启用重试（仅对超时/5xx 重试，4xx 不重试）
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
Dict: 解析后的 JSON 响应
|
|||
|
|
|
|||
|
|
Raises:
|
|||
|
|
BaseClientError: 网络/HTTP/业务错误
|
|||
|
|
"""
|
|||
|
|
url = f"{self.base_url}{path}"
|
|||
|
|
start = time.monotonic()
|
|||
|
|
|
|||
|
|
async def _do() -> httpx.Response:
|
|||
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|||
|
|
return await client.request(
|
|||
|
|
method, url, json=json_data, params=params, headers=headers
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
if retry:
|
|||
|
|
resp: Optional[httpx.Response] = None
|
|||
|
|
async for attempt in AsyncRetrying(
|
|||
|
|
stop=stop_after_attempt(DEFAULT_MAX_ATTEMPTS),
|
|||
|
|
wait=wait_exponential_jitter(initial=0.5, max=3.0),
|
|||
|
|
retry=retry_if_exception_type(
|
|||
|
|
(httpx.TimeoutException, httpx.ConnectError, httpx.HTTPStatusError)
|
|||
|
|
),
|
|||
|
|
reraise=True,
|
|||
|
|
):
|
|||
|
|
with attempt:
|
|||
|
|
resp = await _do()
|
|||
|
|
# 4xx 是业务错误，不重试
|
|||
|
|
if resp.status_code >= 400 and resp.status_code < 500:
|
|||
|
|
raise httpx.HTTPStatusError(
|
|||
|
|
message=f"HTTP {resp.status_code}",
|
|||
|
|
request=resp.request,
|
|||
|
|
response=resp,
|
|||
|
|
)
|
|||
|
|
assert resp is not None
|
|||
|
|
else:
|
|||
|
|
resp = await _do()
|
|||
|
|
|
|||
|
|
latency_ms = int((time.monotonic() - start) * 1000)
|
|||
|
|
try:
|
|||
|
|
data = resp.json()
|
|||
|
|
except Exception:
|
|||
|
|
data = {"raw": resp.text}
|
|||
|
|
|
|||
|
|
status = "success" if resp.status_code < 400 else f"http_{resp.status_code}"
|
|||
|
|
await self._emit_audit(event, "out", json_data or params, data, status, latency_ms)
|
|||
|
|
if resp.status_code >= 400:
|
|||
|
|
raise BaseClientError(
|
|||
|
|
message=f"{self.system_name} 返回 HTTP {resp.status_code}",
|
|||
|
|
code=resp.status_code,
|
|||
|
|
detail=resp.text[:500],
|
|||
|
|
)
|
|||
|
|
return data
|
|||
|
|
except (httpx.TimeoutException, httpx.ConnectError) as e:
|
|||
|
|
latency_ms = int((time.monotonic() - start) * 1000)
|
|||
|
|
await self._emit_audit(event, "out", json_data or params, None, "error", latency_ms, str(e))
|
|||
|
|
raise BaseClientError(message=f"{self.system_name} 网络异常: {e}", code=-1, detail=str(e))
|
|||
|
|
except BaseClientError:
|
|||
|
|
raise
|
|||
|
|
except Exception as e:
|
|||
|
|
latency_ms = int((time.monotonic() - start) * 1000)
|
|||
|
|
await self._emit_audit(event, "out", json_data or params, None, "error", latency_ms, str(e))
|
|||
|
|
raise BaseClientError(message=f"{self.system_name} 请求异常: {e}", code=-1, detail=str(e))
|