"""将冻结的业务结果直接 POST 到结果平台,不经过取数网关。""" import math import httpx from extensions.result_platform.settings import ResultPlatformSettings from runtime.result_models import PlatformResultPayload class ResultPlatformError(RuntimeError): def __init__(self, operation, status=0, *, retryable=False, retry_after=None): self.status, self.retryable, self.retry_after = status, retryable, retry_after super().__init__(f"{operation}失败(状态 {status})") class ResultPlatformClient: def __init__(self, settings: ResultPlatformSettings, *, transport=None): self.settings = settings self.http = httpx.AsyncClient(timeout=settings.timeout, trust_env=False, follow_redirects=False, transport=transport) async def close(self): await self.http.aclose() async def push(self, payload): body = PlatformResultPayload.model_validate(payload).model_dump(mode="json") headers = {"Idempotency-Key": body["taskId"]} if self.settings.auth_header: headers[self.settings.auth_header] = " ".join( s for s in (self.settings.auth_scheme, self.settings.token) if s) try: response = await self.http.post(self.settings.url, json=body, headers=headers) except httpx.RequestError: raise ResultPlatformError("结果平台连接", retryable=True) from None if not 200 <= response.status_code < 300: try: retry_after = float(response.headers["Retry-After"]) if not math.isfinite(retry_after) or retry_after < 0: retry_after = None except (KeyError, ValueError): retry_after = None raise ResultPlatformError("结果平台推送", response.status_code, retryable=response.status_code in {408, 425, 429, 500, 502, 503, 504}, retry_after=retry_after) if response.status_code == 204: return {"httpStatus": 204, "taskId": body["taskId"]} try: receipt = response.json() except ValueError: raise ResultPlatformError("结果平台成功回执不是 JSON", response.status_code) from None if not isinstance(receipt, dict) or receipt.get("success") is False: raise ResultPlatformError("结果平台业务回执", response.status_code) value = receipt for field in self.settings.success_field.split("."): value = value.get(field) if isinstance(value, dict) else None if str(value) not in self.settings.success_values: raise ResultPlatformError("结果平台业务回执", response.status_code) # 对方明确回传任务身份时,不能把另一任务的回执算作成功。 returned_id = receipt.get("taskId") if isinstance(receipt.get("data"), dict): returned_id = returned_id or receipt["data"].get("taskId") if returned_id is not None and returned_id != body["taskId"]: raise ResultPlatformError("结果平台回执任务身份不一致", response.status_code) return receipt