patbond-ai/app/providers/openai/client.py

"""OpenAI 图片接口的 HTTP 传输层。

与火山那边的 `ArkClient` 职责相同（鉴权、超时、错误解包），但**刻意不共用**：
两家的错误体形状不同（OpenAI 是 `{"error": {message, type, code, param}}`，火山是
`{"error": {code, message}}`），且 OpenAI 的图片编辑走 multipart 而非 JSON。
强行抽一个共同基类会让两边都被对方的特例污染——供应商适配层之间**不该**互相
复用，各自复用自家的传输层才是分层的本意。

本模块不认识「图片生成」这个业务：以后接 OpenAI 的对话或语音接口时复用它。
"""

from __future__ import annotations

import logging
from typing import Any, Self

import httpx

from app.domain.generation import Failure

from .errors import classify

log = logging.getLogger(__name__)

#: 官方 API base URL。
DEFAULT_BASE_URL = "https://api.openai.com/v1"


class OpenAiError(RuntimeError):
    """OpenAI 返回的错误，携带领域分类。"""

    def __init__(self, failure: Failure, http_status: int | None = None):
        super().__init__(failure.message or failure.provider_code or "OpenAI 调用失败")
        self.failure = failure
        self.http_status = http_status


class OpenAiClient:
    """OpenAI API 的同步 HTTP 客户端。

    **超时语义与火山那边不同，这里必须说清**：OpenAI 的图片接口是**同步**的——
    一次请求里等到图片生成完再返回。因此这里的 `timeout_sec` **确实是生成耗时的
    上限**，不是「短往返」的上限，默认值也因此比火山那边大一个量级。

    这也是 `image.py` 必须在服务内部把它转成异步的原因（ADR-023 §4：同步型供应商
    由其适配层在内部消化差异，绝不让它渗透到对外接口）。
    """

    def __init__(
        self,
        api_key: str,
        base_url: str = DEFAULT_BASE_URL,
        timeout_sec: float = 180.0,
        organization: str | None = None,
        client: httpx.Client | None = None,
    ):
        if not api_key:
            raise ValueError(
                "OPENAI_API_KEY 未配置：请通过环境变量注入，切勿写入代码或样例文件"
            )
        self._api_key = api_key
        self._base_url = base_url.rstrip("/")
        self._organization = organization
        self._owns_client = client is None
        self._client = client or httpx.Client(timeout=timeout_sec)

    def post_json(self, path: str, body: dict[str, Any]) -> dict[str, Any]:
        return self._send("POST", path, json=body)

    def post_multipart(
        self,
        path: str,
        data: dict[str, Any],
        files: list[tuple[str, tuple[str, bytes, str]]],
    ) -> dict[str, Any]:
        """multipart 提交，图像编辑接口需要它（`image` 是文件而非 URL）。

        :param files: `(field_name, (filename, content, content_type))`。用 list
            而非 dict 是因为**同名字段可以重复**——编辑接口支持传多张图，它们
            共用字段名 `image[]`，dict 表达不了。
        """
        return self._send("POST", path, data=data, files=files)

    # ---------------------------------------------------------------- 内部

    def _send(self, method: str, path: str, **kwargs: Any) -> dict[str, Any]:
        url = f"{self._base_url}{path}"
        try:
            response = self._client.request(
                method, url, headers=self._headers(), **kwargs
            )
        except httpx.HTTPError as exc:
            # 网络层失败按可重试处理。注意**同步接口的超时最危险**：请求可能已经
            # 在服务商侧完成并计费，而我们拿不到结果。OpenAI 的图片接口没有任务
            # ID 可供事后查询，故这个窗口无法通过"再查一次"弥补——只能由
            # max_attempts 兜住。详见 image.py 的说明。
            raise OpenAiError(
                classify(None, message=f"请求 OpenAI 失败：{exc}")
            ) from exc
        return self._unwrap(response)

    def _headers(self) -> dict[str, str]:
        headers = {"Authorization": f"Bearer {self._api_key}"}
        if self._organization:
            headers["OpenAI-Organization"] = self._organization
        # 刻意不设 Content-Type：JSON 与 multipart 的取值不同，交给 httpx 按
        # 实际载荷推断（multipart 还需要 boundary，手写必错）。
        return headers

    @staticmethod
    def _unwrap(response: httpx.Response) -> dict[str, Any]:
        """把响应转成 dict；非 2xx 转成带分类的 `OpenAiError`。

        错误体形状取自官方 Error codes 页：`{"error": {message, type, code, param}}`。
        四个字段**都要取**——`param` 是区分「参数错」与「内容被拒」的唯一结构化
        线索（理由见 `errors.py` 的 `_classify_bad_request`）。解析不出这个形状时
        不静默，按状态码分类并把原始文本带上。
        """
        if response.is_success:
            try:
                return response.json()
            except ValueError as exc:
                raise OpenAiError(
                    classify(
                        response.status_code,
                        message=f"响应不是合法 JSON：{response.text[:200]}",
                    ),
                    response.status_code,
                ) from exc

        error_type: str | None = None
        error_code: str | None = None
        error_param: str | None = None
        message = response.text[:500]
        try:
            error = response.json().get("error") or {}
            error_type = error.get("type")
            error_code = error.get("code")
            error_param = error.get("param")
            message = error.get("message") or message
        except ValueError:
            pass

        failure = classify(
            response.status_code, error_type, error_code, message, error_param
        )
        log.warning(
            "OpenAI 返回错误 http=%s type=%s code=%s param=%s retry=%s",
            response.status_code,
            error_type,
            error_code,
            error_param,
            failure.retry.value,
        )
        raise OpenAiError(failure, response.status_code)

    def close(self) -> None:
        # 只关自己创建的连接池：注入进来的归调用方管理。
        if self._owns_client:
            self._client.close()

    def __enter__(self) -> Self:
        return self

    def __exit__(self, *exc_info: object) -> None:
        self.close()