vent_agent_2.0/extensions/result_platform/client.py

"""将冻结的业务结果直接 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