diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/providers/__init__.py b/app/providers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/providers/errors.py b/app/providers/errors.py new file mode 100644 index 0000000..872313e --- /dev/null +++ b/app/providers/errors.py @@ -0,0 +1,198 @@ +"""火山方舟(Volcengine Ark)错误分类。 + +把厂商的错误码翻译成三个对 Worker 有意义的维度:**是否可重试**、**等多久**、 +**给用户看什么**。分类依据是官方错误码表(见 patbond-doc +`docs/development/providers/volcengine-seedance.md` 的取证记录)。 + +为什么值得单独成一个模块而不是散在客户端里: + +1. **`QuotaExceeded` 一码两义。** 同一个 code 既表示「免费额度/周期限额耗尽」 + (要等到明天或去充值),也表示「当前账号排队中任务数超限」(等几秒就好)。 + 官方文档里这两条是相邻的两行、code 完全相同,只有 message 不同。若按 code + 一刀切,必然有一半场景的文案和出口是错的。 +2. **`expired` 状态没写在查询接口的状态枚举里。** 查询文档列的是 queued / + running / cancelled / succeeded / failed 五个,但回调文档额外列了 `expired` + (任务超过 `execution_expires_after` 被自动终止)。只按查询文档实现,线上 + 遇到 expired 会掉进未知分支。 +3. 敏感内容类错误**重试毫无意义**(同样的输入必然同样被拒),必须与限流类 + 区分开,否则会白烧三次重试额度再失败。 +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from enum import Enum + + +class Retry(Enum): + """重试策略。""" + + #: 不可重试:同样的输入必然得到同样的结果(参数错、内容被拒、鉴权失败)。 + NEVER = "never" + #: 可重试且应尽快:限流、突发流量保护、服务过载。退避几秒到几十秒。 + BACKOFF = "backoff" + #: 可重试但要等很久:额度/配额耗尽,通常要等到下一个计费周期或充值之后。 + #: Worker 不应把重试额度浪费在这上面,应直接终态失败并让用户看到出口。 + QUOTA = "quota" + + +class UserFacing(Enum): + """给终端用户的呈现类别,对应客户端的两个 429 分支(ADR-023 §9)。""" + + #: 配额耗尽:等的是「明天」或「充值」。客户端弹 sheet 并给出出口。 + QUOTA_EXHAUSTED = "quota_exhausted" + #: 频率过快:等的是「几秒」。客户端弹 SnackBar,可自动重试。 + RATE_LIMITED = "rate_limited" + #: 内容被拒:换提示词或换图。不可重试。 + CONTENT_REJECTED = "content_rejected" + #: 服务端问题(我们的 bug 或配置错误)。用户无从处置,只报通用失败。 + SERVER_ERROR = "server_error" + #: 上游暂时不可用。可重试。 + UPSTREAM_UNAVAILABLE = "upstream_unavailable" + + +@dataclass(frozen=True) +class Classification: + retry: Retry + user_facing: UserFacing + #: 原始厂商错误码,原样透传便于排查——绝不丢弃。 + provider_code: str + + +#: 内容安全类:**重试无意义**。覆盖输入/输出的文本、图片、视频、音频,以及 +#: 版权、真人、伪造证件等子类,另含风险识别产品(*RiskDetection)的四个码。 +#: 用前缀匹配而非穷举:官方这一族的码在持续增加(Input/Output × 媒体类型 × +#: 子原因),穷举必然漏,而这一族的处置方式完全一致。 +_CONTENT_PREFIXES = ( + "SensitiveContentDetected", + "InputTextSensitiveContentDetected", + "InputImageSensitiveContentDetected", + "InputVideoSensitiveContentDetected", + "InputAudioSensitiveContentDetected", + "OutputTextSensitiveContentDetected", + "OutputImageSensitiveContentDetected", + "OutputVideoSensitiveContentDetected", + "OutputAudioSensitiveContentDetected", + "InputTextRiskDetection", + "InputImageRiskDetection", + "OutputTextRiskDetection", + "OutputImageRiskDetection", +) + +#: 限流类:退避后重试。等的是秒级。 +_RATE_LIMIT_CODES = frozenset( + { + "RateLimitExceeded.EndpointRPMExceeded", + "RateLimitExceeded.EndpointTPMExceeded", + "RateLimitExceeded.EndpointFlexTPMExceeded", + "ModelAccountRpmRateLimitExceeded", + "ModelAccountTpmRateLimitExceeded", + "ModelAccountFlexTpmRateLimitExceeded", + "ModelAccountIpmRateLimitExceeded", + "APIAccountRpmRateLimitExceeded", + "AccountRateLimitExceeded", + "ServerOverloaded", + "RequestBurstTooFast", + } +) + +#: 配额/计费类:等的是「明天」或「充值」。重试解决不了。 +#: 注意 403 的欠费与未开通也归此类——HTTP 状态码不同,但用户要做的事一样。 +_QUOTA_CODES = frozenset( + { + "SetLimitExceeded", + "InflightBatchsizeExceeded", + "QuotaExceeded.AgentPlanQuotaExceeded", + "QuotaExceeded.DoubaoSearchFreeQuotaExceeded", + "AccountOverdueError", + "OperationDenied.ServiceOverdue", + "OperationDenied.FileQuotaExceeded", + "ModelNotOpen", + "OperationDenied.ServiceNotOpen", + } +) + +#: 我方配置或代码错误:重试无意义,需要人来改。 +_SERVER_FAULT_CODES = frozenset( + { + "AuthenticationError", + "InvalidAccountStatus", + "AccessDenied", + "InvalidEndpointOrModel.NotFound", + "InvalidEndpointOrModel.ModelIDAccessDisabled", + "InvalidEndpoint.ClosedEndpoint", + "OperationDenied.PermissionDenied", + "OperationDenied.ArkAccessRoleNotFound", + "OperationDenied.TosAccessDenied", + "PathNotFound", + } +) + +#: 上游内部错误:可重试。 +_UPSTREAM_CODES = frozenset({"InternalServiceError", "UpstreamUnavailable"}) + +#: `QuotaExceeded` 一码两义的判别式。命中即为「排队任务数超限」,属秒级等待; +#: 否则按额度耗尽处理。取自官方两行 message 的区别性措辞。 +#: 判别放在 message 上是不得已——code 相同,没有别的字段可依据。 +_QUEUE_BACKLOG_MESSAGE = re.compile( + r"exceeded the quota|排队中状态的任务数", re.IGNORECASE +) + + +def classify(provider_code: str | None, message: str | None = None) -> Classification: + """把厂商错误码分类。 + + 未知码一律按**可退避重试的上游问题**处理而不是直接判死:厂商在持续新增错误 + 码,把未知码当终态失败会让一次可恢复的抖动变成用户看到的失败;而重试预算 + 本身有上限(`max_attempts`),最坏情况只是多试两次。 + """ + code = (provider_code or "").strip() + text = message or "" + + if not code: + return Classification(Retry.BACKOFF, UserFacing.UPSTREAM_UNAVAILABLE, "") + + if code.startswith(_CONTENT_PREFIXES): + return Classification(Retry.NEVER, UserFacing.CONTENT_REJECTED, code) + + if code == "ContentSecurityDetectionError": + # 风险识别产品自身请求失败,不是内容真的违规 —— 可重试。 + return Classification(Retry.BACKOFF, UserFacing.UPSTREAM_UNAVAILABLE, code) + + if code == "QuotaExceeded": + # 一码两义,见模块 docstring。 + if _QUEUE_BACKLOG_MESSAGE.search(text): + return Classification(Retry.BACKOFF, UserFacing.RATE_LIMITED, code) + return Classification(Retry.QUOTA, UserFacing.QUOTA_EXHAUSTED, code) + + if code in _RATE_LIMIT_CODES: + return Classification(Retry.BACKOFF, UserFacing.RATE_LIMITED, code) + + if code in _QUOTA_CODES: + return Classification(Retry.QUOTA, UserFacing.QUOTA_EXHAUSTED, code) + + if code in _SERVER_FAULT_CODES: + return Classification(Retry.NEVER, UserFacing.SERVER_ERROR, code) + + if code in _UPSTREAM_CODES: + return Classification(Retry.BACKOFF, UserFacing.UPSTREAM_UNAVAILABLE, code) + + # 参数类:我方拼错了请求体,重试同样的请求体不会变好。 + # 含 InvalidParameter.TaskTypeConstraint / TaskTypeMismatch —— 那两个是 + # Seedance 2.5 全模态任务的参数与模型判定的任务类型不兼容,属我方参数问题。 + if code.startswith( + ( + "InvalidParameter", + "MissingParameter", + "InvalidArgumentError", + "InvalidImageURL", + "InvalidPayload", + "RequestTooLarge", + "RequestBodyTooLarge", + "OutofContextError", + ) + ): + return Classification(Retry.NEVER, UserFacing.SERVER_ERROR, code) + + return Classification(Retry.BACKOFF, UserFacing.UPSTREAM_UNAVAILABLE, code) diff --git a/app/providers/seedance.py b/app/providers/seedance.py new file mode 100644 index 0000000..e6282cd --- /dev/null +++ b/app/providers/seedance.py @@ -0,0 +1,370 @@ +"""火山方舟 Seedance 视频生成客户端。 + +只做两件事:**提交任务**(`POST /contents/generations/tasks`)与**查询任务** +(`GET /contents/generations/tasks/{id}`)。产物的转存不在这里——见 +`SeedanceTask.video_url` 的说明。 + +## 厂商接口的五个关键事实(逐条取自官方文档,详见 patbond-doc +`docs/development/providers/volcengine-seedance.md`) + +1. **没有幂等键。** 请求参数里没有任何幂等相关字段(有的是 model / content / + ratio / duration / generate_audio / watermark / output_format / + return_last_frame / callback_url / execution_expires_after / priority / + safety_identifier)。因此「提交超时后不知道有没有扣费」这个窗口**无法靠厂商 + 的幂等头消除**,只能由 `max_attempts` 兜住。这与原先的乐观假设不同,已如实 + 登记而非假装解决。 + 缓解手段有一个:本接口是**异步**的(提交只入队,立即返回任务 ID),所以提交 + 本身很快、超时概率低;且厂商支持 `callback_url`,一个提交超时但实际已创建的 + 任务会通过回调把 ID 送回来,可作为孤儿任务的回收通道。 +2. **提交是异步的**:返回 `{"id": ...}`,随后靠查询接口拿状态。这正好与 + 「Java↔Python 边界一律异步」的既定设计吻合,无需在本服务内部做后台任务承载。 +3. **产物是有效期 24 小时的 URL**(Seedance 2.5 另有 100 次下载上限)。 + **因此产物必须转存**——把厂商 URL 直接存进 `media.assets.external_url` + 会在 24 小时后变成死链。这一条对「本地存储还是云存储」的决策是硬约束: + 无论选哪个,都必须在 24 小时内把文件复制到我们自己的存储。 +4. **任务记录只保留 7 天**,`execution_expires_after` 取值范围 `[3600, 259200]` + 秒(默认 48 小时)。超时任务被标记为 `expired`。 +5. **限流与配额是两族不同的 429**,映射见 `errors.py`。 + +## 一个文档自身的矛盾,必须按宽的那份实现 + +查询接口文档列的状态是 queued / running / cancelled / succeeded / failed 五个, +但创建接口的 `callback_url` 一节额外列了 **`expired`**。只按查询文档实现,线上 +遇到超时任务会掉进未知分支。这里按六个状态实现,并把 `expired` 当作**可重试的 +失败**(是超时而非内容问题)。 +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from enum import Enum +from typing import Any + +import httpx + +from .errors import Classification, Retry, UserFacing, classify + +log = logging.getLogger(__name__) + +#: 官方 base URL(北京区)。 +DEFAULT_BASE_URL = "https://ark.cn-beijing.volces.com/api/v3" + +_TASKS_PATH = "/contents/generations/tasks" + + +class TaskStatus(Enum): + """厂商任务状态。 + + 六个取值:查询文档给了五个,`expired` 来自创建文档的回调说明一节。 + """ + + QUEUED = "queued" + RUNNING = "running" + SUCCEEDED = "succeeded" + FAILED = "failed" + CANCELLED = "cancelled" + #: 任务在排队或运行中超过 `execution_expires_after` 被自动终止。 + #: **查询接口文档未列出此值**,仅见于创建接口的回调状态说明。 + EXPIRED = "expired" + + @classmethod + def parse(cls, value: str | None) -> TaskStatus | None: + """未知状态返回 None 而不是抛异常。 + + 厂商新增状态时,抛异常会让所有在途任务一起失败;返回 None 让调用方能把 + 它当成「还没结束」继续轮询,最坏是等到我们自己的超时。 + """ + for status in cls: + if status.value == value: + return status + return None + + @property + def is_terminal(self) -> bool: + return self in { + TaskStatus.SUCCEEDED, + TaskStatus.FAILED, + TaskStatus.CANCELLED, + TaskStatus.EXPIRED, + } + + +@dataclass(frozen=True) +class SeedanceTask: + """查询接口返回的任务。 + + :param video_url: 生成视频的 URL,**有效期仅 24 小时**。本服务不负责转存, + 但调用方必须在有效期内完成转存——把它长期存为 `external_url` 会变死链。 + :param last_frame_url: 尾帧图像(png,尺寸与视频一致,无水印),仅在提交时 + 设置 `return_last_frame=true` 才返回。可直接用作**视频封面图**,省掉 + 我们自己抽帧。 + :param duration_sec: 厂商返回的时长**是向下取整的约数**:官方给的公式是 + `duration = 实际总帧数 / 24` 向下取整,例如 133 帧实际 5.54 秒会返回 5。 + 因此据此换算出的 `duration_ms` 最多可能短 1 秒左右,不可当精确值用。 + :param frames: 与 `duration_sec` **只会返回其中一个**(提交时指定了 frames + 才返回 frames)。两者都在时以 frames 为准。 + """ + + id: str + status: TaskStatus | None + raw_status: str | None + video_url: str | None = None + last_frame_url: str | None = None + duration_sec: int | None = None + frames: int | None = None + fps: int | None = None + ratio: str | None = None + resolution: str | None = None + output_format: str | None = None + error_code: str | None = None + error_message: str | None = None + completion_tokens: int | None = None + created_at: int | None = None + updated_at: int | None = None + + @property + def duration_ms(self) -> int | None: + """尽可能精确的时长毫秒数。 + + 有 `frames` + `fps` 时按帧算(精确);否则退回厂商返回的整数秒 + (**向下取整的约数**,见类 docstring)。两者都缺则返回 None,而不是 + 编一个 0 出来。 + """ + if self.frames is not None and self.fps: + return round(self.frames / self.fps * 1000) + if self.duration_sec is not None: + return self.duration_sec * 1000 + return None + + @property + def failure(self) -> Classification | None: + """失败任务的分类;非失败返回 None。 + + `expired` 单独处理:它是超时,不是内容或参数问题,因此可重试——而厂商 + 在 expired 时不一定给 `error.code`,若走通用分类会被当成「无码」处理。 + """ + if self.status is TaskStatus.EXPIRED: + return Classification( + Retry.BACKOFF, UserFacing.UPSTREAM_UNAVAILABLE, "TaskExpired" + ) + if self.status is TaskStatus.FAILED: + return classify(self.error_code, self.error_message) + return None + + +@dataclass +class SeedanceSubmission: + """一次提交所需的全部参数。 + + 刻意不把厂商的全部参数都暴露出来——只暴露 Patbond 会用到的。多暴露一个参数 + 就多一处需要在契约、客户端、文档三处同步维护的地方。 + """ + + model: str + prompt: str | None = None + #: 首帧图片 URL。给了即为图生视频,不给即为文生视频(参考图选填,决策 A2)。 + first_frame_url: str | None = None + #: 宽高比。`adaptive` 让模型自适应;图生视频时 Seedance 2.5 只支持 adaptive。 + ratio: str = "adaptive" + resolution: str = "720p" + #: 时长(秒)。`-1` 交给模型在有效范围内自选。 + duration: int = 5 + generate_audio: bool = True + watermark: bool = False + #: 返回尾帧图,用作视频封面。默认开启——成本为零,省掉我们自己抽帧。 + return_last_frame: bool = True + #: 任务超时阈值(秒)。取值范围 [3600, 259200]。**默认 48 小时对交互式应用 + #: 过长**,故这里取下限 1 小时,避免被放弃的任务在厂商侧挂两天。 + execution_expires_after: int = 3600 + #: 终端用户哈希标识,助厂商做滥用检测。**必须是哈希值,不可传明文用户 ID**。 + safety_identifier: str | None = None + callback_url: str | None = None + + def to_body(self) -> dict[str, Any]: + content: list[dict[str, Any]] = [] + if self.prompt: + content.append({"type": "text", "text": self.prompt}) + if self.first_frame_url: + # role 省略时厂商按 first_frame 处理;显式写出以免依赖默认行为。 + content.append( + { + "type": "image_url", + "image_url": {"url": self.first_frame_url}, + "role": "first_frame", + } + ) + if not content: + raise ValueError("提交内容为空:prompt 与 first_frame_url 至少要有一个") + + body: dict[str, Any] = { + "model": self.model, + "content": content, + "ratio": self.ratio, + "resolution": self.resolution, + "duration": self.duration, + "generate_audio": self.generate_audio, + "watermark": self.watermark, + "return_last_frame": self.return_last_frame, + "execution_expires_after": self.execution_expires_after, + } + if self.safety_identifier: + body["safety_identifier"] = self.safety_identifier + if self.callback_url: + body["callback_url"] = self.callback_url + return body + + +class SeedanceError(RuntimeError): + """厂商返回的错误,带分类。""" + + def __init__( + self, + classification: Classification, + message: str, + http_status: int | None = None, + ): + super().__init__(message) + self.classification = classification + self.http_status = http_status + + +class SeedanceClient: + """Seedance 的同步 HTTP 客户端。 + + **超时设置的含义**:这里的 timeout 是「提交/查询这一次 HTTP 往返」的超时, + **不是生成耗时**。提交接口是异步的(只入队),所以几秒足够;把它调大成 + 「等出图」是错的,会重新引入长连接占用与计费歧义。 + """ + + def __init__( + self, + api_key: str, + base_url: str = DEFAULT_BASE_URL, + timeout: float = 10.0, + client: httpx.Client | None = None, + ): + if not api_key: + raise ValueError("ARK_API_KEY 未配置") + self._api_key = api_key + self._base_url = base_url.rstrip("/") + self._client = client or httpx.Client(timeout=timeout) + + def submit(self, submission: SeedanceSubmission) -> str: + """提交任务,返回厂商任务 ID。 + + 调用方必须在**上游产生任何计费动作之前**把这个 ID 落库——那是超时后 + 唯一能判断「是否已提交」的依据。本接口异步,返回即代表已入队。 + """ + response = self._post(_TASKS_PATH, submission.to_body()) + task_id = response.get("id") + if not task_id: + raise SeedanceError( + classify(None), + f"提交成功但响应缺少任务 ID:{response!r}", + ) + return str(task_id) + + def get(self, task_id: str) -> SeedanceTask: + """查询任务。 + + 注意厂商只保留最近 7 天的任务记录;更早的任务查询会 404。 + """ + payload = self._get(f"{_TASKS_PATH}/{task_id}") + return self._parse_task(payload) + + # ---------------------------------------------------------------- 内部 + + def _headers(self) -> dict[str, str]: + return { + "Authorization": f"Bearer {self._api_key}", + "Content-Type": "application/json", + } + + def _post(self, path: str, body: dict[str, Any]) -> dict[str, Any]: + try: + response = self._client.post( + f"{self._base_url}{path}", json=body, headers=self._headers() + ) + except httpx.HTTPError as exc: + # 网络层失败:可重试。注意提交超时属于「不知道有没有扣费」的窗口, + # 调用方须依赖已落库的 provider_request_id 或回调来判断,不可盲目重发。 + raise SeedanceError(classify(None), f"请求火山方舟失败:{exc}") from exc + return self._unwrap(response) + + def _get(self, path: str) -> dict[str, Any]: + try: + response = self._client.get( + f"{self._base_url}{path}", headers=self._headers() + ) + except httpx.HTTPError as exc: + raise SeedanceError(classify(None), f"查询火山方舟失败:{exc}") from exc + return self._unwrap(response) + + @staticmethod + def _unwrap(response: httpx.Response) -> dict[str, Any]: + """把 HTTP 响应转成 dict,非 2xx 转成带分类的 SeedanceError。 + + 错误体形态取自官方错误码表:`{"error": {"code": ..., "message": ...}}`。 + 响应体解析失败时不静默——那通常意味着我们打到了错误的地址(比如把 + 网关的 HTML 错误页当成了 JSON)。 + """ + if response.is_success: + try: + return response.json() + except ValueError as exc: + raise SeedanceError( + classify(None), + f"响应不是合法 JSON(HTTP {response.status_code}):{response.text[:200]}", + response.status_code, + ) from exc + + code: str | None = None + message = response.text[:500] + try: + error = response.json().get("error") or {} + code = error.get("code") + message = error.get("message") or message + except ValueError: + pass + + classification = classify(code, message) + log.warning( + "火山方舟返回错误 http=%s code=%s retry=%s", + response.status_code, + code, + classification.retry.value, + ) + raise SeedanceError(classification, message, response.status_code) + + @staticmethod + def _parse_task(payload: dict[str, Any]) -> SeedanceTask: + content = payload.get("content") or {} + error = payload.get("error") or {} + usage = payload.get("usage") or {} + raw_status = payload.get("status") + status = TaskStatus.parse(raw_status) + if status is None and raw_status is not None: + # 不抛异常:新状态让调用方按「未结束」继续轮询,最坏等到自己的超时。 + log.warning("火山方舟返回未知任务状态:%r", raw_status) + return SeedanceTask( + id=str(payload.get("id") or ""), + status=status, + raw_status=raw_status, + video_url=content.get("video_url"), + last_frame_url=content.get("last_frame_url"), + duration_sec=payload.get("duration"), + frames=payload.get("frames"), + fps=payload.get("framespersecond"), + ratio=payload.get("ratio"), + resolution=payload.get("resolution"), + output_format=payload.get("output_format"), + error_code=error.get("code"), + error_message=error.get("message"), + completion_tokens=usage.get("completion_tokens"), + created_at=payload.get("created_at"), + updated_at=payload.get("updated_at"), + ) + + def close(self) -> None: + self._client.close() diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..2e437d2 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,4 @@ +-r requirements.txt +# 与 CI 门禁同一套工具(.gitea/workflows/ci.yml) +ruff>=0.6 +pytest>=8.0 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..33bb5e0 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +# 运行时依赖。版本下限按实测可用版本给,不锁死上限——本服务只用 httpx 的 +# 基础同步 API,其小版本变更风险低。 +httpx>=0.27 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_errors.py b/tests/test_errors.py new file mode 100644 index 0000000..da07709 --- /dev/null +++ b/tests/test_errors.py @@ -0,0 +1,175 @@ +"""火山方舟错误分类的测试。 + +每个用例对应官方错误码表里的一行或一类。重点在三处**如果搞错只会在生产暴露** +的地方:`QuotaExceeded` 的一码两义、内容安全类必须不可重试、未知码必须偏向 +可重试而不是直接判死。 +""" + +from app.providers.errors import Retry, UserFacing, classify + + +class TestQuotaExceededIsTwoDifferentThings: + """同一个 code,两种完全不同的处置。只看 code 必然错一半。""" + + def test_queue_backlog_is_a_seconds_long_wait(self): + # 官方原文:「当前账号处于排队中状态的任务数已超过限制,请稍后重试。」 + result = classify( + "QuotaExceeded", "The request has exceeded the quota. Request ID: x." + ) + assert result.retry is Retry.BACKOFF + assert result.user_facing is UserFacing.RATE_LIMITED + + def test_queue_backlog_recognised_from_chinese_message(self): + result = classify("QuotaExceeded", "当前账号处于排队中状态的任务数已超过限制") + assert result.retry is Retry.BACKOFF + assert result.user_facing is UserFacing.RATE_LIMITED + + def test_free_trial_exhausted_needs_billing_action(self): + # 官方原文:「免费试用额度已消耗完毕,请前往控制台开通对应模型服务。」 + result = classify( + "QuotaExceeded", + "Your account [x] has exhausted its free trial quota for the [y] model.", + ) + assert result.retry is Retry.QUOTA + assert result.user_facing is UserFacing.QUOTA_EXHAUSTED + + def test_periodic_limit_exhausted_needs_to_wait_for_reset(self): + result = classify( + "QuotaExceeded", + "You have exceeded the 5-hour/weekly/monthly usage quota. It will reset at T.", + ) + assert result.retry is Retry.QUOTA + assert result.user_facing is UserFacing.QUOTA_EXHAUSTED + + +class TestContentRejectionMustNotBeRetried: + """同样的输入必然同样被拒。重试只会白烧额度再失败。""" + + def test_input_image_sensitive(self): + result = classify("InputImageSensitiveContentDetected") + assert result.retry is Retry.NEVER + assert result.user_facing is UserFacing.CONTENT_REJECTED + + def test_output_video_sensitive(self): + result = classify("OutputVideoSensitiveContentDetected") + assert result.retry is Retry.NEVER + assert result.user_facing is UserFacing.CONTENT_REJECTED + + def test_policy_violation_subcode(self): + result = classify("InputVideoSensitiveContentDetected.PolicyViolation") + assert result.retry is Retry.NEVER + + def test_real_person_subcode(self): + result = classify("InputImageSensitiveContentDetected.PrivacyInformation") + assert result.retry is Retry.NEVER + + def test_risk_detection_product_codes(self): + assert classify("InputTextRiskDetection").retry is Retry.NEVER + assert classify("OutputImageRiskDetection").retry is Retry.NEVER + + def test_but_the_detector_itself_failing_is_retryable(self): + # 风险识别产品请求失败 ≠ 内容真的违规。 + result = classify("ContentSecurityDetectionError", "Internal error.") + assert result.retry is Retry.BACKOFF + assert result.user_facing is UserFacing.UPSTREAM_UNAVAILABLE + + +class TestRateLimitFamily: + def test_endpoint_rpm(self): + result = classify("RateLimitExceeded.EndpointRPMExceeded") + assert result.retry is Retry.BACKOFF + assert result.user_facing is UserFacing.RATE_LIMITED + + def test_model_account_tpm(self): + assert classify("ModelAccountTpmRateLimitExceeded").retry is Retry.BACKOFF + + def test_server_overloaded(self): + assert classify("ServerOverloaded").retry is Retry.BACKOFF + + def test_request_burst_too_fast(self): + # doubao-seed-2.0 之后版本的突增流量保护码。 + assert classify("RequestBurstTooFast").retry is Retry.BACKOFF + + def test_images_per_minute(self): + assert classify("ModelAccountIpmRateLimitExceeded").retry is Retry.BACKOFF + + +class TestQuotaAndBillingFamily: + def test_inference_limit_set_by_account(self): + result = classify("SetLimitExceeded") + assert result.retry is Retry.QUOTA + assert result.user_facing is UserFacing.QUOTA_EXHAUSTED + + def test_concurrency_cap_needs_topup(self): + assert classify("InflightBatchsizeExceeded").retry is Retry.QUOTA + + def test_account_overdue_is_403_but_same_user_action(self): + # HTTP 状态码不同(403 而非 429),但用户要做的事一样:去充值。 + result = classify("AccountOverdueError") + assert result.retry is Retry.QUOTA + assert result.user_facing is UserFacing.QUOTA_EXHAUSTED + + def test_model_not_activated(self): + assert classify("ModelNotOpen").retry is Retry.QUOTA + + +class TestOurOwnFaults: + """重试无意义,需要人改代码或改配置。""" + + def test_bad_api_key(self): + result = classify("AuthenticationError") + assert result.retry is Retry.NEVER + assert result.user_facing is UserFacing.SERVER_ERROR + + def test_unknown_model_id(self): + assert classify("InvalidEndpointOrModel.NotFound").retry is Retry.NEVER + + def test_invalid_parameter(self): + assert classify("InvalidParameter").retry is Retry.NEVER + + def test_task_type_constraint_is_our_parameter_problem(self): + # Seedance 2.5 全模态任务的参数与模型判定的任务类型不兼容。 + assert classify("InvalidParameter.TaskTypeConstraint").retry is Retry.NEVER + + def test_task_type_mismatch(self): + assert classify("InvalidParameter.TaskTypeMismatch").retry is Retry.NEVER + + def test_missing_parameter(self): + assert classify("MissingParameter").retry is Retry.NEVER + + def test_payload_too_large(self): + assert classify("RequestTooLarge").retry is Retry.NEVER + + +class TestUpstreamErrors: + def test_internal_service_error_is_retryable(self): + result = classify("InternalServiceError") + assert result.retry is Retry.BACKOFF + assert result.user_facing is UserFacing.UPSTREAM_UNAVAILABLE + + +class TestUnknownCodesLeanTowardsRetry: + """厂商在持续新增错误码。把未知码判死会让一次抖动变成用户看到的失败; + 重试预算本身有上限,最坏只是多试两次。""" + + def test_unknown_code(self): + result = classify("SomeBrandNewCodeVolcengineAddedLastWeek") + assert result.retry is Retry.BACKOFF + assert result.user_facing is UserFacing.UPSTREAM_UNAVAILABLE + + def test_missing_code(self): + assert classify(None).retry is Retry.BACKOFF + assert classify("").retry is Retry.BACKOFF + + +class TestProviderCodeIsAlwaysPreserved: + """原始码绝不丢弃——排查时它是唯一线索。""" + + def test_known_code(self): + assert classify("ServerOverloaded").provider_code == "ServerOverloaded" + + def test_unknown_code(self): + assert classify("WeirdCode").provider_code == "WeirdCode" + + def test_whitespace_is_trimmed(self): + assert classify(" ServerOverloaded ").provider_code == "ServerOverloaded" diff --git a/tests/test_seedance.py b/tests/test_seedance.py new file mode 100644 index 0000000..c88cf69 --- /dev/null +++ b/tests/test_seedance.py @@ -0,0 +1,341 @@ +"""Seedance 客户端的测试。全部离线——用 httpx 的 MockTransport 拦请求,不打真实厂商。 + +夹具里的响应体逐字取自官方文档的「返回格式」示例,避免测试通过而真实字段名对不上。 +""" + +from __future__ import annotations + +import httpx +import pytest + +from app.providers.errors import Retry, UserFacing +from app.providers.seedance import ( + SeedanceClient, + SeedanceError, + SeedanceSubmission, + TaskStatus, +) + +#: 官方查询文档「返回格式」一节的原样示例。 +OFFICIAL_SUCCESS_PAYLOAD = { + "id": "cgt-2026****-****", + "model": "doubao-seedance-2-0-260128", + "status": "succeeded", + "content": { + "video_url": "https://ark-content-generation-cn-beijing.tos-cn-beijing.volces.com/xxx" + }, + "usage": {"completion_tokens": 108900, "total_tokens": 108900}, + "created_at": 1779348818, + "updated_at": 1779348874, + "seed": 78674, + "resolution": "720p", + "ratio": "16:9", + "duration": 5, + "framespersecond": 24, + "service_tier": "default", + "execution_expires_after": 172800, + "generate_audio": True, + "draft": False, + "priority": 0, +} + + +#: 显式的假值。防泄漏扫描会拦下 `api_key = "..."` 形态的赋值(规则正确——这正是 +#: 真实密钥最常见的泄漏形态),故测试夹具用 ALLOW 清单认可的占位词而非随手起名。 +FAKE_API_KEY = "dummy-api-key-for-tests" + + +def client_with(handler) -> SeedanceClient: + return SeedanceClient( + api_key=FAKE_API_KEY, + client=httpx.Client(transport=httpx.MockTransport(handler)), + ) + + +class TestSubmissionBody: + """请求体必须符合官方 content 数组的形状。""" + + def test_text_to_video_sends_only_text(self): + body = SeedanceSubmission(model="m", prompt="一只猫").to_body() + assert body["content"] == [{"type": "text", "text": "一只猫"}] + assert body["model"] == "m" + + def test_image_to_video_adds_first_frame_with_explicit_role(self): + body = SeedanceSubmission( + model="m", prompt="动起来", first_frame_url="https://x/a.png" + ).to_body() + assert body["content"][0]["type"] == "text" + image = body["content"][1] + assert image["type"] == "image_url" + assert image["image_url"] == {"url": "https://x/a.png"} + # role 省略时厂商按 first_frame 处理,但显式写出以免依赖默认行为。 + assert image["role"] == "first_frame" + + def test_image_only_without_prompt_is_allowed(self): + body = SeedanceSubmission( + model="m", first_frame_url="https://x/a.png" + ).to_body() + assert len(body["content"]) == 1 + assert body["content"][0]["type"] == "image_url" + + def test_empty_content_is_rejected_before_hitting_the_vendor(self): + # 提交空内容必然被厂商拒,但那要花一次往返。本地先拦。 + with pytest.raises(ValueError, match="至少要有一个"): + SeedanceSubmission(model="m").to_body() + + def test_expiry_defaults_to_the_minimum_not_the_vendor_default(self): + # 厂商默认 48 小时,对交互式应用过长——被放弃的任务会在厂商侧挂两天。 + # 取值范围 [3600, 259200],故取下限。 + body = SeedanceSubmission(model="m", prompt="x").to_body() + assert body["execution_expires_after"] == 3600 + + def test_last_frame_requested_by_default_for_free_cover_image(self): + body = SeedanceSubmission(model="m", prompt="x").to_body() + assert body["return_last_frame"] is True + + def test_optional_fields_are_omitted_when_unset(self): + body = SeedanceSubmission(model="m", prompt="x").to_body() + assert "safety_identifier" not in body + assert "callback_url" not in body + + def test_safety_identifier_is_passed_through_when_given(self): + body = SeedanceSubmission( + model="m", prompt="x", safety_identifier="sha256hash" + ).to_body() + assert body["safety_identifier"] == "sha256hash" + + +class TestSubmit: + def test_returns_vendor_task_id(self): + def handler(request: httpx.Request) -> httpx.Response: + assert request.headers["Authorization"] == f"Bearer {FAKE_API_KEY}" + assert request.url.path.endswith("/contents/generations/tasks") + return httpx.Response(200, json={"id": "cgt-123"}) + + assert ( + client_with(handler).submit(SeedanceSubmission(model="m", prompt="x")) + == "cgt-123" + ) + + def test_missing_id_in_success_response_is_an_error(self): + # 2xx 但没有 ID 意味着我们无法再查询这个任务——不能当成功放过。 + def handler(request): + return httpx.Response(200, json={"unexpected": True}) + + with pytest.raises(SeedanceError, match="缺少任务 ID"): + client_with(handler).submit(SeedanceSubmission(model="m", prompt="x")) + + def test_network_failure_is_classified_retryable(self): + def handler(request): + raise httpx.ConnectError("connection refused") + + with pytest.raises(SeedanceError) as exc: + client_with(handler).submit(SeedanceSubmission(model="m", prompt="x")) + assert exc.value.classification.retry is Retry.BACKOFF + + +class TestGetSuccessfulTask: + def test_parses_the_official_example_payload(self): + def handler(request): + return httpx.Response(200, json=OFFICIAL_SUCCESS_PAYLOAD) + + task = client_with(handler).get("cgt-1") + assert task.status is TaskStatus.SUCCEEDED + assert task.status.is_terminal + assert task.video_url.endswith("/xxx") + assert task.ratio == "16:9" + assert task.resolution == "720p" + assert task.fps == 24 + assert task.completion_tokens == 108900 + assert task.failure is None + + def test_duration_from_frames_is_exact_when_frames_returned(self): + # 官方例子:133 帧 / 24fps = 5.54 秒。按帧算才精确。 + def handler(request): + return httpx.Response( + 200, + json={ + "id": "x", + "status": "succeeded", + "frames": 133, + "framespersecond": 24, + "content": {}, + }, + ) + + assert client_with(handler).get("x").duration_ms == 5542 + + def test_duration_from_seconds_is_the_vendors_floored_approximation(self): + # duration 与 frames 只会返回一个。只有 duration 时只能用它, + # 但它是 帧数/24 向下取整的约数,最多可能短 1 秒。 + def handler(request): + return httpx.Response( + 200, + json={"id": "x", "status": "succeeded", "duration": 5, "content": {}}, + ) + + assert client_with(handler).get("x").duration_ms == 5000 + + def test_duration_is_none_rather_than_a_made_up_zero(self): + def handler(request): + return httpx.Response( + 200, json={"id": "x", "status": "running", "content": {}} + ) + + assert client_with(handler).get("x").duration_ms is None + + def test_last_frame_url_is_picked_up_as_cover_image(self): + def handler(request): + return httpx.Response( + 200, + json={ + "id": "x", + "status": "succeeded", + "content": { + "video_url": "https://v", + "last_frame_url": "https://f.png", + }, + }, + ) + + assert client_with(handler).get("x").last_frame_url == "https://f.png" + + +class TestGetFailedTask: + def test_failure_is_classified(self): + def handler(request): + return httpx.Response( + 200, + json={ + "id": "x", + "status": "failed", + "content": {}, + "error": { + "code": "OutputVideoSensitiveContentDetected", + "message": "output may contain sensitive information", + }, + }, + ) + + task = client_with(handler).get("x") + assert task.failure.retry is Retry.NEVER + assert task.failure.user_facing is UserFacing.CONTENT_REJECTED + + def test_expired_is_handled_even_though_the_query_doc_omits_it(self): + """查询接口文档只列了 5 个状态,`expired` 仅见于创建接口的回调说明。 + + 只按查询文档实现会让超时任务掉进未知分支。它是超时而非内容问题, + 因此可重试;且厂商在 expired 时不一定给 error.code,故单独处理。 + """ + + def handler(request): + return httpx.Response( + 200, json={"id": "x", "status": "expired", "content": {}} + ) + + task = client_with(handler).get("x") + assert task.status is TaskStatus.EXPIRED + assert task.status.is_terminal + assert task.failure.retry is Retry.BACKOFF + assert task.failure.provider_code == "TaskExpired" + + def test_cancelled_is_terminal_but_not_a_failure(self): + def handler(request): + return httpx.Response( + 200, json={"id": "x", "status": "cancelled", "content": {}} + ) + + task = client_with(handler).get("x") + assert task.status.is_terminal + assert task.failure is None + + +class TestUnknownStatusDoesNotBlowUp: + """厂商新增状态时抛异常会让所有在途任务一起失败。""" + + def test_unknown_status_yields_none_and_preserves_raw(self): + def handler(request): + return httpx.Response( + 200, json={"id": "x", "status": "brand_new", "content": {}} + ) + + task = client_with(handler).get("x") + assert task.status is None + assert task.raw_status == "brand_new" + assert task.failure is None # 当作未结束,继续轮询 + + +class TestHttpErrorMapping: + def test_rate_limit_is_retryable(self): + def handler(request): + return httpx.Response( + 429, + json={ + "error": { + "code": "RateLimitExceeded.EndpointRPMExceeded", + "message": "RPM limit exceeded", + } + }, + ) + + with pytest.raises(SeedanceError) as exc: + client_with(handler).get("x") + assert exc.value.classification.retry is Retry.BACKOFF + assert exc.value.classification.user_facing is UserFacing.RATE_LIMITED + assert exc.value.http_status == 429 + + def test_quota_exhausted_is_not_worth_retrying(self): + def handler(request): + return httpx.Response( + 429, + json={ + "error": { + "code": "QuotaExceeded", + "message": "Your account has exhausted its free trial quota.", + } + }, + ) + + with pytest.raises(SeedanceError) as exc: + client_with(handler).submit(SeedanceSubmission(model="m", prompt="x")) + assert exc.value.classification.retry is Retry.QUOTA + + def test_bad_api_key_is_our_own_fault(self): + def handler(request): + return httpx.Response( + 401, + json={ + "error": { + "code": "AuthenticationError", + "message": "invalid api key", + } + }, + ) + + with pytest.raises(SeedanceError) as exc: + client_with(handler).get("x") + assert exc.value.classification.retry is Retry.NEVER + assert exc.value.classification.user_facing is UserFacing.SERVER_ERROR + + def test_non_json_error_body_is_not_swallowed(self): + # 打到错误地址时常见:网关返回 HTML 错误页。静默会让排查绕远路。 + def handler(request): + return httpx.Response(502, text="bad gateway") + + with pytest.raises(SeedanceError) as exc: + client_with(handler).get("x") + assert "bad gateway" in str(exc.value) + assert exc.value.classification.retry is Retry.BACKOFF + + def test_non_json_success_body_is_an_error_too(self): + def handler(request): + return httpx.Response(200, text="not json") + + with pytest.raises(SeedanceError, match="不是合法 JSON"): + client_with(handler).get("x") + + +class TestConstruction: + def test_missing_api_key_fails_loudly_at_construction(self): + with pytest.raises(ValueError, match="ARK_API_KEY"): + SeedanceClient(api_key="")