382 lines
16 KiB
Python
382 lines
16 KiB
Python
# ============================================================
|
||
# HTTP MES 适配器框架（moduleId: integ-mes-http, 可复用 ✔）
|
||
# round-45 方向 II（矩阵 73/75）：config-driven HTTP MES 适配器
|
||
# - 鉴权：MES_HTTP_TOKEN + MES_HTTP_TOKEN_HEADER（默认 Authorization: Bearer）
|
||
# - 幂等：idemKey 头（X-Idem-Key，可配 MES_HTTP_IDEM_HEADER）或 body.idemKey
|
||
# - 超时：MES_HTTP_TIMEOUT_SECONDS（默认 10s）
|
||
# - 重试：连接失败/超时/5xx/429 指数退避（MES_HTTP_MAX_RETRIES，默认 2）
|
||
# - 状态/报工回流：GET /work-orders/{id} 轮询 + POST /work-orders/{id}/reports 回执
|
||
# - 失败语义：MesHttpError{code,status_code} 显式抛错（供 round-39 saga 补偿识别）；
|
||
# 未配置 MES_HTTP_BASE_URL → fail-closed（MES_HTTP_NOT_CONFIGURED）
|
||
# 与 MockMesClient 同接口（status/create_work_order/post_report/cancel_work_order），
|
||
# server/aps_domain/mes.py 按配置切换客户端（stub 默认，HTTP 显式启用）。
|
||
# ============================================================
|
||
# round-46 方向 LL（矩阵 73 现场对接准备）：readiness() 配置就绪检查
|
||
# - 稳定 JSON 字段 {configured, baseUrl, tokenPresent, timeoutSec, maxRetries,
|
||
# connectivity, lastError, message}；未配置 fail-closed 明确提示（网关 200 不 500）
|
||
# - readiness(probe=True) 轻量 GET /health（短超时 ≤3s），失败记录 lastError 不抛
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import time
|
||
from typing import Any
|
||
from urllib.parse import quote
|
||
|
||
import httpx
|
||
|
||
ENV_BASE_URL = "MES_HTTP_BASE_URL"
|
||
ENV_TOKEN = "MES_HTTP_TOKEN"
|
||
ENV_TOKEN_HEADER = "MES_HTTP_TOKEN_HEADER"
|
||
ENV_TIMEOUT = "MES_HTTP_TIMEOUT_SECONDS"
|
||
ENV_MAX_RETRIES = "MES_HTTP_MAX_RETRIES"
|
||
ENV_IDEM_HEADER = "MES_HTTP_IDEM_HEADER"
|
||
|
||
_DEFAULT_TOKEN_HEADER = "Authorization"
|
||
_DEFAULT_IDEM_HEADER = "X-Idem-Key"
|
||
_DEFAULT_TIMEOUT_SEC = 10.0
|
||
_DEFAULT_MAX_RETRIES = 2
|
||
_PROBE_TIMEOUT_SEC = 3.0 # 连通探测短超时（readiness probe）
|
||
_RETRY_BACKOFF_BASE = 0.25 # 指数退避：0.25/0.5/1.0/... 秒
|
||
_RETRYABLE_STATUS = {429, 500, 502, 503, 504}
|
||
|
||
|
||
class MesHttpError(Exception):
|
||
"""显式失败语义：code + message + status_code（供 saga 补偿识别）。"""
|
||
|
||
def __init__(self, code: str, message: str, status_code: int | None = None) -> None:
|
||
super().__init__(message)
|
||
self.code = code
|
||
self.message = message
|
||
self.status_code = status_code
|
||
|
||
def __str__(self) -> str:
|
||
base = f"[{self.code}] {self.message}"
|
||
if self.status_code is not None:
|
||
base += f" (HTTP {self.status_code})"
|
||
return base
|
||
|
||
|
||
class HttpMesClient:
|
||
"""真实 MES HTTP 适配器：鉴权/幂等/超时/重试/回执（config-driven）。"""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
base_url: str | None = None,
|
||
token: str | None = None,
|
||
token_header: str | None = None,
|
||
timeout_sec: float | None = None,
|
||
max_retries: int | None = None,
|
||
idem_header: str | None = None,
|
||
transport: httpx.BaseTransport | None = None,
|
||
) -> None:
|
||
self.base_url = (
|
||
base_url if base_url is not None else os.environ.get(ENV_BASE_URL) or ""
|
||
).rstrip("/")
|
||
self.token = token if token is not None else os.environ.get(ENV_TOKEN) or ""
|
||
self.token_header = (
|
||
token_header or os.environ.get(ENV_TOKEN_HEADER) or _DEFAULT_TOKEN_HEADER
|
||
).strip()
|
||
try:
|
||
self.timeout_sec = float(
|
||
timeout_sec if timeout_sec is not None
|
||
else os.environ.get(ENV_TIMEOUT) or _DEFAULT_TIMEOUT_SEC
|
||
)
|
||
except ValueError as exc:
|
||
raise MesHttpError("MES_HTTP_CONFIG_INVALID", f"{ENV_TIMEOUT} 配置无效", 503) from exc
|
||
if self.timeout_sec <= 0:
|
||
raise MesHttpError("MES_HTTP_CONFIG_INVALID", f"{ENV_TIMEOUT} 必须大于 0", 503)
|
||
raw_retries = max_retries if max_retries is not None else os.environ.get(ENV_MAX_RETRIES)
|
||
try:
|
||
self.max_retries = max(0, int(raw_retries or _DEFAULT_MAX_RETRIES))
|
||
except ValueError as exc:
|
||
raise MesHttpError("MES_HTTP_CONFIG_INVALID", f"{ENV_MAX_RETRIES} 配置无效", 503) from exc
|
||
self.idem_header = (
|
||
idem_header or os.environ.get(ENV_IDEM_HEADER) or _DEFAULT_IDEM_HEADER
|
||
).strip()
|
||
self.transport = transport
|
||
|
||
# ---------------- 配置 ----------------
|
||
@property
|
||
def configured(self) -> bool:
|
||
return bool(self.base_url)
|
||
|
||
def _ensure_configured(self) -> None:
|
||
if not self.base_url:
|
||
raise MesHttpError(
|
||
"MES_HTTP_NOT_CONFIGURED",
|
||
f"未配置 {ENV_BASE_URL}：HTTP MES 适配器不可用（fail-closed）。"
|
||
"请设置环境变量或传入 base_url 指向真实 MES / 本地 stub。",
|
||
503,
|
||
)
|
||
|
||
def _auth_headers(self) -> dict[str, str]:
|
||
headers: dict[str, str] = {"Accept": "application/json"}
|
||
if self.token:
|
||
if self.token_header.lower() == "authorization" and not self.token.startswith("Bearer "):
|
||
headers[self.token_header] = "Bearer " + self.token
|
||
else:
|
||
headers[self.token_header] = self.token
|
||
return headers
|
||
|
||
# ---------------- HTTP 原语（超时 + 指数退避重试） ----------------
|
||
def _request(
|
||
self,
|
||
method: str,
|
||
path: str,
|
||
*,
|
||
json_body: dict[str, Any] | None = None,
|
||
params: dict[str, str] | None = None,
|
||
idem_key: str | None = None,
|
||
allow_retry: bool = True,
|
||
) -> httpx.Response:
|
||
self._ensure_configured()
|
||
headers = self._auth_headers()
|
||
if idem_key:
|
||
headers[self.idem_header] = idem_key
|
||
last_error: MesHttpError | None = None
|
||
attempt = 0
|
||
while True:
|
||
try:
|
||
with httpx.Client(
|
||
base_url=self.base_url,
|
||
timeout=self.timeout_sec,
|
||
transport=self.transport,
|
||
) as client:
|
||
resp = client.request(method, path, json=json_body, params=params, headers=headers)
|
||
except httpx.TimeoutException:
|
||
last_error = MesHttpError(
|
||
"MES_HTTP_TIMEOUT",
|
||
f"MES 请求超时（>{self.timeout_sec}s）：{method} {path}",
|
||
504,
|
||
)
|
||
retryable = True
|
||
except httpx.RequestError as exc:
|
||
last_error = MesHttpError(
|
||
"MES_HTTP_CONNECT_FAILED",
|
||
f"无法连接 MES（{self.base_url}）：{exc}",
|
||
502,
|
||
)
|
||
retryable = True
|
||
else:
|
||
if resp.status_code < 400:
|
||
return resp
|
||
last_error = self._error_from_response(resp)
|
||
retryable = resp.status_code in _RETRYABLE_STATUS
|
||
if not allow_retry or not retryable or attempt >= self.max_retries:
|
||
raise last_error
|
||
attempt += 1
|
||
time.sleep(_RETRY_BACKOFF_BASE * (2 ** (attempt - 1)))
|
||
|
||
def _error_from_response(self, resp: httpx.Response) -> MesHttpError:
|
||
try:
|
||
body = resp.json()
|
||
except ValueError:
|
||
body = {}
|
||
message = ""
|
||
if isinstance(body, dict):
|
||
detail = body.get("detail")
|
||
if isinstance(detail, dict):
|
||
message = str(detail.get("message") or detail.get("error") or "")
|
||
else:
|
||
message = str(body.get("message") or body.get("error") or detail or "")
|
||
fallback = message or f"MES HTTP {resp.status_code}"
|
||
if resp.status_code in (401, 403):
|
||
return MesHttpError("MES_HTTP_AUTH_FAILED", f"MES 鉴权失败：{fallback}", resp.status_code)
|
||
if resp.status_code == 404:
|
||
return MesHttpError("MES_HTTP_NOT_FOUND", f"MES 资源不存在：{fallback}", resp.status_code)
|
||
if resp.status_code >= 500 or resp.status_code == 429:
|
||
return MesHttpError("MES_HTTP_UPSTREAM_ERROR", f"MES 上游错误：{fallback}", resp.status_code)
|
||
return MesHttpError("MES_HTTP_REJECTED", f"MES 拒绝请求：{fallback}", resp.status_code)
|
||
|
||
@staticmethod
|
||
def _json(resp: httpx.Response) -> dict[str, Any]:
|
||
try:
|
||
data = resp.json()
|
||
except ValueError as exc:
|
||
raise MesHttpError("MES_HTTP_INVALID_RESPONSE", "MES 返回非 JSON 响应", resp.status_code) from exc
|
||
if not isinstance(data, dict):
|
||
raise MesHttpError("MES_HTTP_INVALID_RESPONSE", "MES 返回非对象响应", resp.status_code)
|
||
return data
|
||
|
||
@staticmethod
|
||
def _wo_path(wo_id: str) -> str:
|
||
return "/work-orders/" + quote(str(wo_id), safe="")
|
||
|
||
# ---------------- 领域方法（与 MockMesClient 同接口） ----------------
|
||
def status(self) -> dict[str, Any]:
|
||
"""连接状态（供预览/清单展示）：失败不抛错，返回 connected=False。"""
|
||
if not self.base_url:
|
||
return {
|
||
"connected": False, "mode": "http", "system": "MES-HTTP",
|
||
"detail": f"未配置 {ENV_BASE_URL}",
|
||
}
|
||
try:
|
||
resp = self._request("GET", "/health", allow_retry=False)
|
||
body = self._json(resp)
|
||
except MesHttpError as exc:
|
||
return {
|
||
"connected": False, "mode": "http", "system": "MES-HTTP",
|
||
"error": exc.code, "detail": exc.message,
|
||
}
|
||
return {
|
||
"connected": True, "mode": "http",
|
||
"system": body.get("system") or "MES-HTTP",
|
||
"plant": body.get("plant") or "CNWH",
|
||
"updatedAt": body.get("updatedAt") or "",
|
||
"woCount": int(body.get("woCount") or 0),
|
||
"openCount": int(body.get("openCount") or 0),
|
||
"reportCount": int(body.get("reportCount") or 0),
|
||
}
|
||
|
||
# ---------------- 对接就绪检查（round-46 方向 LL · 矩阵 73 现场对接准备） ----------------
|
||
def readiness(self, *, probe: bool = False) -> dict[str, Any]:
|
||
"""现场对接就绪检查：稳定 JSON 字段，不抛错（网关可直接转发给前端）。
|
||
|
||
- 未配置 → configured=False + 中文提示（fail-closed；网关 200 不 500）。
|
||
- 已配置未探测 → connectivity="unknown"。
|
||
- probe=True → 对 base_url 发轻量 GET /health（短超时 ≤3s），
|
||
成功 connectivity="ok"，失败 connectivity="failed" 且 lastError 记录错误（不抛）。
|
||
"""
|
||
result: dict[str, Any] = {
|
||
"configured": self.configured,
|
||
"baseUrl": self.base_url or None,
|
||
"tokenPresent": bool(self.token),
|
||
"timeoutSec": self.timeout_sec,
|
||
"maxRetries": self.max_retries,
|
||
"connectivity": None, # "unknown" | "ok" | "failed" | null
|
||
"lastError": None,
|
||
"message": "",
|
||
}
|
||
if not self.configured:
|
||
result["message"] = (
|
||
f"未配置 {ENV_BASE_URL}：HTTP MES 适配器不可用（fail-closed）。"
|
||
"请设置环境变量或传入 base_url 指向真实 MES / 本地 stub。"
|
||
)
|
||
return result
|
||
if not probe:
|
||
result["connectivity"] = "unknown"
|
||
result["message"] = (
|
||
f"已配置 {ENV_BASE_URL}，尚未探测连通性；"
|
||
"可调用 POST /api/integrations/mes/readiness/probe 触发轻量探测。"
|
||
)
|
||
return result
|
||
ok, error = self._probe_health()
|
||
if ok:
|
||
result["connectivity"] = "ok"
|
||
result["message"] = "连通正常"
|
||
else:
|
||
result["connectivity"] = "failed"
|
||
result["lastError"] = {
|
||
"code": error.code,
|
||
"message": error.message,
|
||
"statusCode": error.status_code,
|
||
}
|
||
result["message"] = f"连通失败：{error.message}"
|
||
return result
|
||
|
||
def _probe_health(self) -> tuple[bool, MesHttpError | None]:
|
||
"""轻量连通探测：GET /health，短超时（≤3s），失败返回 (False, error) 不抛出。"""
|
||
timeout = min(self.timeout_sec, _PROBE_TIMEOUT_SEC)
|
||
try:
|
||
with httpx.Client(
|
||
base_url=self.base_url,
|
||
timeout=timeout,
|
||
transport=self.transport,
|
||
) as client:
|
||
resp = client.get("/health", headers=self._auth_headers())
|
||
except httpx.ConnectTimeout:
|
||
# 连接阶段超时（拒连/防火墙丢包等）→ 判为连接失败（502），而非上游超时
|
||
return False, MesHttpError(
|
||
"MES_HTTP_CONNECT_FAILED",
|
||
f"无法连接 MES（{self.base_url}）：连接超时（>{timeout}s）",
|
||
502,
|
||
)
|
||
except httpx.TimeoutException:
|
||
return False, MesHttpError(
|
||
"MES_HTTP_TIMEOUT",
|
||
f"MES 连通探测超时（>{timeout}s）",
|
||
504,
|
||
)
|
||
except httpx.RequestError as exc:
|
||
return False, MesHttpError(
|
||
"MES_HTTP_CONNECT_FAILED",
|
||
f"无法连接 MES（{self.base_url}）：{exc}",
|
||
502,
|
||
)
|
||
if resp.status_code >= 400:
|
||
return False, self._error_from_response(resp)
|
||
return True, None
|
||
|
||
def create_work_order(self, payload: dict[str, Any], idem_key: str) -> dict[str, Any]:
|
||
"""幂等下发：idemKey 头/body 携带；重复 idemKey → duplicate=True（不重复创建）。"""
|
||
if not idem_key:
|
||
raise MesHttpError("MES_HTTP_REJECTED", "下发缺少 idemKey（幂等键）", 400)
|
||
body = dict(payload)
|
||
body.setdefault("idemKey", idem_key)
|
||
resp = self._request("POST", "/work-orders", json_body=body, idem_key=idem_key)
|
||
data = self._json(resp)
|
||
if "externalWo" not in data:
|
||
if "id" in data:
|
||
data = {"duplicate": False, "externalWo": data}
|
||
else:
|
||
raise MesHttpError(
|
||
"MES_HTTP_INVALID_RESPONSE",
|
||
"MES 下发响应缺少 externalWo", resp.status_code,
|
||
)
|
||
return data
|
||
|
||
def post_report(self, external_wo_id: str, payload: dict[str, Any]) -> dict[str, Any]:
|
||
"""报工回执：POST 进度/数量/状态回流到真实 MES。"""
|
||
resp = self._request(
|
||
"POST", self._wo_path(external_wo_id) + "/reports",
|
||
json_body=payload,
|
||
)
|
||
data = self._json(resp)
|
||
if "report" not in data:
|
||
raise MesHttpError("MES_HTTP_INVALID_RESPONSE", "MES 报工回执缺少 report", resp.status_code)
|
||
return data
|
||
|
||
def fetch_work_order(self, external_wo_id: str) -> dict[str, Any]:
|
||
"""状态查询：GET 轮询外部工单。"""
|
||
resp = self._request("GET", self._wo_path(external_wo_id))
|
||
data = self._json(resp)
|
||
if "id" not in data:
|
||
raise MesHttpError("MES_HTTP_INVALID_RESPONSE", "MES 状态查询缺少 id", resp.status_code)
|
||
return data
|
||
|
||
def cancel_work_order(
|
||
self,
|
||
external_wo_id: str,
|
||
*,
|
||
reason: str = "saga-compensation",
|
||
) -> dict[str, Any]:
|
||
"""幂等撤单（saga 补偿）：重复撤单 → duplicate=True。"""
|
||
resp = self._request(
|
||
"POST", self._wo_path(external_wo_id) + "/cancel",
|
||
json_body={"reason": reason},
|
||
idem_key="cancel:" + str(external_wo_id),
|
||
)
|
||
data = self._json(resp)
|
||
if "externalWo" not in data:
|
||
raise MesHttpError("MES_HTTP_INVALID_RESPONSE", "MES 撤单响应缺少 externalWo", resp.status_code)
|
||
return data
|
||
|
||
|
||
_client: HttpMesClient | None = None
|
||
|
||
|
||
def get_http_mes_client() -> HttpMesClient:
|
||
"""进程内单例（按环境变量构建）；测试/热切换用 reset_http_mes_client 重建。"""
|
||
global _client
|
||
if _client is None:
|
||
_client = HttpMesClient()
|
||
return _client
|
||
|
||
|
||
def reset_http_mes_client(**kwargs: Any) -> HttpMesClient:
|
||
"""测试/配置热切换：重建单例（kwargs 透传 HttpMesClient 构造参数）。"""
|
||
global _client
|
||
_client = HttpMesClient(**kwargs)
|
||
return _client |