diff --git a/.env.sample b/.env.sample index 307e06c..51910d6 100644 --- a/.env.sample +++ b/.env.sample @@ -18,6 +18,28 @@ ARK_VIDEO_MODEL= # 默认取下限 1 小时：厂商默认 48 小时对交互式应用过长，被放弃的任务会挂两天。 # ARK_TASK_EXPIRY_SEC=3600 +# ---- OpenAI 图片（gpt-image-2）---- +# 密钥无默认值。缺失时该适配器不装配，服务仍可启动并提供健康检查。 +OPENAI_API_KEY=your-openai-api-key +# OPENAI_BASE_URL=https://api.openai.com/v1 +# 属于多个组织时才需要指定。 +# OPENAI_ORGANIZATION= +# ⚠️ 与 ARK_TIMEOUT_SEC 语义不同：OpenAI 图片接口是**同步**的，一次请求里等到 +# 图片生成完才返回，故这个值**确实是生成耗时的上限**，不是短往返的上限。 +# OPENAI_TIMEOUT_SEC=180 +# 图片模型 ID，仅作兜底。同 ARK_VIDEO_MODEL：正常由 Java 侧传入任务行的快照。 +OPENAI_IMAGE_MODEL= +# 同时执行的同步调用数上限。这同时也是「同时有多少钱在花」的上界——超出的提交 +# 排队，而不是一起打到服务商那里。 +# OPENAI_MAX_CONCURRENT=4 +# 已完成任务在内存注册表里的保留时长（秒）。**必须大于 Java 侧最长的轮询间隔** +# （当前阶梯最长 30 秒），否则结果会在被取走前被清掉——症状是「任务成功了但查 +# 不到」，极难查。 +# OPENAI_COMPLETED_TTL_SEC=900 +# 注册表条目上限。满了以后新提交被拒（503），**不挤掉在途任务**——挤掉一个在途 +# 任务会让一次可能正在计费的调用变成无人认领的开销。 +# OPENAI_MAX_INFLIGHT=256 + # ---- 服务身份 JWT（ADR-023 §5）---- # 与 patbond-creation 持有的私钥配对的 RS256 公钥。注意这是一对**独立于用户 # token 的**密钥：现有 JwtVerifier 只验签名、不校验 aud/iss，复用同一密钥对会 diff --git a/README.md b/README.md index 90f6bf5..4d554f6 100644 --- a/README.md +++ b/README.md @@ -71,7 +71,7 @@ python -m app # 默认 127.0.0.1:8000 ## 代码结构 -分四层，依赖**单向**向内（`config` → `providers` → `domain`），由 +分层，依赖**单向**向内（`api` → `assembly`/`config` → `providers` → `domain`），由 `tests/test_layering.py` 自动守着——靠纪律守不住的东西，靠测试才守得住。 ``` @@ -80,11 +80,21 @@ app/ ├── providers/ │ ├── base.py 生成能力抽象（新增供应商 = 实现它） │ ├── registry.py provider_code → 适配器 -│ └── volcengine/ -│ ├── client.py HTTP 传输与鉴权（不认识「视频」） -│ ├── errors.py 火山错误码 → 领域 Failure +│ ├── volcengine/ +│ │ ├── client.py HTTP 传输与鉴权（不认识「视频」） +│ │ ├── errors.py 火山错误码 → 领域 Failure +│ │ ├── models.py 厂商请求/响应 DTO +│ │ ├── capabilities.py 模型能力与参数限制 +│ │ ├── dimensions.py ratio+resolution → 像素对照表 +│ │ └── video.py Seedance 视频业务 +│ └── openai/ +│ ├── client.py HTTP 传输与鉴权（JSON + multipart） +│ ├── errors.py OpenAI 错误 → 领域 Failure │ ├── models.py 厂商请求/响应 DTO -│ └── video.py Seedance 视频业务 +│ ├── capabilities.py 尺寸算术约束与画质档 +│ ├── tasks.py 同步→异步桥（本服务唯一的有状态组件） +│ └── image.py gpt-image-2 图片业务（文生图 + 图生图） +├── api/ HTTP 端点、服务票校验、错误信封 ├── config.py 集中配置（全部旋钮从环境变量读） └── assembly.py 装配根（唯一同时知道配置与具体供应商的地方） ``` @@ -98,6 +108,10 @@ app/ - **`client.py` 与 `video.py` 分开。** client 只管鉴权、超时、错误解包，不认识 「视频」。以后接火山的**图片或语音**接口时新建兄弟模块复用同一个 client， 而不是再抄一遍鉴权与错误解包——这是拆它出来的全部理由。 +- **两家的 `client.py` 刻意不共用。** 错误体形状不同（OpenAI 是 + `{"error": {message, type, code, param}}`，火山是 `{"error": {code, message}}`）， + 且 OpenAI 的编辑端点走 multipart。抽一个共同基类会让两边都被对方的特例污染 + ——**同一家的不同能力之间才复用传输层，不同供应商之间不复用**。 - **适配层不读配置**，全部通过构造参数注入。否则单测得摆弄环境变量，且同一进程 内没法跑两套不同配置的实例。 @@ -116,12 +130,28 @@ app/ | 服务商 | 能力 | provider_code | 状态 | | --- | --- | --- | --- | | 火山方舟 Seedance | 视频生成 | `volcengine-seedance` | 提交与查询已实现，产物转存待存储决策 | +| OpenAI gpt-image-2 | 图片生成（文生图 / 图生图） | `openai-gpt-image` | 同上；另见下方「同步型供应商的代价」 | 接口事实与错误码映射的完整取证见 patbond-doc -`docs/development/providers/volcengine-seedance.md`。其中两条尤其要记住： +`docs/development/providers/volcengine-seedance.md` 与 +`docs/development/providers/openai-gpt-image.md`。其中几条尤其要记住： - **火山不支持幂等键**，「提交超时后不知道有没有扣费」只能由 `max_attempts` 兜住； -- **产物 URL 只有 24 小时有效期**，因此厂商 URL 不能当存储用，必须在时限内转存。 +- **火山产物 URL 只有 24 小时有效期**，因此厂商 URL 不能当存储用，必须在时限内转存； +- **OpenAI 的两份文档都没有错误码表**，故 `providers/openai/errors.py` 的映射是 + 按 HTTP 状态码为主推断的，**未经文档证实**，拿到真实错误数据后须校正。 + +### 同步型供应商的代价 + +OpenAI 的图片接口是**同步**的（没有任务 ID，也没有查询端点），由 +`providers/openai/tasks.py` 在本服务内部包成异步。两笔代价无法靠代码消除： + +- **本服务从此持有状态。** 进程重启会丢掉所有在途图片任务（Java 侧轮询得到 + `task_not_found`，按失败处理）。也因此 **`patbond-ai` 横向扩容前必须先解决 + 任务 ID 只在签发进程内有效的问题**——多副本 + 无粘性路由会让轮询落到不认识 + 该 ID 的副本上。视频那侧没有这个约束，火山的任务 ID 是全局的。 +- **超时窗口无法靠重查弥补。** 超时那次调用可能已在服务商侧完成并计费，而图片 + 接口没有任务 ID 可供事后查询。只能由 `max_attempts` 兜住。 ## 配置 @@ -134,6 +164,12 @@ app/ | `ARK_TIMEOUT_SEC` | `10` | **单次 HTTP 往返**超时，不是生成耗时 | | `ARK_VIDEO_MODEL` | _（无）_ | 仅作兜底；正常由 Java 侧传入任务行的模型快照 | | `ARK_TASK_EXPIRY_SEC` | `3600` | 厂商 `execution_expires_after`，区间 `[3600, 259200]`。默认取下限——厂商默认 48 小时对交互式应用过长 | +| `OPENAI_API_KEY` | _（无）_ | OpenAI 密钥。**缺失时该适配器不装配**，服务仍可启动 | +| `OPENAI_TIMEOUT_SEC` | `180` | ⚠️ 与 `ARK_TIMEOUT_SEC` **语义不同**：图片接口同步，这里是**生成耗时的上限** | +| `OPENAI_IMAGE_MODEL` | _（无）_ | 仅作兜底；正常由 Java 侧传入任务行的模型快照 | +| `OPENAI_MAX_CONCURRENT` | `4` | 同时执行的同步调用数上限，也是「同时有多少钱在花」的上界 | +| `OPENAI_COMPLETED_TTL_SEC` | `900` | 已完成任务在内存注册表的保留时长。**须大于 Java 侧最长轮询间隔**，否则结果会在被取走前被清掉 | +| `OPENAI_MAX_INFLIGHT` | `256` | 注册表条目上限。满了拒绝新提交（503），**不挤掉在途任务** | | `PATBOND_SERVICE_JWT_PUBLIC_KEY` | _（无）_ | 校验 Java 侧服务票的 RS256 公钥 | | `PATBOND_SERVICE_JWT_AUDIENCE` | `patbond-ai` | **必须强制校验**，否则独立密钥对失去意义 | diff --git a/app/api/routes.py b/app/api/routes.py index ebd69e3..d03d6d7 100644 --- a/app/api/routes.py +++ b/app/api/routes.py @@ -16,6 +16,7 @@ from fastapi import APIRouter, Depends, Request from app.domain.generation import MediaKind from app.providers.base import GenerationProvider +from app.providers.openai.client import OpenAiError from app.providers.registry import ProviderRegistry from app.providers.volcengine.client import ArkError @@ -62,13 +63,18 @@ def submit_task(payload: SubmitRequest, request: Request) -> SubmitResponse: provider = _provider_for(_registry(request), payload.media_kind) try: provider_request_id = provider.submit(payload.to_domain()) - except ArkError as exc: + except (ArkError, OpenAiError) as exc: # 服务商调用失败：原样带上分类，调用方据 retry 决定重试。 raise ProviderError(exc.failure) from exc except ValueError as exc: - # 本地预校验拦下的参数问题（超出该模型的时长/分辨率/ratio 限制，或内容为空）。 - # 属 invalid_request 而非 provider 错误——请求根本没发出去。 + # 本地预校验拦下的参数问题（超出该模型的时长/尺寸/ratio 限制、内容为空、 + # 参考图不可下载）。属 invalid_request 而非 provider 错误——请求根本没发出去。 raise ServiceError(ServiceErrorCode.INVALID_REQUEST, str(exc)) from exc + except RuntimeError as exc: + # 同步型适配器的在途任务已达上限。这不是请求的问题而是容量问题，且稍后 + # 重试会好，故归 provider_unavailable（503）而非 invalid_request——后者会 + # 让调用方判为不可重试并直接失败掉一个本可成功的任务。 + raise ServiceError(ServiceErrorCode.PROVIDER_UNAVAILABLE, str(exc)) from exc log.info( "已提交生成任务 job_id=%s provider_request_id=%s", @@ -96,6 +102,13 @@ def get_task( provider = _provider_for(_registry(request), media_kind) try: result = provider.fetch(provider_request_id) + except KeyError as exc: + # 同步型适配器的本地注册表里没有这个 ID：从未存在过，或进程重启后丢失 + # （两者无法区分）。两种情形下调用方都该**停止轮询**而非退避重试。 + raise ServiceError( + ServiceErrorCode.TASK_NOT_FOUND, + f"本服务无此任务（可能因进程重启而丢失）：{provider_request_id}", + ) from exc except ArkError as exc: # 服务商说任务不存在（含超过 7 天保留期）→ 这是 service 类的 task_not_found， # 而不是 provider 错误：调用方要做的是停止轮询，不是退避重试。 @@ -105,4 +118,6 @@ def get_task( f"服务商无此任务（或已超出其 7 天保留期）：{provider_request_id}", ) from exc raise ProviderError(exc.failure) from exc + except OpenAiError as exc: + raise ProviderError(exc.failure) from exc return TaskResponse.of(result) diff --git a/app/api/schemas.py b/app/api/schemas.py index f6809ce..ef4f965 100644 --- a/app/api/schemas.py +++ b/app/api/schemas.py @@ -37,6 +37,10 @@ class SubmitRequest(BaseModel): input_url: str | None = None ratio: str | None = None resolution: str | None = None + #: 图片路径用像素尺寸；视频路径用 ratio+resolution。两家服务商的尺寸表达方式 + #: 不同，契约同时保留两种而不强行统一——任何统一都会在某一侧失真。 + width_px: int | None = None + height_px: int | None = None duration_sec: int | None = None generate_audio: bool = True watermark: bool = False @@ -63,6 +67,8 @@ class SubmitRequest(BaseModel): input_url=self.input_url, ratio=self.ratio, resolution=self.resolution, + width_px=self.width_px, + height_px=self.height_px, duration_sec=self.duration_sec, generate_audio=self.generate_audio, watermark=self.watermark, diff --git a/app/assembly.py b/app/assembly.py index 82b35fd..cda9851 100644 --- a/app/assembly.py +++ b/app/assembly.py @@ -14,6 +14,9 @@ import logging from app.config import Settings from app.providers.base import GenerationProvider +from app.providers.openai.client import OpenAiClient +from app.providers.openai.image import OpenAiImageProvider +from app.providers.openai.tasks import SyncTaskRegistry from app.providers.registry import ProviderRegistry from app.providers.volcengine.client import ArkClient from app.providers.volcengine.video import SeedanceVideoProvider @@ -48,6 +51,29 @@ def build_registry(settings: Settings) -> ProviderRegistry: # 与它相隔很远。 log.warning("ARK_API_KEY 未配置，火山方舟视频适配器未装配") + openai = settings.openai + if openai.api_key: + providers.append( + OpenAiImageProvider( + OpenAiClient( + api_key=openai.api_key, + base_url=openai.base_url, + timeout_sec=openai.timeout_sec, + organization=openai.organization or None, + ), + # 同步→异步的任务注册表。每个适配器实例一份：它承载的是「本进程 + # 内哪些同步调用正在跑」，共享一份会让两个适配器互相挤占容量。 + SyncTaskRegistry( + max_workers=openai.max_concurrent, + completed_ttl_sec=openai.completed_ttl_sec, + max_entries=openai.max_inflight, + ), + default_model=openai.image_model, + ) + ) + else: + log.warning("OPENAI_API_KEY 未配置，图片生成适配器未装配") + registry = ProviderRegistry(providers) log.info("已装配 %d 个生成适配器：%s", len(registry), sorted(registry.codes)) return registry diff --git a/app/config.py b/app/config.py index d0a88d3..d1bb9f1 100644 --- a/app/config.py +++ b/app/config.py @@ -14,6 +14,7 @@ from __future__ import annotations import os from dataclasses import dataclass, field +from app.providers.openai.client import DEFAULT_BASE_URL as OPENAI_DEFAULT_BASE_URL from app.providers.volcengine.models import DEFAULT_BASE_URL @@ -83,6 +84,59 @@ class VolcengineSettings: ) +@dataclass(frozen=True) +class OpenAiSettings: + """OpenAI 图片接入配置。 + + :param api_key: `OPENAI_API_KEY`。**无默认值**，缺失时该适配器不装配。 + :param timeout_sec: **这里确实是生成耗时的上限**，与火山那边不同——OpenAI 的 + 图片接口是同步的，一次请求里等到图片生成完才返回。故默认值比火山的 + 「短往返超时」大一个量级。 + :param image_model: 图片模型 ID，仅作兜底；正常由 Java 侧传入任务行的快照。 + :param max_concurrent: 同时执行的同步调用数上限。**同时也是对服务商并发和 + 「同时有多少钱在花」的自我限制**——超出的提交排队而非一起打过去。 + :param completed_ttl_sec: 已完成任务在内存注册表里的保留时长。须大于 Java 侧 + 最长的轮询间隔（当前阶梯最长 30 秒），否则结果会在被取走前被清理。 + """ + + api_key: str = "" + base_url: str = OPENAI_DEFAULT_BASE_URL + timeout_sec: float = 180.0 + image_model: str = "" + organization: str = "" + max_concurrent: int = 4 + completed_ttl_sec: float = 900.0 + max_inflight: int = 256 + + @classmethod + def from_env(cls) -> OpenAiSettings: + return cls( + api_key=_env("OPENAI_API_KEY"), + base_url=_env("OPENAI_BASE_URL", cls.base_url), + timeout_sec=_env_float("OPENAI_TIMEOUT_SEC", cls.timeout_sec), + image_model=_env("OPENAI_IMAGE_MODEL"), + organization=_env("OPENAI_ORGANIZATION"), + max_concurrent=_env_int("OPENAI_MAX_CONCURRENT", cls.max_concurrent), + completed_ttl_sec=_env_float( + "OPENAI_COMPLETED_TTL_SEC", cls.completed_ttl_sec + ), + max_inflight=_env_int("OPENAI_MAX_INFLIGHT", cls.max_inflight), + ) + + def __post_init__(self) -> None: + if self.max_concurrent < 1: + raise ValueError( + f"OPENAI_MAX_CONCURRENT 必须 >= 1，当前 {self.max_concurrent}" + ) + # 小于最长轮询间隔会让结果在被取走前就被清掉——那表现为「任务成功了但查 + # 不到」，极难查。 + if self.completed_ttl_sec < 60: + raise ValueError( + f"OPENAI_COMPLETED_TTL_SEC 不应小于 60 秒（须大于 Java 侧最长轮询" + f"间隔，否则结果会在被取走前被清理），当前 {self.completed_ttl_sec}" + ) + + @dataclass(frozen=True) class ServiceAuthSettings: """校验 Java 侧服务身份 JWT 所需的配置（ADR-023 §5）。 @@ -113,6 +167,7 @@ class Settings: """本服务的全部配置。""" volcengine: VolcengineSettings = field(default_factory=VolcengineSettings) + openai: OpenAiSettings = field(default_factory=OpenAiSettings) service_auth: ServiceAuthSettings = field(default_factory=ServiceAuthSettings) log_level: str = "INFO" @@ -120,6 +175,7 @@ class Settings: def from_env(cls) -> Settings: return cls( volcengine=VolcengineSettings.from_env(), + openai=OpenAiSettings.from_env(), service_auth=ServiceAuthSettings.from_env(), log_level=_env("PATBOND_AI_LOG_LEVEL", cls.log_level).upper(), ) diff --git a/app/domain/generation.py b/app/domain/generation.py index 0e0a4d7..a01b510 100644 --- a/app/domain/generation.py +++ b/app/domain/generation.py @@ -109,6 +109,12 @@ class GenerationRequest: input_url: str | None = None ratio: str | None = None resolution: str | None = None + #: 像素尺寸。**图片路径用它，视频路径用 ratio+resolution**——两家服务商的 + #: 尺寸表达方式不同：火山用「分辨率档 + 宽高比」，OpenAI 用 WIDTHxHEIGHT。 + #: 领域层同时保留两种表达而不强行统一，是因为任何统一都会在某一侧失真 + #: （把 720p+16:9 折成像素会丢掉"档位"语义，把像素折成比例会丢掉精确值）。 + width_px: int | None = None + height_px: int | None = None duration_sec: int | None = None generate_audio: bool = True watermark: bool = False diff --git a/app/providers/openai/__init__.py b/app/providers/openai/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/providers/openai/capabilities.py b/app/providers/openai/capabilities.py new file mode 100644 index 0000000..3646955 --- /dev/null +++ b/app/providers/openai/capabilities.py @@ -0,0 +1,151 @@ +"""gpt-image-2 的尺寸与参数约束。 + +与火山那边不同，OpenAI 的图片接口不用「分辨率 + 宽高比」两个字段，而是一个 +`WIDTHxHEIGHT` 字符串，并对它施加几条**算术约束**。逐条取自官方文档： + +- 宽高**都必须能被 16 整除** +- 宽高比必须在 `1:3` 到 `3:1` 之间 +- 最大支持 `3840x2160` +- 超过 `2560x1440` 属**实验性** +- 另有「模型当前的像素与边长限制」——文档未给具体数值，故无法在本地校验 + +在本地先校验这些，是因为违反约束的请求会被 OpenAI 以 400 拒绝，而那要花一次 +往返、计入配额，并表现为一个失败的任务。错误信息还能直接指出违反了哪一条。 + +**标准尺寸**（`1024x1024` / `1536x1024` / `1024x1536`）与 `auto` 一律放行， +不做算术校验：它们是文档明示支持的取值，其中 `auto` 由模型自行决定。 +""" + +from __future__ import annotations + +import logging +import re +from dataclasses import dataclass + +log = logging.getLogger(__name__) + +#: 文档明示的标准尺寸，直接放行。 +STANDARD_SIZES = frozenset({"1024x1024", "1536x1024", "1024x1536"}) + +#: 交给模型自选尺寸。 +AUTO_SIZE = "auto" + +#: `WIDTHxHEIGHT`。 +_SIZE_PATTERN = re.compile(r"^(\d+)x(\d+)$") + +#: 宽高必须同时是它的倍数。 +EDGE_MULTIPLE = 16 + +#: 宽高比的允许区间（宽/高），即 1:3 ~ 3:1。 +MIN_ASPECT = 1 / 3 +MAX_ASPECT = 3.0 + +#: 最大支持尺寸，文档原文是「the maximum supported resolution is 3840x2160」。 +#: +#: ⚠️ **这里有一处文档未明确、由我们推断的语义**：那句话没说清限制是「逐边」 +#: （宽≤3840 且 高≤2160）还是「按总像素」。按逐边理解会**禁掉所有高于 2160 的 +#: 竖图**，而 `1024x1536` 是文档明示的标准竖图尺寸、竖版显然受支持——故逐边理解 +#: 必然过严。 +#: +#: 采用的读法：**最长边 ≤ 3840，且总像素 ≤ 3840×2160**。它允许 `2160x3840` +#: 这类竖图（与横图同像素量），又拦住明显越界的尺寸。若日后拿到更明确的文档或 +#: 实测结果，回来校正本处并更新取证页。 +MAX_EDGE = 3840 +MAX_PIXELS = 3840 * 2160 + +#: 超过此面积属实验性——放行但记一条日志，便于把偶发的异常表现与它关联起来。 +#: 同样按面积而非逐边理解，理由同上。 +EXPERIMENTAL_ABOVE_PIXELS = 2560 * 1440 + +#: 可选的画质档。xhigh 与 max 仅 2.5 系列支持，gpt-image-2 不含。 +QUALITIES_GPT_IMAGE_2 = frozenset({"low", "medium", "high", "auto"}) + +#: 输出格式。注意 transparent 背景要求 png 或 webp。 +OUTPUT_FORMATS = frozenset({"png", "jpeg", "webp"}) + + +@dataclass(frozen=True) +class Size: + width: int + height: int + + def __str__(self) -> str: + return f"{self.width}x{self.height}" + + +def parse_size(value: str) -> Size | None: + """把 `WIDTHxHEIGHT` 解析成尺寸；标准尺寸与 `auto` 之外的非法串返回 None。""" + match = _SIZE_PATTERN.match(value.strip()) + if not match: + return None + return Size(int(match.group(1)), int(match.group(2))) + + +def validate_size(value: str | None) -> None: + """校验尺寸，不合法则抛 `ValueError`（消息指明违反了哪一条约束）。 + + `None`、`auto`、标准尺寸一律放行。 + """ + if not value or value == AUTO_SIZE or value in STANDARD_SIZES: + return + + size = parse_size(value) + if size is None: + raise ValueError( + f"size 形态非法：{value!r}，应为 'WIDTHxHEIGHT'、'auto' " + f"或标准尺寸 {sorted(STANDARD_SIZES)}" + ) + + if size.width % EDGE_MULTIPLE or size.height % EDGE_MULTIPLE: + raise ValueError(f"size {size} 的宽高必须都能被 {EDGE_MULTIPLE} 整除") + + # 宽高比先判：它是文档最明确的一条，且比尺寸上限更常被违反（比例写错比 + # 尺寸写大更容易发生），先报它的错误信息更有用。 + aspect = size.width / size.height + if not MIN_ASPECT <= aspect <= MAX_ASPECT: + raise ValueError(f"size {size} 的宽高比 {aspect:.2f} 超出允许区间 1:3 ~ 3:1") + + if max(size.width, size.height) > MAX_EDGE: + raise ValueError(f"size {size} 的最长边超出上限 {MAX_EDGE}") + + pixels = size.width * size.height + if pixels > MAX_PIXELS: + raise ValueError( + f"size {size} 的总像素 {pixels} 超出上限 {MAX_PIXELS}" + f"（即 3840x2160 的像素量）" + ) + + if pixels > EXPERIMENTAL_ABOVE_PIXELS: + # 放行但记一条：文档称实验性，出问题时这条日志能把现象与它关联起来。 + log.info("size %s 超过 2560x1440，属服务商标注的实验性范围", size) + + +def validate_quality(value: str | None) -> None: + """校验画质档。 + + 只覆盖 gpt-image-2：`xhigh`/`max` 是 2.5 系列才支持的档，传给 gpt-image-2 + 会被拒。不按模型名分表是因为目前只接入一个图片模型——接入第二个时再分。 + """ + if value and value not in QUALITIES_GPT_IMAGE_2: + raise ValueError( + f"quality {value!r} 不被 gpt-image-2 支持，" + f"可选：{sorted(QUALITIES_GPT_IMAGE_2)}" + f"（xhigh / max 仅 gpt-image-2.5 系列支持）" + ) + + +def validate_output_format(value: str | None, *, transparent: bool = False) -> None: + """校验输出格式。 + + :param transparent: 背景为透明时**只能** png 或 webp——jpeg 没有 alpha 通道， + 指定它会得到一张背景被填成不透明的图，而那不会报错、只会静默不符合预期。 + """ + if value and value not in OUTPUT_FORMATS: + raise ValueError( + f"output_format {value!r} 非法，可选：{sorted(OUTPUT_FORMATS)}" + ) + if transparent and value == "jpeg": + raise ValueError( + "背景为 transparent 时 output_format 不能是 jpeg（无 alpha 通道），" + "请用 png 或 webp" + ) diff --git a/app/providers/openai/client.py b/app/providers/openai/client.py new file mode 100644 index 0000000..6f58369 --- /dev/null +++ b/app/providers/openai/client.py @@ -0,0 +1,161 @@ +"""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`。 + + 错误体按 OpenAI 通用约定解析（`{"error": {message, type, code, param}}`）。 + **该形状未经用户提供的文档证实**——两份文档都没有错误码一节，见 + `errors.py` 顶部说明。解析不出这个形状时不静默，按状态码分类并把原始 + 文本带上。 + """ + 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 + message = response.text[:500] + try: + error = response.json().get("error") or {} + error_type = error.get("type") + error_code = error.get("code") + message = error.get("message") or message + except ValueError: + pass + + failure = classify(response.status_code, error_type, error_code, message) + log.warning( + "OpenAI 返回错误 http=%s type=%s code=%s retry=%s", + response.status_code, + error_type, + error_code, + 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() diff --git a/app/providers/openai/errors.py b/app/providers/openai/errors.py new file mode 100644 index 0000000..c884cf8 --- /dev/null +++ b/app/providers/openai/errors.py @@ -0,0 +1,105 @@ +"""OpenAI 图片接口的错误 → 领域 `Failure`。 + +⚠️ **本模块的映射未经官方文档证实。** + +用户提供的两份文档（`gpt生图.md` / `gpt图像编辑.md`）只覆盖了请求参数与返回体， +**完全没有错误码表**——这与火山那边给了完整错误码表的情况不同。因此这里的映射 +建立在 OpenAI 通用错误约定之上：错误体形如 +`{"error": {"message", "type", "code", "param"}}`，并以 HTTP 状态码作为主要依据。 + +**处置策略**：以 **HTTP 状态码为主、`type`/`code` 为辅**。状态码语义是 HTTP 规范 +保证的，而 `type`/`code` 的取值集没有文档支撑，凭记忆列一张表只会得到一张看起来 +权威、实则可能错漏的映射。故这里对 `type`/`code` 只做**少量高置信度的加细**， +其余一律按状态码归类。 + +**待补**：拿到官方错误码文档、或在真实调用中观察到实际取值后，回来校正本模块并 +更新 patbond-doc 的取证页。在那之前，本文件顶部这段说明不要删。 +""" + +from __future__ import annotations + +import logging + +from app.domain.generation import Failure, Retry, UserFacing + +log = logging.getLogger(__name__) + +#: 高置信度的 `type` 取值：这几个在 OpenAI 的错误体里语义明确且与状态码一致， +#: 加细它们不会引入猜测——它们只是让分类比"仅看状态码"更准。 +#: 未列出的 type 一律回落到状态码判定。 +_TYPE_OVERRIDES: dict[str, tuple[Retry, UserFacing]] = { + # 配额/计费耗尽：等的是充值或下一个周期，重试无用。 + "insufficient_quota": (Retry.QUOTA, UserFacing.QUOTA_EXHAUSTED), + # 限流：等几秒。 + "rate_limit_exceeded": (Retry.BACKOFF, UserFacing.RATE_LIMITED), + # 内容被安全策略拒绝：同样的输入必然同样被拒。 + "image_generation_user_error": (Retry.NEVER, UserFacing.CONTENT_REJECTED), +} + +#: 同上，但匹配 `code` 字段。 +_CODE_OVERRIDES: dict[str, tuple[Retry, UserFacing]] = { + "insufficient_quota": (Retry.QUOTA, UserFacing.QUOTA_EXHAUSTED), + "rate_limit_exceeded": (Retry.BACKOFF, UserFacing.RATE_LIMITED), + "moderation_blocked": (Retry.NEVER, UserFacing.CONTENT_REJECTED), + "content_policy_violation": (Retry.NEVER, UserFacing.CONTENT_REJECTED), + "invalid_api_key": (Retry.NEVER, UserFacing.SERVER_ERROR), +} + + +def classify( + http_status: int | None, + error_type: str | None = None, + error_code: str | None = None, + message: str | None = None, +) -> Failure: + """把一次失败分类。 + + :param http_status: `None` 表示网络层失败（连接不上、超时），按可重试处理。 + + 未知情形一律偏向**可退避重试**而非判死：把一次可恢复的抖动判成用户可见的 + 失败是更糟的错误，而重试预算本身有上限（`max_attempts`）。 + """ + text = message or "" + # provider_code 优先用 code，其次 type——两者都缺时用状态码，保证这个字段 + # 永远不为空（它是排查时的第一线索）。 + provider_code = error_code or error_type or _status_code_label(http_status) + + override = _CODE_OVERRIDES.get(error_code or "") or _TYPE_OVERRIDES.get( + error_type or "" + ) + if override: + retry, user_facing = override + return Failure(retry, user_facing, provider_code, text) + + if http_status is None: + return Failure( + Retry.BACKOFF, UserFacing.UPSTREAM_UNAVAILABLE, provider_code, text + ) + + if http_status == 429: + # 429 在 OpenAI 上同时用于「限流」与「配额耗尽」，靠 type/code 区分； + # 上面的 override 没命中说明拿不到那两个字段，此时**按限流处理**： + # 猜成配额耗尽会让一次本可重试的限流被判死，代价更大。 + log.info("429 但无 type/code 可据，按限流处理（http=%s）", http_status) + return Failure(Retry.BACKOFF, UserFacing.RATE_LIMITED, provider_code, text) + + if http_status in (401, 403): + # 密钥无效或无权访问：我方配置问题，重试无用。 + return Failure(Retry.NEVER, UserFacing.SERVER_ERROR, provider_code, text) + + if http_status == 400: + # 参数问题**或**内容被拒——OpenAI 两者都用 400。拿不到 type/code 时无法 + # 区分，按不可重试处理（两种情形重试同一请求都不会变好），用户侧归 + # SERVER_ERROR 而不猜成 CONTENT_REJECTED：错报「内容被拒」会让用户去改 + # 一个其实没问题的提示词。 + return Failure(Retry.NEVER, UserFacing.SERVER_ERROR, provider_code, text) + + if 400 <= http_status < 500: + return Failure(Retry.NEVER, UserFacing.SERVER_ERROR, provider_code, text) + + # 5xx 与其余一切：可重试。 + return Failure(Retry.BACKOFF, UserFacing.UPSTREAM_UNAVAILABLE, provider_code, text) + + +def _status_code_label(http_status: int | None) -> str: + return f"HTTP_{http_status}" if http_status else "NETWORK_ERROR" diff --git a/app/providers/openai/image.py b/app/providers/openai/image.py new file mode 100644 index 0000000..b166d90 --- /dev/null +++ b/app/providers/openai/image.py @@ -0,0 +1,253 @@ +"""gpt-image-2 图片生成：文生图与图生图（图像编辑）。 + +## 与火山视频适配器的三处根本差异 + +1. **接口是同步的。** `POST /images/generations` 一次请求等到底，**没有任务 ID、 + 没有查询接口**。差异由 `tasks.SyncTaskRegistry` 在本服务内部吸收（ADR-023 §4）， + 代价与风险见该模块顶部说明。 +2. **产物是 base64 字节，不是 URL。** GPT image 系列**始终**返回 base64 + （`response_format` 对它们不支持）。好处是没有「URL 24 小时过期」问题；坏处是 + 字节只在内存里，必须立刻转存——而转存阻塞于存储决策（ADR-023 §7）。 +3. **图生图走 multipart，不是 JSON。** `/images/edits` 的 `image` 字段是文件。 + 我们的输入是一个 URL，故需先把它下载下来再作为 multipart 上传。 + +## 尺寸从响应直接拿，不需要对照表 + +响应带 `size` 字段（`WIDTHxHEIGHT`），故像素宽高是确定值——不像火山那边只给 +`ratio`+`resolution`、需要一张随模型族变化的对照表。 +""" + +from __future__ import annotations + +import logging + +import httpx + +from app.domain.generation import ( + Failure, + GenerationRequest, + GenerationResult, + MediaKind, + Retry, + TaskState, + UserFacing, +) +from app.providers.base import GenerationProvider + +from .capabilities import ( + parse_size, + validate_output_format, + validate_quality, + validate_size, +) +from .client import OpenAiClient +from .models import ImagesResponse, ImageSubmission +from .tasks import SyncTaskRegistry + +log = logging.getLogger(__name__) + +#: `creation.generation_models.provider_code` 里对应本适配器的值。 +PROVIDER_CODE = "openai-gpt-image" + +#: 下载参考图的超时。这是一次内部/公网取图，与生成耗时无关，故很短。 +_INPUT_FETCH_TIMEOUT_SEC = 15.0 + +#: 参考图大小上限。服务商限制单图 < 50MB；这里取更小的值，因为我们还要把它整个 +#: 读进内存再转发——一张几十 MB 的图会让内存占用随并发线性上涨。 +_MAX_INPUT_BYTES = 20 * 1024 * 1024 + + +class OpenAiImageProvider(GenerationProvider): + """gpt-image-2 的图片生成适配器。""" + + def __init__( + self, + client: OpenAiClient, + registry: SyncTaskRegistry, + default_model: str = "", + input_fetch_client: httpx.Client | None = None, + ): + self._client = client + self._registry = registry + self._default_model = default_model + self._owns_fetch_client = input_fetch_client is None + self._fetch_client = input_fetch_client or httpx.Client( + timeout=_INPUT_FETCH_TIMEOUT_SEC, follow_redirects=True + ) + + @property + def provider_code(self) -> str: + return PROVIDER_CODE + + @property + def media_kind(self) -> MediaKind: + return MediaKind.IMAGE + + # ------------------------------------------------------------- submit + + def submit(self, request: GenerationRequest) -> str: + if request.media_kind is not MediaKind.IMAGE: + raise ValueError( + f"{PROVIDER_CODE} 只处理图片生成，收到 {request.media_kind.value}" + ) + + submission = self._to_submission(request) + # 参数校验放在提交进注册表**之前**：违规请求应当立刻以 400 回给调用方， + # 而不是变成一个注定失败的后台任务——后者要等一轮轮询才被发现，还占了一个 + # 线程池槽位。 + validate_size(submission.size) + validate_quality(submission.quality) + validate_output_format( + submission.output_format, transparent=submission.background == "transparent" + ) + + if request.input_url: + # 图生图：先取图。**取图也放在受理之前**——取不到就是参数问题 + # （URL 不可达），应当立刻拒绝而不是变成一个后台失败。 + image_bytes, filename, content_type = self._fetch_input(request.input_url) + work = lambda: self._run_edit( + submission, image_bytes, filename, content_type + ) + else: + work = lambda: self._run_generate(submission) + + # 注册表满时 `submit` 抛 RuntimeError，这里**刻意不接**：归成 ValueError 让上层 + # 映射成 invalid_request 是错的——那不是请求的问题而是容量问题，且稍后重试会 + # 好。原样上抛，由端点映射为 provider_unavailable（503）。 + return self._registry.submit(work) + + def _to_submission(self, request: GenerationRequest) -> ImageSubmission: + """领域请求 → 厂商请求体。 + + 模型优先取请求里的（来自任务行冻结的快照），仅在缺失时回落到配置的默认值 + ——适配层不自行选模型，且任务行的快照必须与实际调用一致，否则无法对账。 + """ + model = request.model or self._default_model + if not model: + raise ValueError( + "未指定模型：请求未携带 model 且 OPENAI_IMAGE_MODEL 未配置" + ) + + return ImageSubmission( + model=model, + prompt=request.prompt or "", + # 领域层的 ratio/resolution 是为视频设计的；图片用像素尺寸。宽高都有 + # 才拼 size，否则交给模型自选（auto）——半个尺寸不如不给。 + size=( + f"{request.width_px}x{request.height_px}" + if request.width_px and request.height_px + else None + ), + safety_identifier=request.safety_identifier, + ) + + # -------------------------------------------------------------- fetch + + def fetch(self, provider_request_id: str) -> GenerationResult: + result = self._registry.fetch(provider_request_id) + if result is None: + # 未知 ID：从未存在过，或进程重启后丢失（两者无法区分，见 tasks.py）。 + # 抛 KeyError 让端点映射成 task_not_found——调用方要做的是停止轮询， + # 而不是退避重试。 + raise KeyError(provider_request_id) + return result + + # ------------------------------------------------- 实际的同步调用 + + def _run_generate(self, submission: ImageSubmission) -> GenerationResult: + """文生图。在注册表的后台线程里执行。""" + payload = self._client.post_json( + "/images/generations", submission.to_json_body() + ) + return self._to_result(ImagesResponse.from_payload(payload)) + + def _run_edit( + self, + submission: ImageSubmission, + image_bytes: bytes, + filename: str, + content_type: str, + ) -> GenerationResult: + """图生图（图像编辑）。走 multipart。""" + payload = self._client.post_multipart( + "/images/edits", + data=submission.to_form_fields(), + # 字段名用 `image[]`：编辑接口支持多图且共用该字段名。单图时也用它， + # 避免"单图一种写法、多图另一种"的分叉。 + files=[("image[]", (filename, image_bytes, content_type))], + ) + return self._to_result(ImagesResponse.from_payload(payload)) + + def _to_result(self, response: ImagesResponse) -> GenerationResult: + """厂商响应 → 领域结果。""" + image = response.first_image() + if image is None or not image.b64_json: + # 2xx 却没有图片数据。当成可重试的上游异常，而不是把一个没有产物的 + # 「成功」放给上层去空指针。 + log.error("OpenAI 返回 2xx 但缺少图片数据：%r", response.raw_keys()) + return GenerationResult( + provider_request_id="", + state=TaskState.FAILED, + failure=Failure( + Retry.BACKOFF, + UserFacing.UPSTREAM_UNAVAILABLE, + "MissingImageData", + "服务商返回成功但响应中没有图片数据", + ), + ) + + size = parse_size(response.size or "") + return GenerationResult( + # 真实 ID 由注册表在 fetch 时填入——这里填不了，因为工作函数运行时 + # 还拿不到自己的任务 ID。 + provider_request_id="", + state=TaskState.SUCCEEDED, + progress=100, + # ⚠️ 产物是 base64 字节而非可访问 URL。这里放 data URI 只是为了让领域 + # 模型的「SUCCEEDED 必须有 asset_url」不变式成立，**它不是一个可长期 + # 保存的地址**：一张 1024×1024 的 png 约 1~2MB，base64 后更大，绝不 + # 应该原样写进 media.assets.external_url。转存落地后这里应改为返回 + # 上传后的 object key。 + asset_url=f"data:image/{response.output_format or 'png'};base64," + + image.b64_json, + # 无时效——字节在手，不像火山 URL 会过期。用 None 而非一个假的秒数。 + asset_url_expires_in_sec=None, + width_px=size.width if size else None, + height_px=size.height if size else None, + output_format=response.output_format, + billing_tokens=response.total_tokens, + ) + + # ------------------------------------------------------------ helpers + + def _fetch_input(self, url: str) -> tuple[bytes, str, str]: + """下载参考图，供 multipart 上传。 + + 服务商要求 png / webp / jpg 且单图 < 50MB。这里额外设了更小的上限，因为 + 字节要整个进内存再转发。 + """ + try: + with self._fetch_client.stream("GET", url) as response: + response.raise_for_status() + declared = response.headers.get("Content-Length") + if declared and int(declared) > _MAX_INPUT_BYTES: + raise ValueError( + f"参考图过大（{declared} 字节），上限 {_MAX_INPUT_BYTES}" + ) + chunks: list[bytes] = [] + total = 0 + for chunk in response.iter_bytes(): + total += len(chunk) + if total > _MAX_INPUT_BYTES: + # 不信 Content-Length：它可能缺失或撒谎，边读边数才拦得住。 + raise ValueError(f"参考图超过上限 {_MAX_INPUT_BYTES} 字节") + chunks.append(chunk) + content_type = response.headers.get("Content-Type", "image/png") + except httpx.HTTPError as exc: + raise ValueError(f"参考图无法下载：{url}（{exc}）") from exc + + return b"".join(chunks), "input.png", content_type.split(";")[0] + + def close(self) -> None: + if self._owns_fetch_client: + self._fetch_client.close() diff --git a/app/providers/openai/models.py b/app/providers/openai/models.py new file mode 100644 index 0000000..0e1debc --- /dev/null +++ b/app/providers/openai/models.py @@ -0,0 +1,149 @@ +"""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, + ) diff --git a/app/providers/openai/tasks.py b/app/providers/openai/tasks.py new file mode 100644 index 0000000..7fd7e93 --- /dev/null +++ b/app/providers/openai/tasks.py @@ -0,0 +1,224 @@ +"""同步型供应商的任务注册表：把「一次调用等到底」转成「提交 + 查询」。 + +## 为什么必须有它 + +ADR-023 §4 定的是「Java↔Python 边界一律异步，同步型供应商由其适配层在内部消化 +这个差异」。火山的视频接口本身异步，所以视频适配器不需要这一层。**OpenAI 的图片 +接口是同步的**——`POST /images/generations` 在一次请求里等到图片生成完才返回， +没有任务 ID，也没有查询接口。差异必须在这里被吸收，否则就会渗透到客户端契约、 +等待态设计和 E2E，那是 §4 明确要避免的。 + +## 这与「Python 不持有状态」的张力，以及取舍 + +Seedance 那条路径上，Python 确实无状态：任务状态的唯一事实来源是 Java 的 +`generation_jobs`，Python 只是个转发器。**图片这条路径做不到**——同步接口没有可 +供事后查询的厂商 ID，所以「在途任务」这件事只能记在本进程内存里。 + +代价与处置如实登记： + +1. **进程重启会丢失在途任务。** 之后 Java 轮询会得到 `task_not_found`，任务被判 + 终态失败，而**那次调用可能已经完成并计费**。OpenAI 的图片接口没有任务 ID 可供 + 事后查询，所以这个窗口**无法靠"再查一次"弥补**，只能由 `max_attempts` 兜住。 + 这与火山那边「提交超时不知道有没有扣费」是同一类风险，但触发面更大（那边只有 + 提交超时，这边还包括进程重启）。 +2. **产物是 base64 字节，只在内存里。** 完成的任务必须在 TTL 内被取走，否则丢失。 + 而转存阻塞于对象存储决策（ADR-023 §7），所以**图片链路在存储决策落地前比视频 + 更受限**：视频至少有个 24 小时的厂商 URL 可以先看效果，图片的字节一旦被清理 + 就什么都没有了。 +3. 注册表**有界且有 TTL**，否则它就是个内存泄漏。 + +## 为什么用线程而不是 asyncio + +FastAPI 的端点在本服务里是同步函数（`def` 而非 `async def`），且 `httpx.Client` +是同步客户端。用线程池承载那个长调用与现有代码一致；改成 asyncio 需要把传输层、 +端点、测试全部改成异步，收益只是少几个线程——本服务的并发量级（一台轻量服务器上 +的少量生成请求）用不上那个收益。 +""" + +from __future__ import annotations + +import logging +import threading +import time +import uuid +from collections.abc import Callable +from concurrent.futures import Future, ThreadPoolExecutor +from dataclasses import dataclass, field + +from app.domain.generation import Failure, GenerationResult, TaskState + +log = logging.getLogger(__name__) + +#: 本地任务 ID 的前缀，便于在日志里与厂商 ID 区分开。 +LOCAL_TASK_PREFIX = "local-" + + +@dataclass +class _Entry: + """一条在途或已完成的任务。""" + + future: Future + created_at: float = field(default_factory=time.monotonic) + #: **首次观察到**完成的时刻；未完成为 None。TTL 从这里起算而不是从创建起算—— + #: 否则一个耗时接近 TTL 的任务会在刚完成时就被清理掉，结果还没被取走。 + #: + #: 为什么是「首次观察到」而不是「实际完成」：实际完成时刻需要一个 + #: `add_done_callback` 才能捕捉，而那会**死锁**——快速完成的任务在 + #: `add_done_callback` 时已经 done，回调便**同步地在调用线程执行**，而该线程 + #: 此刻正持有本注册表的锁（非可重入）。改用惰性记录既无此风险，又只会让结果 + #: 多留一会儿（保守方向），不会让它提前消失。 + finished_at: float | None = None + + +class SyncTaskRegistry: + """把同步调用包装成可轮询的任务。 + + 线程安全：`submit`/`fetch`/清理可能来自不同请求线程。 + """ + + def __init__( + self, + *, + max_workers: int = 4, + completed_ttl_sec: float = 900.0, + max_entries: int = 256, + ): + """ + :param max_workers: 并发执行的同步调用数上限。**同时也是对服务商并发的 + 自我限制**：超出的提交会排队而不是一起打过去，既避免把自己的 RPM 打爆， + 也让「同时有多少钱在花」有个上界。 + :param completed_ttl_sec: 已完成任务的保留时长。需大于 Java 侧最长的轮询 + 间隔（当前阶梯最长 30 秒），否则结果会在被取走之前被清理。默认 15 + 分钟留足余量。 + :param max_entries: 注册表容量上限。满了先清理过期项，仍满则拒绝新提交 + （而不是无声挤掉某个在途任务——那会让一个已经在计费的调用变成孤儿）。 + """ + self._executor = ThreadPoolExecutor( + max_workers=max_workers, thread_name_prefix="sync-gen" + ) + self._completed_ttl_sec = completed_ttl_sec + self._max_entries = max_entries + self._entries: dict[str, _Entry] = {} + self._lock = threading.Lock() + + def submit(self, work: Callable[[], GenerationResult]) -> str: + """安排一次同步调用在后台执行，立刻返回本地任务 ID。 + + :raises RuntimeError: 注册表已满（并发在途任务过多） + """ + with self._lock: + self._evict_expired_locked() + if len(self._entries) >= self._max_entries: + raise RuntimeError( + f"在途生成任务已达上限 {self._max_entries}，请稍后重试" + ) + task_id = LOCAL_TASK_PREFIX + uuid.uuid4().hex + # 刻意不注册 done_callback，原因见 _Entry.finished_at 的说明（会死锁）。 + self._entries[task_id] = _Entry(future=self._executor.submit(work)) + log.info("已受理同步生成任务 %s（在途 %d）", task_id, len(self._entries)) + return task_id + + def fetch(self, task_id: str) -> GenerationResult | None: + """查任务；未知 ID 返回 None。 + + 未知 ID 的两种成因**无法区分**：从未存在过，或进程重启后丢失。两者对调用 + 方的处置相同（停止轮询、任务判失败），故不强行区分——区分不了的事不要假装 + 区分。 + """ + with self._lock: + entry = self._entries.get(task_id) + if entry is not None and entry.finished_at is None and entry.future.done(): + # 首次观察到完成：记下时刻，TTL 由此起算。 + entry.finished_at = time.monotonic() + if entry is None: + return None + + if not entry.future.done(): + # 同步接口不提供进度，故只能报 0。客户端契约本就不承诺平滑进度 + # （ADR-023 §10），一个诚实的 0 不该被读成「卡住」。 + return GenerationResult( + provider_request_id=task_id, state=TaskState.RUNNING + ) + + try: + result = entry.future.result() + except Exception as exc: # noqa: BLE001 - 兜底转成领域失败 + return self._failure_result(task_id, exc) + # 用真实的任务 ID 覆盖工作函数里可能留空的那个字段。 + return GenerationResult( + provider_request_id=task_id, + state=result.state, + progress=result.progress, + asset_url=result.asset_url, + asset_url_expires_in_sec=result.asset_url_expires_in_sec, + cover_url=result.cover_url, + ratio=result.ratio, + resolution=result.resolution, + width_px=result.width_px, + height_px=result.height_px, + duration_ms=result.duration_ms, + output_format=result.output_format, + failure=result.failure, + billing_tokens=result.billing_tokens, + ) + + @staticmethod + def _failure_result(task_id: str, exc: BaseException) -> GenerationResult: + """工作函数抛出的异常转成领域失败。 + + 已分类的服务商异常保留其分类；其余按可重试的上游问题处理，并把类型名带进 + `provider_code`——否则排查时只有一句空洞的错误消息。 + """ + from app.domain.generation import Retry, UserFacing + + from .client import OpenAiError + + if isinstance(exc, OpenAiError): + failure = exc.failure + else: + log.exception("同步生成任务 %s 抛出未分类异常", task_id) + failure = Failure( + Retry.BACKOFF, + UserFacing.UPSTREAM_UNAVAILABLE, + type(exc).__name__, + str(exc), + ) + return GenerationResult( + provider_request_id=task_id, state=TaskState.FAILED, failure=failure + ) + + def _evict_expired_locked(self) -> None: + """清理已完成且超过 TTL 的条目。调用方必须已持锁。 + + **只清已完成的**：在途任务无论多久都不清——它可能正在花钱，清掉会让它 + 变成一个无人认领的开销。真正卡死的在途任务由线程池的上限兜住。 + """ + now = time.monotonic() + # 顺带给「已完成但从未被 fetch 过」的条目补上观察时刻，否则它们的 + # finished_at 永远是 None，就永远不会过期——那正是内存泄漏。 + for entry in self._entries.values(): + if entry.finished_at is None and entry.future.done(): + entry.finished_at = now + expired = [ + task_id + for task_id, entry in self._entries.items() + if entry.finished_at is not None + and now - entry.finished_at > self._completed_ttl_sec + ] + for task_id in expired: + del self._entries[task_id] + if expired: + log.debug("清理了 %d 个已完成的同步任务条目", len(expired)) + + @property + def size(self) -> int: + with self._lock: + return len(self._entries) + + def shutdown(self) -> None: + """停止接受新任务。 + + `wait=False`：在途的同步调用可能还要跑几分钟，进程退出时不等它们——结果 + 反正已经无处可交付了。它们会随进程一起消失，这正是上面第 1 点登记的风险。 + """ + self._executor.shutdown(wait=False, cancel_futures=True) diff --git a/openapi-internal.yaml b/openapi-internal.yaml index 09c184f..77dc66f 100644 --- a/openapi-internal.yaml +++ b/openapi-internal.yaml @@ -32,6 +32,18 @@ info: `service` 类属我方 bug 或配置问题（除 `internal_error`/`provider_unavailable` 外均不可重试）；`provider` 类一律以 `retry` 字段为准，不要自己解读厂商码。 + ## 一处与服务商形态有关的说明 + + **服务商可能是同步的。** 火山的视频接口本身异步（提交返回任务 ID）；OpenAI 的 + 图片接口是**同步**的——一次请求等到底，没有任务 ID 也没有查询接口。该差异由本 + 服务在内部吸收（ADR-023 §4），故本契约对两者形状完全一致。 + + 代价如实登记：同步型供应商的在途任务只能记在本服务内存里，**进程重启会丢失**， + 之后查询返回 `task_not_found`。那次调用可能已经完成并计费，而同步接口没有任务 + ID 可供事后查询，故该窗口**无法靠"再查一次"弥补**，只能由调用方的 + `max_attempts` 兜住。并发在途任务达上限时提交返回 `provider_unavailable` + （503，可重试）而非 `invalid_request`。 + ## 鉴权 所有端点要求服务身份 JWT：`Authorization: Bearer `。本服务强制校验 @@ -192,8 +204,25 @@ components: resolution: type: [string, 'null'] description: | - 缺省 `720p`。按模型预校验：2.5 无 4k，2.0 系列有。 + 缺省 `720p`。按模型预校验：2.5 无 4k，2.0 系列有。**视频路径专用**。 example: 720p + width_px: + type: [integer, 'null'] + description: | + **图片路径专用**的像素宽。与 `height_px` 必须同时给，否则交由模型自选 + （半个尺寸不如不给）。 + + 两家服务商的尺寸表达方式不同：火山用「分辨率档 + 宽高比」 + （`resolution` + `ratio`），OpenAI 用 `WIDTHxHEIGHT` 像素串。契约同时 + 保留两种而**不强行统一**——任何统一都会在某一侧失真：把 720p+16:9 折成 + 像素会丢掉「档位」语义，把像素折成比例会丢掉精确值。 + + OpenAI 侧的算术约束会在本服务**本地预校验**：宽高都须被 16 整除、 + 宽高比在 1:3 ~ 3:1 之间、不超过 3840x2160（超过 2560x1440 属服务商 + 标注的实验性范围）。违规请求以 `invalid_request` 拒绝，不花一次往返。 + height_px: + type: [integer, 'null'] + description: 见 `width_px`。 duration_sec: type: [integer, 'null'] description: | diff --git a/tests/providers/openai/__init__.py b/tests/providers/openai/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/providers/openai/test_capabilities.py b/tests/providers/openai/test_capabilities.py new file mode 100644 index 0000000..f7580a0 --- /dev/null +++ b/tests/providers/openai/test_capabilities.py @@ -0,0 +1,150 @@ +"""gpt-image-2 尺寸与参数约束的测试。 + +每个用例对应官方文档里的一条约束。这些约束若搞错，表现是用户提交后拿到一个失败 +任务（服务商 400），而本地测试全绿——所以值得逐条钉住。 +""" + +from __future__ import annotations + +import pytest + +from app.providers.openai.capabilities import ( + parse_size, + validate_output_format, + validate_quality, + validate_size, +) + + +class TestStandardSizesAndAutoAlwaysPass: + """文档明示支持的取值不做算术校验。""" + + def test_standard_sizes(self): + for size in ("1024x1024", "1536x1024", "1024x1536"): + validate_size(size) + + def test_auto(self): + validate_size("auto") + + def test_none(self): + validate_size(None) + + +class TestEdgeMultipleOf16: + """官方：宽高都必须能被 16 整除。""" + + def test_both_divisible_passes(self): + validate_size("1536x864") # 文档给的例子 + + def test_width_not_divisible_is_rejected(self): + with pytest.raises(ValueError, match="16 整除"): + validate_size("1000x864") + + def test_height_not_divisible_is_rejected(self): + with pytest.raises(ValueError, match="16 整除"): + validate_size("1536x865") + + +class TestAspectRatioBounds: + """官方：宽高比必须在 1:3 ~ 3:1 之间。""" + + def test_three_to_one_is_allowed(self): + # 2880x960 = 3:1，且都能被 16 整除。 + validate_size("2880x960") + + def test_one_to_three_is_allowed(self): + validate_size("960x2880") + + def test_portrait_at_the_pixel_limit_is_allowed(self): + """竖版 4k（2160x3840）与横版同像素量，必须放行。 + + 按「逐边」理解上限会禁掉所有高于 2160 的竖图，而 1024x1536 是文档明示的 + 标准竖图尺寸——故逐边理解必然过严。见 capabilities 里的说明。 + """ + validate_size("2160x3840") + + def test_wider_than_three_to_one_is_rejected(self): + # 3840x960 = 4:1。 + with pytest.raises(ValueError, match="宽高比"): + validate_size("3840x960") + + def test_taller_than_one_to_three_is_rejected(self): + # 640x2880 = 1:4.5，且在像素上限之内，故触发的必须是宽高比那条。 + with pytest.raises(ValueError, match="宽高比"): + validate_size("640x2880") + + +class TestMaximumResolution: + def test_the_maximum_itself_is_allowed(self): + validate_size("3840x2160") + + def test_longest_edge_beyond_the_cap_is_rejected(self): + with pytest.raises(ValueError, match="最长边"): + validate_size("4096x2048") + + def test_within_edge_cap_but_beyond_the_pixel_cap_is_rejected(self): + # 3840x2688：最长边合规、宽高比合规，但总像素超了。 + with pytest.raises(ValueError, match="总像素"): + validate_size("3840x2688") + + def test_a_tall_portrait_within_both_caps_passes(self): + validate_size("1024x2176") + + +class TestMalformedSize: + def test_not_a_size_string(self): + with pytest.raises(ValueError, match="形态非法"): + validate_size("large") + + def test_missing_separator(self): + with pytest.raises(ValueError, match="形态非法"): + validate_size("1024") + + def test_parse_returns_none_for_garbage(self): + assert parse_size("nonsense") is None + + def test_parse_reads_a_valid_pair(self): + size = parse_size("1536x864") + assert (size.width, size.height) == (1536, 864) + assert str(size) == "1536x864" + + +class TestQuality: + def test_supported_values_pass(self): + for quality in ("low", "medium", "high", "auto"): + validate_quality(quality) + + def test_xhigh_is_rejected_because_it_is_2_5_only(self): + # 官方：xhigh 与 max 仅 gpt-image-2.5 系列支持。 + with pytest.raises(ValueError, match="xhigh"): + validate_quality("xhigh") + + def test_max_is_rejected_too(self): + with pytest.raises(ValueError): + validate_quality("max") + + def test_none_passes(self): + validate_quality(None) + + +class TestOutputFormat: + def test_supported_formats(self): + for fmt in ("png", "jpeg", "webp"): + validate_output_format(fmt) + + def test_unknown_format_is_rejected(self): + with pytest.raises(ValueError, match="非法"): + validate_output_format("gif") + + def test_jpeg_with_transparent_background_is_rejected(self): + """jpeg 没有 alpha 通道。 + + 服务商不会因此报错——只会给一张背景被填成不透明的图，而那是**静默的** + 不符合预期，比报错难查得多。 + """ + with pytest.raises(ValueError, match="alpha"): + validate_output_format("jpeg", transparent=True) + + def test_png_with_transparent_background_is_fine(self): + validate_output_format("png", transparent=True) + validate_output_format("webp", transparent=True) diff --git a/tests/providers/openai/test_client.py b/tests/providers/openai/test_client.py new file mode 100644 index 0000000..9601c8a --- /dev/null +++ b/tests/providers/openai/test_client.py @@ -0,0 +1,202 @@ +"""OpenAI HTTP 传输层的测试：鉴权、两种载荷形式、错误解包。 + +全部离线（`httpx.MockTransport`），不触网。 +""" + +from __future__ import annotations + +import httpx +import pytest + +from app.domain.generation import Retry +from app.providers.openai.client import ( + DEFAULT_BASE_URL, + OpenAiClient, + OpenAiError, +) + +FAKE_API_KEY = "dummy-api-key-for-tests" + + +def client_with(handler, **kwargs) -> OpenAiClient: + return OpenAiClient( + api_key=FAKE_API_KEY, + client=httpx.Client(transport=httpx.MockTransport(handler)), + **kwargs, + ) + + +class TestCredentialDiscipline: + def test_an_empty_key_fails_loudly_at_construction(self): + """缺密钥是配置问题，应当在装配时就说清楚，而不是等第一次调用。""" + with pytest.raises(ValueError, match="OPENAI_API_KEY 未配置"): + OpenAiClient(api_key="") + + def test_the_key_is_sent_as_a_bearer_token(self): + seen: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["auth"] = request.headers.get("Authorization") + return httpx.Response(200, json={"ok": True}) + + client_with(handler).post_json("/images/generations", {}) + assert seen["auth"] == f"Bearer {FAKE_API_KEY}" + + def test_organization_header_is_sent_only_when_configured(self): + seen: list = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request.headers.get("OpenAI-Organization")) + return httpx.Response(200, json={}) + + client_with(handler).post_json("/x", {}) + client_with(handler, organization="org-abc").post_json("/x", {}) + assert seen == [None, "org-abc"] + + +class TestUrlComposition: + def test_default_base_url_is_the_official_one(self): + assert DEFAULT_BASE_URL == "https://api.openai.com/v1" + + def test_a_trailing_slash_on_the_base_url_does_not_double_up(self): + seen: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["url"] = str(request.url) + return httpx.Response(200, json={}) + + client_with(handler, base_url="https://proxy.invalid/v1/").post_json("/img", {}) + assert seen["url"] == "https://proxy.invalid/v1/img" + + +class TestPayloadForms: + def test_post_json_sends_json(self): + seen: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["content_type"] = request.headers.get("Content-Type") + seen["body"] = request.content + return httpx.Response(200, json={"echo": 1}) + + result = client_with(handler).post_json("/x", {"prompt": "hi"}) + assert seen["content_type"] == "application/json" + assert b'"prompt"' in seen["body"] + assert result == {"echo": 1} + + def test_post_multipart_repeats_a_field_name(self): + """用 list 而非 dict 传 files 的全部理由：同名字段可以重复。 + + 编辑端点支持多图，它们共用字段名 `image[]`——dict 表达不了。 + """ + seen: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["content_type"] = request.headers.get("Content-Type", "") + seen["raw"] = request.content + return httpx.Response(200, json={}) + + client_with(handler).post_multipart( + "/images/edits", + data={"prompt": "merge"}, + files=[ + ("image[]", ("a.png", b"first", "image/png")), + ("image[]", ("b.png", b"second", "image/png")), + ], + ) + + assert "multipart/form-data" in seen["content_type"] + assert seen["raw"].count(b'name="image[]"') == 2 + assert b"first" in seen["raw"] + assert b"second" in seen["raw"] + + def test_content_type_is_not_hand_written(self): + """multipart 还需要 boundary，手写必错——交给 httpx 按载荷推断。""" + seen: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["ct"] = request.headers.get("Content-Type", "") + return httpx.Response(200, json={}) + + client_with(handler).post_multipart( + "/x", data={}, files=[("image[]", ("a", b"x", "image/png"))] + ) + assert "boundary=" in seen["ct"] + + +class TestErrorUnwrapping: + def test_the_vendor_error_envelope_is_parsed(self): + def handler(request): + return httpx.Response( + 429, + json={ + "error": { + "message": "Rate limit reached", + "type": "rate_limit_exceeded", + "code": None, + "param": None, + } + }, + ) + + with pytest.raises(OpenAiError) as caught: + client_with(handler).post_json("/x", {}) + + error = caught.value + assert error.http_status == 429 + assert error.failure.retry is Retry.BACKOFF + assert "Rate limit reached" in error.failure.message + + def test_a_non_json_error_body_still_classifies_by_status(self): + """错误体形状未经文档证实，解析不出时不能静默丢掉线索。""" + + def handler(request): + return httpx.Response(502, text="bad gateway") + + with pytest.raises(OpenAiError) as caught: + client_with(handler).post_json("/x", {}) + + assert caught.value.failure.retry is Retry.BACKOFF + assert caught.value.failure.provider_code == "HTTP_502" + # 原始文本被带上，否则排查时只知道「502」。 + assert "bad gateway" in caught.value.failure.message + + def test_a_2xx_with_a_non_json_body_is_an_error_not_a_success(self): + def handler(request): + return httpx.Response(200, text="not json at all") + + with pytest.raises(OpenAiError, match="不是合法 JSON"): + client_with(handler).post_json("/x", {}) + + def test_a_network_failure_is_classified_as_retryable(self): + """⚠️ 同步接口的超时最危险：那次调用可能已经完成并计费。 + + 图片接口没有任务 ID 可供事后查询，故这个窗口只能由 max_attempts 兜住。 + """ + + def handler(request): + raise httpx.ConnectTimeout("timed out") + + with pytest.raises(OpenAiError) as caught: + client_with(handler).post_json("/x", {}) + + assert caught.value.http_status is None + assert caught.value.failure.retry is Retry.BACKOFF + assert caught.value.failure.provider_code == "NETWORK_ERROR" + + +class TestConnectionOwnership: + def test_an_injected_client_is_not_closed_by_us(self): + """注入进来的连接池归调用方管理。 + + 关掉它会让共享同一个 client 的其他适配器一起失效。 + """ + injected = httpx.Client( + transport=httpx.MockTransport(lambda r: httpx.Response(200, json={})) + ) + OpenAiClient(api_key=FAKE_API_KEY, client=injected).close() + assert not injected.is_closed + + def test_our_own_client_is_closed_on_exit(self): + with OpenAiClient(api_key=FAKE_API_KEY) as client: + inner = client._client + assert inner.is_closed diff --git a/tests/providers/openai/test_errors.py b/tests/providers/openai/test_errors.py new file mode 100644 index 0000000..de6643c --- /dev/null +++ b/tests/providers/openai/test_errors.py @@ -0,0 +1,119 @@ +"""OpenAI 错误分类的逐条测试。 + +`errors.py` 顶部标了「映射未经官方文档证实」——正因为它是推断出来的，才更要把 +每条分支钉住：日后拿到真实错误数据来校正时，改动的影响面一眼可见。 +""" + +from __future__ import annotations + +import pytest + +from app.domain.generation import Retry, UserFacing +from app.providers.openai.errors import classify + + +class TestHighConfidenceOverrides: + """文档没给错误码表，故只对少数语义明确的取值加细。""" + + @pytest.mark.parametrize("field", ["code", "type"]) + def test_insufficient_quota_is_not_worth_retrying(self, field): + failure = classify(429, **{f"error_{field}": "insufficient_quota"}) + assert failure.retry is Retry.QUOTA + assert failure.user_facing is UserFacing.QUOTA_EXHAUSTED + + @pytest.mark.parametrize("field", ["code", "type"]) + def test_rate_limit_backs_off(self, field): + failure = classify(429, **{f"error_{field}": "rate_limit_exceeded"}) + assert failure.retry is Retry.BACKOFF + assert failure.user_facing is UserFacing.RATE_LIMITED + + @pytest.mark.parametrize("code", ["moderation_blocked", "content_policy_violation"]) + def test_content_rejection_is_terminal(self, code): + # 同样的输入必然同样被拒——重试只是白花一次配额。 + failure = classify(400, error_code=code) + assert failure.retry is Retry.NEVER + assert failure.user_facing is UserFacing.CONTENT_REJECTED + + def test_invalid_key_is_our_problem_not_the_users(self): + failure = classify(401, error_code="invalid_api_key") + assert failure.retry is Retry.NEVER + # 用户看到的是「服务端错误」，不是「你的内容有问题」。 + assert failure.user_facing is UserFacing.SERVER_ERROR + + def test_code_wins_over_type_when_both_are_present(self): + # code 比 type 更具体，故优先。 + failure = classify( + 429, error_type="rate_limit_exceeded", error_code="insufficient_quota" + ) + assert failure.retry is Retry.QUOTA + + +class TestStatusCodeFallback: + """拿不到 type/code 时只靠状态码——这是本模块的主依据。""" + + def test_429_without_any_hint_is_treated_as_rate_limiting(self): + """429 两义兼有；猜成配额耗尽会把一次可重试的限流判死，代价更大。""" + failure = classify(429) + assert failure.retry is Retry.BACKOFF + assert failure.user_facing is UserFacing.RATE_LIMITED + + def test_400_without_any_hint_does_not_blame_the_prompt(self): + """400 同时用于参数错与内容被拒。 + + 错报「内容被拒」会让用户去改一个其实没问题的提示词，故归服务端错误。 + """ + failure = classify(400) + assert failure.retry is Retry.NEVER + assert failure.user_facing is UserFacing.SERVER_ERROR + + @pytest.mark.parametrize("status", [401, 403]) + def test_auth_failures_are_terminal(self, status): + assert classify(status).retry is Retry.NEVER + + @pytest.mark.parametrize("status", [404, 409, 422]) + def test_other_4xx_are_terminal(self, status): + assert classify(status).retry is Retry.NEVER + + @pytest.mark.parametrize("status", [500, 502, 503, 504]) + def test_5xx_is_retryable(self, status): + failure = classify(status) + assert failure.retry is Retry.BACKOFF + assert failure.user_facing is UserFacing.UPSTREAM_UNAVAILABLE + + def test_network_failure_is_retryable(self): + # http_status=None 表示连都没连上。 + failure = classify(None, message="connect timeout") + assert failure.retry is Retry.BACKOFF + assert failure.user_facing is UserFacing.UPSTREAM_UNAVAILABLE + + def test_an_unknown_status_leans_towards_retrying(self): + """未知一律偏可重试：把可恢复的抖动判死是更糟的错误。 + + 重试预算本身有上限（max_attempts），所以偏向重试是有界的。 + """ + assert classify(599).retry is Retry.BACKOFF + + +class TestProviderCodeIsNeverEmpty: + """它是排查时的第一线索，空字符串等于把线索丢了。""" + + def test_code_is_preferred(self): + assert classify(400, error_type="t", error_code="c").provider_code == "c" + + def test_type_is_used_when_code_is_absent(self): + assert classify(400, error_type="t").provider_code == "t" + + def test_status_label_is_the_last_resort(self): + assert classify(503).provider_code == "HTTP_503" + + def test_network_failures_get_a_named_label_too(self): + assert classify(None).provider_code == "NETWORK_ERROR" + + +class TestMessageIsCarriedThrough: + def test_message_is_kept(self): + assert classify(400, message="bad size").message == "bad size" + + def test_missing_message_becomes_empty_not_none(self): + # 契约里 message 是字符串；None 会在序列化处变成 null 而非 ""。 + assert classify(400).message == "" diff --git a/tests/providers/openai/test_image.py b/tests/providers/openai/test_image.py new file mode 100644 index 0000000..624258d --- /dev/null +++ b/tests/providers/openai/test_image.py @@ -0,0 +1,405 @@ +"""gpt-image-2 适配器的测试：文生图、图生图、响应映射。 + +全部离线（`httpx.MockTransport`），并用同步的注册表配置（`max_workers=1`）+ 轮询 +等待，使断言不依赖时序。 +""" + +from __future__ import annotations + +import base64 +import time + +import httpx +import pytest + +from app.domain.generation import ( + GenerationRequest, + MediaKind, + Retry, + TaskState, + UserFacing, +) +from app.providers.openai.client import OpenAiClient +from app.providers.openai.image import PROVIDER_CODE, OpenAiImageProvider +from app.providers.openai.tasks import SyncTaskRegistry + +FAKE_API_KEY = "dummy-api-key-for-tests" +MODEL = "gpt-image-2" +PNG_B64 = base64.b64encode(b"fake-png-bytes").decode() + +#: 官方文档「Generate image」一节的响应形状。 +OFFICIAL_RESPONSE = { + "created": 1713833628, + "data": [{"b64_json": PNG_B64}], + "size": "1024x1024", + "output_format": "png", + "quality": "high", + "usage": { + "total_tokens": 100, + "input_tokens": 50, + "output_tokens": 50, + "input_tokens_details": {"text_tokens": 10, "image_tokens": 40}, + }, +} + + +def provider_with(handler, *, fetch_handler=None, **kwargs) -> OpenAiImageProvider: + client = OpenAiClient( + api_key=FAKE_API_KEY, + base_url="https://api.openai.invalid/v1", + client=httpx.Client(transport=httpx.MockTransport(handler)), + ) + fetch_client = ( + httpx.Client(transport=httpx.MockTransport(fetch_handler)) + if fetch_handler + else None + ) + return OpenAiImageProvider( + client, + SyncTaskRegistry(max_workers=1, max_entries=8), + input_fetch_client=fetch_client, + **kwargs, + ) + + +def image_request(**kwargs) -> GenerationRequest: + defaults = { + "job_id": "job-1", + "media_kind": MediaKind.IMAGE, + "model": MODEL, + "prompt": "一只柯基在草地上", + } + return GenerationRequest(**{**defaults, **kwargs}) + + +def await_result(provider: OpenAiImageProvider, task_id: str, timeout: float = 3.0): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + result = provider.fetch(task_id) + if result.state.is_terminal: + return result + time.sleep(0.01) + raise AssertionError("任务未在超时内结束") + + +class TestIdentity: + def test_provider_code_matches_the_catalog_column(self): + assert ( + provider_with(lambda r: httpx.Response(200, json={})).provider_code + == PROVIDER_CODE + ) + + def test_declares_itself_as_an_image_provider(self): + assert ( + provider_with(lambda r: httpx.Response(200, json={})).media_kind + is MediaKind.IMAGE + ) + + def test_rejects_a_misrouted_video_request(self): + with pytest.raises(ValueError, match="只处理图片生成"): + provider_with(lambda r: httpx.Response(200, json={})).submit( + image_request(media_kind=MediaKind.VIDEO) + ) + + +class TestSubmitIsNonBlockingDespiteTheSyncVendor: + """整个 tasks 层存在的理由：服务商同步，但我们对外必须是异步的。""" + + def test_submit_returns_a_task_id_without_waiting_for_the_image(self): + import threading + + release = threading.Event() + + def handler(request): + release.wait(timeout=3) + return httpx.Response(200, json=OFFICIAL_RESPONSE) + + provider = provider_with(handler) + begin = time.monotonic() + task_id = provider.submit(image_request()) + elapsed = time.monotonic() - begin + + assert task_id.startswith("local-") + assert elapsed < 0.5 + assert provider.fetch(task_id).state is TaskState.RUNNING + release.set() + await_result(provider, task_id) + + +class TestTextToImage: + def test_posts_json_to_the_generations_endpoint(self): + seen: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + import json + + seen["path"] = request.url.path + seen["body"] = json.loads(request.content) + seen["auth"] = request.headers.get("Authorization") + return httpx.Response(200, json=OFFICIAL_RESPONSE) + + provider = provider_with(handler) + await_result(provider, provider.submit(image_request())) + + assert seen["path"].endswith("/images/generations") + assert seen["auth"] == f"Bearer {FAKE_API_KEY}" + assert seen["body"]["prompt"] == "一只柯基在草地上" + assert seen["body"]["model"] == MODEL + # 一个任务一个产物：generation_jobs.output_asset_id 是单值。 + assert seen["body"]["n"] == 1 + + def test_pixel_size_is_sent_only_when_both_dimensions_are_given(self): + seen: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + import json + + seen.update(json.loads(request.content)) + return httpx.Response(200, json=OFFICIAL_RESPONSE) + + provider = provider_with(handler) + await_result( + provider, provider.submit(image_request(width_px=1536, height_px=864)) + ) + assert seen["size"] == "1536x864" + + def test_half_a_size_is_treated_as_no_size_at_all(self): + # 只给宽不给高时交给模型自选，而不是猜一个高度出来。 + seen: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + import json + + seen.update(json.loads(request.content)) + return httpx.Response(200, json=OFFICIAL_RESPONSE) + + provider = provider_with(handler) + await_result(provider, provider.submit(image_request(width_px=1024))) + assert "size" not in seen + + def test_safety_identifier_is_sent_as_the_vendors_user_field(self): + seen: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + import json + + seen.update(json.loads(request.content)) + return httpx.Response(200, json=OFFICIAL_RESPONSE) + + provider = provider_with(handler) + await_result( + provider, provider.submit(image_request(safety_identifier="abc123hash")) + ) + # 厂商字段名是 user，不是 safety_identifier。 + assert seen["user"] == "abc123hash" + assert "safety_identifier" not in seen + + +class TestModelResolution: + def test_request_model_wins_over_the_configured_default(self): + seen: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + import json + + seen.update(json.loads(request.content)) + return httpx.Response(200, json=OFFICIAL_RESPONSE) + + provider = provider_with(handler, default_model="fallback") + await_result(provider, provider.submit(image_request(model="snapshot-model"))) + assert seen["model"] == "snapshot-model" + + def test_no_model_anywhere_fails_before_any_call(self): + def handler(request): + raise AssertionError("不应发出请求") + + with pytest.raises(ValueError, match="未指定模型"): + provider_with(handler).submit(image_request(model="")) + + +class TestLocalValidationHappensBeforeAccepting: + """违规请求应当立刻被拒，而不是变成一个注定失败的后台任务。 + + 后者要等一轮轮询才被发现，还白占一个线程池槽位。 + """ + + def test_bad_size_is_rejected_synchronously(self): + def handler(request): + raise AssertionError("不应发出请求") + + with pytest.raises(ValueError, match="16 整除"): + provider_with(handler).submit(image_request(width_px=1000, height_px=864)) + + +class TestImageToImage: + def test_downloads_the_input_then_posts_multipart_to_edits(self): + seen: dict = {} + + def openai_handler(request: httpx.Request) -> httpx.Response: + seen["path"] = request.url.path + seen["content_type"] = request.headers.get("Content-Type", "") + seen["raw"] = request.content + return httpx.Response(200, json=OFFICIAL_RESPONSE) + + def fetch_handler(request: httpx.Request) -> httpx.Response: + seen["fetched"] = str(request.url) + return httpx.Response( + 200, + content=b"reference-image-bytes", + headers={"Content-Type": "image/png"}, + ) + + provider = provider_with(openai_handler, fetch_handler=fetch_handler) + await_result( + provider, provider.submit(image_request(input_url="https://x/ref.png")) + ) + + assert seen["fetched"] == "https://x/ref.png" + # 编辑接口走 multipart 而非 JSON——`image` 是文件。 + assert seen["path"].endswith("/images/edits") + assert "multipart/form-data" in seen["content_type"] + assert b"reference-image-bytes" in seen["raw"] + # 字段名统一用 image[]（该接口支持多图且共用此名），避免单图多图两种写法。 + assert b'name="image[]"' in seen["raw"] + + def test_an_undownloadable_input_is_rejected_synchronously(self): + """取不到参考图就是参数问题，应立刻拒绝而不是变成后台失败。""" + + def openai_handler(request): + raise AssertionError("不应发出生成请求") + + def fetch_handler(request): + raise httpx.ConnectError("refused") + + with pytest.raises(ValueError, match="无法下载"): + provider_with(openai_handler, fetch_handler=fetch_handler).submit( + image_request(input_url="https://unreachable/ref.png") + ) + + def test_an_oversized_input_is_rejected_by_its_declared_length(self): + """第一道：声明的 Content-Length 就超了，连读都不必读。""" + + def openai_handler(request): + raise AssertionError("不应发出生成请求") + + def fetch_handler(request): + return httpx.Response(200, content=b"x" * (21 * 1024 * 1024)) + + with pytest.raises(ValueError, match="参考图过大"): + provider_with(openai_handler, fetch_handler=fetch_handler).submit( + image_request(input_url="https://x/huge.png") + ) + + def test_an_oversized_input_is_caught_even_when_the_length_lies(self): + """第二道：不信 Content-Length——它可能缺失或撒谎，必须边读边数。 + + 用分块流式响应（无 Content-Length）构造这种情形：只有累计计数才拦得住。 + """ + + def openai_handler(request): + raise AssertionError("不应发出生成请求") + + def chunks(): + # 每块 1MB，共 21MB，分块传输故无 Content-Length。 + for _ in range(21): + yield b"x" * (1024 * 1024) + + def fetch_handler(request): + return httpx.Response(200, content=chunks()) + + with pytest.raises(ValueError, match="超过上限"): + provider_with(openai_handler, fetch_handler=fetch_handler).submit( + image_request(input_url="https://x/lying.png") + ) + + +class TestResponseMapping: + def test_dimensions_come_straight_from_the_response(self): + """不像火山那边要查对照表——响应直接给 size。""" + provider = provider_with(lambda r: httpx.Response(200, json=OFFICIAL_RESPONSE)) + result = await_result(provider, provider.submit(image_request())) + + assert result.state is TaskState.SUCCEEDED + assert (result.width_px, result.height_px) == (1024, 1024) + assert result.output_format == "png" + assert result.billing_tokens == 100 + assert result.failure is None + + def test_arbitrary_size_is_parsed_too(self): + payload = {**OFFICIAL_RESPONSE, "size": "1536x864"} + provider = provider_with(lambda r: httpx.Response(200, json=payload)) + result = await_result(provider, provider.submit(image_request())) + assert (result.width_px, result.height_px) == (1536, 864) + + def test_asset_url_is_a_data_uri_with_no_expiry(self): + """产物是 base64 字节，不是可访问 URL。 + + 没有「24 小时过期」问题（与火山不同），故 expires_in 为 None 而不是编一个 + 假的秒数。但它**也不是可长期保存的地址**——绝不能原样写进 external_url。 + """ + provider = provider_with(lambda r: httpx.Response(200, json=OFFICIAL_RESPONSE)) + result = await_result(provider, provider.submit(image_request())) + + assert result.asset_url.startswith("data:image/png;base64,") + assert PNG_B64 in result.asset_url + assert result.asset_url_expires_in_sec is None + + def test_a_2xx_without_image_data_is_a_retryable_failure(self): + """服务商说成功却没给图——那不是成功。""" + provider = provider_with( + lambda r: httpx.Response(200, json={"created": 1, "data": []}) + ) + result = await_result(provider, provider.submit(image_request())) + + assert result.state is TaskState.FAILED + assert result.failure.retry is Retry.BACKOFF + assert result.failure.provider_code == "MissingImageData" + + +class TestFailurePropagation: + def test_vendor_rate_limit_becomes_a_retryable_failure(self): + def handler(request): + return httpx.Response( + 429, + json={"error": {"message": "slow down", "type": "rate_limit_exceeded"}}, + ) + + provider = provider_with(handler) + result = await_result(provider, provider.submit(image_request())) + + assert result.state is TaskState.FAILED + assert result.failure.retry is Retry.BACKOFF + assert result.failure.user_facing is UserFacing.RATE_LIMITED + + def test_insufficient_quota_is_not_worth_retrying(self): + def handler(request): + return httpx.Response( + 429, + json={"error": {"message": "no credit", "code": "insufficient_quota"}}, + ) + + provider = provider_with(handler) + result = await_result(provider, provider.submit(image_request())) + + assert result.failure.retry is Retry.QUOTA + assert result.failure.user_facing is UserFacing.QUOTA_EXHAUSTED + + def test_content_policy_violation_is_terminal(self): + def handler(request): + return httpx.Response( + 400, + json={"error": {"message": "blocked", "code": "moderation_blocked"}}, + ) + + provider = provider_with(handler) + result = await_result(provider, provider.submit(image_request())) + + assert result.failure.retry is Retry.NEVER + assert result.failure.user_facing is UserFacing.CONTENT_REJECTED + + +class TestUnknownTaskId: + def test_fetch_raises_key_error_so_the_endpoint_can_map_it(self): + provider = provider_with(lambda r: httpx.Response(200, json={})) + with pytest.raises(KeyError): + provider.fetch("local-never-existed") diff --git a/tests/providers/openai/test_tasks.py b/tests/providers/openai/test_tasks.py new file mode 100644 index 0000000..a93d307 --- /dev/null +++ b/tests/providers/openai/test_tasks.py @@ -0,0 +1,249 @@ +"""同步→异步任务注册表的测试。 + +这一层是 gpt-image-2 与 Seedance 的根本差异所在（同步接口没有任务 ID），也是唯一 +让本服务**持有状态**的地方。它写错的表现是丢任务或内存泄漏，两者都要到生产才显形， +所以逐条钉住。 +""" + +from __future__ import annotations + +import threading +import time + +import pytest + +from app.domain.generation import ( + Failure, + GenerationResult, + Retry, + TaskState, + UserFacing, +) +from app.providers.openai.client import OpenAiError +from app.providers.openai.tasks import LOCAL_TASK_PREFIX, SyncTaskRegistry + + +def succeeded(url: str = "data:image/png;base64,AAA") -> GenerationResult: + return GenerationResult( + provider_request_id="", state=TaskState.SUCCEEDED, progress=100, asset_url=url + ) + + +def wait_until_done(registry: SyncTaskRegistry, task_id: str, timeout: float = 3.0): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + result = registry.fetch(task_id) + if result is not None and result.state.is_terminal: + return result + time.sleep(0.01) + raise AssertionError(f"任务 {task_id} 在 {timeout}s 内未结束") + + +class TestSubmitReturnsImmediately: + def test_submit_does_not_block_on_the_slow_call(self): + """这是整层存在的理由：同步调用必须在后台跑，提交立刻返回。""" + registry = SyncTaskRegistry(max_workers=2) + started = threading.Event() + release = threading.Event() + + def slow(): + started.set() + release.wait(timeout=3) + return succeeded() + + begin = time.monotonic() + task_id = registry.submit(slow) + elapsed = time.monotonic() - begin + + assert started.wait(timeout=2) + # 提交本身几乎不耗时，尽管那个调用还在跑。 + assert elapsed < 0.5 + assert registry.fetch(task_id).state is TaskState.RUNNING + release.set() + wait_until_done(registry, task_id) + registry.shutdown() + + def test_task_id_is_prefixed_so_logs_distinguish_it_from_a_vendor_id(self): + registry = SyncTaskRegistry() + task_id = registry.submit(succeeded) + assert task_id.startswith(LOCAL_TASK_PREFIX) + registry.shutdown() + + def test_ids_are_unique(self): + registry = SyncTaskRegistry() + ids = {registry.submit(succeeded) for _ in range(20)} + assert len(ids) == 20 + registry.shutdown() + + +class TestInFlightAndCompletedStates: + def test_running_reports_zero_progress(self): + # 同步接口不提供进度，只能报 0。契约本就不承诺平滑进度（ADR-023 §10）。 + registry = SyncTaskRegistry() + release = threading.Event() + task_id = registry.submit(lambda: (release.wait(timeout=3), succeeded())[1]) + + result = registry.fetch(task_id) + assert result.state is TaskState.RUNNING + assert result.progress == 0 + release.set() + registry.shutdown() + + def test_completed_result_is_returned_with_the_real_task_id(self): + """工作函数拿不到自己的任务 ID，故由注册表在 fetch 时填入。""" + registry = SyncTaskRegistry() + task_id = registry.submit(succeeded) + + result = wait_until_done(registry, task_id) + assert result.state is TaskState.SUCCEEDED + assert result.provider_request_id == task_id + assert result.asset_url is not None + registry.shutdown() + + def test_result_can_be_fetched_more_than_once(self): + # Java 侧可能重复轮询同一个已完成任务（比如一次响应丢了）。 + registry = SyncTaskRegistry() + task_id = registry.submit(succeeded) + wait_until_done(registry, task_id) + + assert registry.fetch(task_id).state is TaskState.SUCCEEDED + assert registry.fetch(task_id).state is TaskState.SUCCEEDED + registry.shutdown() + + +class TestUnknownIdIsIndistinguishableOnPurpose: + def test_never_existed_returns_none(self): + """从未存在过与「进程重启后丢失」无法区分——不假装区分。 + + 两种情形对调用方的处置相同：停止轮询、任务判失败。 + """ + assert SyncTaskRegistry().fetch("local-nonexistent") is None + + +class TestFailurePropagation: + def test_a_classified_provider_error_keeps_its_classification(self): + registry = SyncTaskRegistry() + failure = Failure( + Retry.QUOTA, UserFacing.QUOTA_EXHAUSTED, "insufficient_quota", "no credit" + ) + + def failing(): + raise OpenAiError(failure, 429) + + task_id = registry.submit(failing) + result = wait_until_done(registry, task_id) + + assert result.state is TaskState.FAILED + assert result.failure.retry is Retry.QUOTA + assert result.failure.provider_code == "insufficient_quota" + registry.shutdown() + + def test_an_unclassified_exception_becomes_a_retryable_failure_naming_its_type( + self, + ): + """未分类异常不该让任务卡住，也不该丢掉线索。 + + 把异常类型名放进 provider_code——否则排查时只有一句空洞的错误消息。 + """ + registry = SyncTaskRegistry() + + def boom(): + raise ZeroDivisionError("unexpected") + + task_id = registry.submit(boom) + result = wait_until_done(registry, task_id) + + assert result.state is TaskState.FAILED + assert result.failure.retry is Retry.BACKOFF + assert result.failure.provider_code == "ZeroDivisionError" + registry.shutdown() + + +class TestCapacityIsBounded: + def test_submitting_beyond_the_cap_is_rejected_rather_than_evicting(self): + """满了要拒绝新提交，**不能挤掉在途任务**。 + + 挤掉一个在途任务会让一个可能正在计费的调用变成无人认领的开销。 + """ + registry = SyncTaskRegistry(max_workers=1, max_entries=2) + release = threading.Event() + + def blocked(): + release.wait(timeout=3) + return succeeded() + + registry.submit(blocked) + registry.submit(blocked) + + with pytest.raises(RuntimeError, match="上限"): + registry.submit(blocked) + + release.set() + registry.shutdown() + + def test_concurrency_is_capped_by_max_workers(self): + """并发上限同时也是「同时有多少钱在花」的上界。""" + registry = SyncTaskRegistry(max_workers=2, max_entries=10) + concurrent = [] + lock = threading.Lock() + release = threading.Event() + + def work(): + with lock: + concurrent.append(1) + peak = len(concurrent) + release.wait(timeout=3) + with lock: + concurrent.pop() + return succeeded(f"data:image/png;base64,{peak}") + + for _ in range(5): + registry.submit(work) + time.sleep(0.2) + + with lock: + assert len(concurrent) <= 2 + + release.set() + registry.shutdown() + + +class TestExpiryOnlyTouchesCompletedEntries: + def test_completed_entries_expire(self): + registry = SyncTaskRegistry(completed_ttl_sec=0.05, max_entries=10) + task_id = registry.submit(succeeded) + wait_until_done(registry, task_id) + time.sleep(0.1) + + # 清理发生在下一次 submit 时（不额外起一个定时线程——本服务的规模用不上）。 + registry.submit(succeeded) + assert registry.fetch(task_id) is None + registry.shutdown() + + def test_in_flight_entries_are_never_expired(self): + """在途任务无论多久都不清——它可能正在花钱。""" + registry = SyncTaskRegistry(completed_ttl_sec=0.01, max_entries=10) + release = threading.Event() + task_id = registry.submit(lambda: (release.wait(timeout=3), succeeded())[1]) + time.sleep(0.1) + + registry.submit(succeeded) + # 仍在注册表里。 + assert registry.fetch(task_id) is not None + assert registry.fetch(task_id).state is TaskState.RUNNING + + release.set() + registry.shutdown() + + def test_ttl_starts_at_completion_not_at_creation(self): + """否则一个耗时接近 TTL 的任务会在刚完成时就被清掉——结果还没被取走。""" + registry = SyncTaskRegistry(completed_ttl_sec=5.0, max_entries=10) + release = threading.Event() + task_id = registry.submit(lambda: (release.wait(timeout=3), succeeded())[1]) + time.sleep(0.15) + release.set() + wait_until_done(registry, task_id) + + registry.submit(succeeded) + assert registry.fetch(task_id).state is TaskState.SUCCEEDED + registry.shutdown()