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

"""OpenAI 图片接口的请求/响应 DTO。

只放**厂商词汇**。领域概念在 `app/domain/generation.py`，翻译在 `image.py`。

一处值得注意的厂商差异：同一份请求体在**文生图走 JSON、图生图走 multipart**
（后者的 `image` 是文件）。故本类提供两种输出形态——`to_json_body()` 与
`to_form_fields()`。multipart 的字段值必须是字符串，布尔与数字要转，不能直接塞。
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any


@dataclass(frozen=True)
class ImageSubmission:
    """一次图片生成请求。

    刻意只暴露 Patbond 会用到的参数。厂商参数远多于此（`n`、`partial_images`、
    `stream`、`mask`、`input_fidelity`、`output_compression`、`moderation`…），
    多暴露一个就多一处要在契约、客户端、文档三处同步维护的地方。

    :param n: **固定为 1，不开放**。领域模型的一个任务对应一个产物
        （`generation_jobs.output_asset_id` 是单值）；允许 n>1 就得改数据模型。
    :param quality: `None` 交给服务商默认（auto）。注意 `xhigh`/`max` 仅
        gpt-image-2.5 系列支持，gpt-image-2 传了会被拒（`capabilities` 会先拦）。
    :param background: `transparent` 时 `output_format` 只能 png/webp。
        文档称 gpt-image-2 对透明背景的支持**处于预览**，故默认不指定。
    :param safety_identifier: 对应厂商的 `user` 字段——终端用户标识，供其做滥用
        检测。必须是哈希，不可传明文用户 ID。
    """

    model: str
    prompt: str
    size: str | None = None
    quality: str | None = None
    output_format: str | None = None
    background: str | None = None
    safety_identifier: str | None = None

    def to_json_body(self) -> dict[str, Any]:
        """文生图（`/images/generations`）的 JSON 体。"""
        body: dict[str, Any] = {
            "model": self.model,
            "prompt": self.prompt,
            # 一个任务一个产物，见类 docstring。
            "n": 1,
        }
        self._add_optional(body)
        return body

    def to_form_fields(self) -> dict[str, str]:
        """图生图（`/images/edits`）的 multipart 文本字段。

        multipart 的字段值必须是字符串——数字与布尔直接塞进去会被 httpx 拒绝或
        编码成意外形态。
        """
        fields: dict[str, Any] = {
            "model": self.model,
            "prompt": self.prompt,
            "n": 1,
        }
        self._add_optional(fields)
        return {key: str(value) for key, value in fields.items()}

    def _add_optional(self, target: dict[str, Any]) -> None:
        """只加已设置的可选项。

        不设的字段**留空而不显式传 null**：服务商对每个参数都有文档化的默认值，
        显式 null 可能覆盖掉那些默认值。
        """
        if self.size:
            target["size"] = self.size
        if self.quality:
            target["quality"] = self.quality
        if self.output_format:
            target["output_format"] = self.output_format
        if self.background:
            target["background"] = self.background
        if self.safety_identifier:
            # 厂商字段名是 `user`，不是 safety_identifier。
            target["user"] = self.safety_identifier


@dataclass(frozen=True)
class GeneratedImage:
    """`data[]` 里的一张图。

    :param b64_json: base64 编码的图片。**GPT image 系列始终返回它**
        （`response_format` 对这些模型不支持，`url` 只用于 dall-e）。
    """

    b64_json: str | None = None
    url: str | None = None
    revised_prompt: str | None = None


@dataclass(frozen=True)
class ImagesResponse:
    """`/images/generations` 与 `/images/edits` 的共同响应形状。

    :param size: `WIDTHxHEIGHT` 字符串。**像素宽高由此直接得到**，不像火山那边
        只给 ratio+resolution、需要一张随模型族变化的对照表。
    :param total_tokens: 计费对账依据（`usage.total_tokens`）。
    """

    created: int | None = None
    data: tuple[GeneratedImage, ...] = ()
    size: str | None = None
    output_format: str | None = None
    quality: str | None = None
    background: str | None = None
    total_tokens: int | None = None
    input_tokens: int | None = None
    output_tokens: int | None = None
    _raw: dict[str, Any] | None = None

    def first_image(self) -> GeneratedImage | None:
        return self.data[0] if self.data else None

    def raw_keys(self) -> list[str]:
        """原始响应的顶层键，仅用于「2xx 但没有图」时的日志。"""
        return sorted(self._raw) if self._raw else []

    @classmethod
    def from_payload(cls, payload: dict[str, Any]) -> ImagesResponse:
        usage = payload.get("usage") or {}
        images = tuple(
            GeneratedImage(
                b64_json=item.get("b64_json"),
                url=item.get("url"),
                revised_prompt=item.get("revised_prompt"),
            )
            for item in (payload.get("data") or [])
            if isinstance(item, dict)
        )
        return cls(
            created=payload.get("created"),
            data=images,
            size=payload.get("size"),
            output_format=payload.get("output_format"),
            quality=payload.get("quality"),
            background=payload.get("background"),
            total_tokens=usage.get("total_tokens"),
            input_tokens=usage.get("input_tokens"),
            output_tokens=usage.get("output_tokens"),
            _raw=payload,
        )