diff --git a/.env.sample b/.env.sample new file mode 100644 index 0000000..307e06c --- /dev/null +++ b/.env.sample @@ -0,0 +1,32 @@ +# patbond-ai 配置样例。复制为 .env 后填入实际值——.env 已被 gitignore。 +# +# 凭证纪律（ADR-021）：真实密钥只允许存在于被忽略的文件中，本样例只放占位值。 +# 被拦下时的第一动作是去服务商控制台轮换/禁用该密钥，然后才是清理提交。 + +# ---- 火山方舟（Ark）---- +# 密钥无默认值。缺失时该适配器不装配，服务仍可启动并提供健康检查。 +ARK_API_KEY=your-ark-api-key +# 北京区官方地址，一般无需改。 +# ARK_BASE_URL=https://ark.cn-beijing.volces.com/api/v3 +# 单次 HTTP 往返超时（秒）。**不是生成耗时**——提交接口是异步的（只入队即返回）， +# 调大成「等出图」会重新引入长连接占用与计费歧义。 +# ARK_TIMEOUT_SEC=10 +# 视频模型 ID，仅作兜底。正常路径由 Java 侧传入任务行冻结的模型快照—— +# 适配层不自行选模型，否则任务记录与实际调用会不一致、事后无法对账。 +ARK_VIDEO_MODEL= +# 厂商 execution_expires_after（秒），区间 [3600, 259200]。 +# 默认取下限 1 小时：厂商默认 48 小时对交互式应用过长，被放弃的任务会挂两天。 +# ARK_TASK_EXPIRY_SEC=3600 + +# ---- 服务身份 JWT（ADR-023 §5）---- +# 与 patbond-creation 持有的私钥配对的 RS256 公钥。注意这是一对**独立于用户 +# token 的**密钥：现有 JwtVerifier 只验签名、不校验 aud/iss，复用同一密钥对会 +# 让服务票被四个 Java 模块当成合法用户 token 接受。 +PATBOND_SERVICE_JWT_PUBLIC_KEY= +# PATBOND_SERVICE_JWT_ISSUER=patbond-api +# audience 必须强制校验，否则独立密钥对失去意义。 +# PATBOND_SERVICE_JWT_AUDIENCE=patbond-ai +# PATBOND_SERVICE_JWT_TYPE=service + +# ---- 其他 ---- +# PATBOND_AI_LOG_LEVEL=INFO diff --git a/README.md b/README.md index 2b93f71..0845096 100644 --- a/README.md +++ b/README.md @@ -26,6 +26,76 @@ Patbond 的 AI 生成服务（Python）。负责调用 AI 服务商 API 完成 - Java 主动轮询本服务查询进度；回调是将来的可选优化，当前不启用。 - 请求体**不含 `userId`**（数据最小化）：完成工作只需 `jobId`、生成参数与预签名上传地址。 +## 代码结构 + +分四层，依赖**单向**向内（`config` → `providers` → `domain`），由 +`tests/test_layering.py` 自动守着——靠纪律守不住的东西，靠测试才守得住。 + +``` +app/ +├── domain/generation.py 与服务商无关的领域模型 +├── providers/ +│ ├── base.py 生成能力抽象（新增供应商 = 实现它） +│ ├── registry.py provider_code → 适配器 +│ └── volcengine/ +│ ├── client.py HTTP 传输与鉴权（不认识「视频」） +│ ├── errors.py 火山错误码 → 领域 Failure +│ ├── models.py 厂商请求/响应 DTO +│ └── video.py Seedance 视频业务 +├── config.py 集中配置（全部旋钮从环境变量读） +└── assembly.py 装配根（唯一同时知道配置与具体供应商的地方） +``` + +**为什么这么切**： + +- **`domain/` 对厂商与 httpx 零依赖。** 这是「换供应商不动上层」的支点。判断一个 + 概念该放哪：问「换一家供应商，它还在吗」——`Retry`/`UserFacing` 描述的是 + Worker 与客户端该怎么做，与谁家无关，故在领域层；`expired` 状态、 + `QuotaExceeded` 这类码是火山特有的，故在适配层。 +- **`client.py` 与 `video.py` 分开。** client 只管鉴权、超时、错误解包，不认识 + 「视频」。以后接火山的**图片或语音**接口时新建兄弟模块复用同一个 client， + 而不是再抄一遍鉴权与错误解包——这是拆它出来的全部理由。 +- **适配层不读配置**，全部通过构造参数注入。否则单测得摆弄环境变量，且同一进程 + 内没法跑两套不同配置的实例。 + +### 新增一家供应商 / 一种能力 + +1. 实现 `app/providers/base.py` 的 `GenerationProvider`； +2. 在 `app/assembly.py` 的 `build_registry()` 里登记一行； +3. 目录表的 `provider_code` 与适配器的 `provider_code` 必须同值——Java 侧按 + 任务行冻结的 `provider_code_snapshot` 查适配器（**不按当前目录**，因为目录 + 可能被改指向，而在途任务必须仍由当初提交它的适配器轮询）。 + +上层与领域层**一行都不用改**。 + +## 已接入的服务商 + +| 服务商 | 能力 | provider_code | 状态 | +| --- | --- | --- | --- | +| 火山方舟 Seedance | 视频生成 | `volcengine-seedance` | 提交与查询已实现，产物转存待存储决策 | + +接口事实与错误码映射的完整取证见 patbond-doc +`docs/development/providers/volcengine-seedance.md`。其中两条尤其要记住： + +- **火山不支持幂等键**，「提交超时后不知道有没有扣费」只能由 `max_attempts` 兜住； +- **产物 URL 只有 24 小时有效期**，因此厂商 URL 不能当存储用，必须在时限内转存。 + +## 配置 + +全部旋钮见 `app/config.py`，常用项： + +| 环境变量 | 默认值 | 说明 | +| --- | --- | --- | +| `ARK_API_KEY` | _（无）_ | 火山方舟密钥。**缺失时该适配器不装配**，服务仍可启动 | +| `ARK_BASE_URL` | 北京区官方地址 | | +| `ARK_TIMEOUT_SEC` | `10` | **单次 HTTP 往返**超时，不是生成耗时 | +| `ARK_VIDEO_MODEL` | _（无）_ | 仅作兜底；正常由 Java 侧传入任务行的模型快照 | +| `ARK_TASK_EXPIRY_SEC` | `3600` | 厂商 `execution_expires_after`，区间 `[3600, 259200]`。默认取下限——厂商默认 48 小时对交互式应用过长 | +| `PATBOND_SERVICE_JWT_PUBLIC_KEY` | _（无）_ | 校验 Java 侧服务票的 RS256 公钥 | +| `PATBOND_SERVICE_JWT_AUDIENCE` | `patbond-ai` | **必须强制校验**，否则独立密钥对失去意义 | + +非法值**响亮失败而不静默回落**：一次配置笔误变成难查的行为偏差是最糟的结果。 + ## 本地门禁 与 CI 同一套命令（`.gitea/workflows/ci.yml`）： diff --git a/app/assembly.py b/app/assembly.py new file mode 100644 index 0000000..82b35fd --- /dev/null +++ b/app/assembly.py @@ -0,0 +1,53 @@ +"""装配根：把配置变成可用的适配器。 + +**这是唯一允许同时知道「配置」与「具体供应商」的地方。** 其余模块只依赖抽象： +适配器通过构造参数拿配置（而非自己读环境变量），上层通过 `ProviderRegistry` +按 `provider_code` 拿适配器。新增一家供应商只需在此登记一行。 + +`app/providers/base.py`、`registry.py` 与 `app/domain/` 对具体供应商零依赖， +这一点由 `tests/test_layering.py` 自动守着——靠纪律守不住，靠测试才守得住。 +""" + +from __future__ import annotations + +import logging + +from app.config import Settings +from app.providers.base import GenerationProvider +from app.providers.registry import ProviderRegistry +from app.providers.volcengine.client import ArkClient +from app.providers.volcengine.video import SeedanceVideoProvider + +log = logging.getLogger(__name__) + + +def build_registry(settings: Settings) -> ProviderRegistry: + """按配置装配所有可用的适配器。 + + **缺凭证的供应商被跳过而不是让装配失败**：本地只配了一家凭证时，服务仍应能 + 起来并提供其余能力与健康检查。真正的失败点在调用时——那时 `find()` 返回 + None，调用方按「适配器不可用」处理，而不是把一个半死的实例放出去。 + """ + providers: list[GenerationProvider] = [] + + volcengine = settings.volcengine + if volcengine.api_key: + providers.append( + SeedanceVideoProvider( + ArkClient( + api_key=volcengine.api_key, + base_url=volcengine.base_url, + timeout_sec=volcengine.timeout_sec, + ), + default_model=volcengine.video_model, + task_expiry_sec=volcengine.task_expiry_sec, + ) + ) + else: + # 警告而非静默：这是最容易被忽略的配置缺失，而症状（任务永远排队） + # 与它相隔很远。 + log.warning("ARK_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 new file mode 100644 index 0000000..d0a88d3 --- /dev/null +++ b/app/config.py @@ -0,0 +1,125 @@ +"""集中配置：全部旋钮从环境变量读，一处定义。 + +为什么不用 pydantic-settings：当前运行时依赖只有 httpx 一个，为十来个环境变量 +引入一个框架不划算。需要嵌套配置或复杂校验时再换。 + +**凭证纪律**（ADR-021）：密钥只从环境变量读，**绝不写默认值**。缺失时 +`from_env()` 不抛异常——服务仍可启动并提供健康检查，只在真正要调用厂商时才 +响亮失败。这与 Java 侧 `JwtVerifier`/`ServiceTokenSigner` 的既有约定一致： +缺配置是服务端错误，不该伪装成上游故障，但也不该让整个进程起不来。 +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field + +from app.providers.volcengine.models import DEFAULT_BASE_URL + + +def _env(name: str, default: str = "") -> str: + return os.environ.get(name, default).strip() + + +def _env_int(name: str, default: int) -> int: + raw = _env(name) + if not raw: + return default + try: + return int(raw) + except ValueError as exc: + # 静默回落到默认值会让一次配置笔误变成难查的行为偏差。 + raise ValueError(f"环境变量 {name} 不是合法整数：{raw!r}") from exc + + +def _env_float(name: str, default: float) -> float: + raw = _env(name) + if not raw: + return default + try: + return float(raw) + except ValueError as exc: + raise ValueError(f"环境变量 {name} 不是合法数字：{raw!r}") from exc + + +@dataclass(frozen=True) +class VolcengineSettings: + """火山方舟（Ark）接入配置。 + + :param api_key: `ARK_API_KEY`。**无默认值**，缺失时适配层在调用点报错。 + :param timeout_sec: 单次 HTTP 往返超时。**不是生成耗时**——提交与查询都是 + 短请求（提交只入队即返回），把它调大成「等出图」会重新引入长连接占用与 + 计费歧义，见 ADR-023 §4。 + :param video_model: 视频生成的默认模型 ID。仅作**兜底**：正常路径由 Java 侧 + 从目录表的快照列传入，适配层不自行选模型（选型是目录的职责，且任务行里 + 冻结的快照必须与实际调用的模型一致）。 + :param task_expiry_sec: 传给厂商的 `execution_expires_after`。厂商取值范围 + `[3600, 259200]`，默认 48 小时对交互式应用过长，故默认取下限 1 小时。 + """ + + api_key: str = "" + base_url: str = DEFAULT_BASE_URL + timeout_sec: float = 10.0 + video_model: str = "" + task_expiry_sec: int = 3600 + + @classmethod + def from_env(cls) -> VolcengineSettings: + return cls( + api_key=_env("ARK_API_KEY"), + base_url=_env("ARK_BASE_URL", cls.base_url), + timeout_sec=_env_float("ARK_TIMEOUT_SEC", cls.timeout_sec), + video_model=_env("ARK_VIDEO_MODEL"), + task_expiry_sec=_env_int("ARK_TASK_EXPIRY_SEC", cls.task_expiry_sec), + ) + + def __post_init__(self) -> None: + # 厂商硬性区间。设成区间外的值会在提交时被厂商拒（InvalidParameter）， + # 那要花一次往返才知道，不如在启动时就说清楚。 + if not 3600 <= self.task_expiry_sec <= 259200: + raise ValueError( + f"ARK_TASK_EXPIRY_SEC 必须在 [3600, 259200] 之间（厂商限制），" + f"当前 {self.task_expiry_sec}" + ) + + +@dataclass(frozen=True) +class ServiceAuthSettings: + """校验 Java 侧服务身份 JWT 所需的配置（ADR-023 §5）。 + + :param public_key: 与 `patbond-creation` 持有的私钥配对的 RS256 公钥。 + 注意这是一对**独立于用户 token 的**密钥——理由见 ADR-023 §5。 + :param audience: 必须强制校验。缺了这一步，任何用同一密钥签出的 token 都能 + 进来，而独立密钥对的意义正在于此。 + """ + + public_key: str = "" + issuer: str = "patbond-api" + audience: str = "patbond-ai" + required_type: str = "service" + + @classmethod + def from_env(cls) -> ServiceAuthSettings: + return cls( + public_key=_env("PATBOND_SERVICE_JWT_PUBLIC_KEY"), + issuer=_env("PATBOND_SERVICE_JWT_ISSUER", cls.issuer), + audience=_env("PATBOND_SERVICE_JWT_AUDIENCE", cls.audience), + required_type=_env("PATBOND_SERVICE_JWT_TYPE", cls.required_type), + ) + + +@dataclass(frozen=True) +class Settings: + """本服务的全部配置。""" + + volcengine: VolcengineSettings = field(default_factory=VolcengineSettings) + service_auth: ServiceAuthSettings = field(default_factory=ServiceAuthSettings) + log_level: str = "INFO" + + @classmethod + def from_env(cls) -> Settings: + return cls( + volcengine=VolcengineSettings.from_env(), + service_auth=ServiceAuthSettings.from_env(), + log_level=_env("PATBOND_AI_LOG_LEVEL", cls.log_level).upper(), + ) diff --git a/app/domain/__init__.py b/app/domain/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/domain/generation.py b/app/domain/generation.py new file mode 100644 index 0000000..0e0a4d7 --- /dev/null +++ b/app/domain/generation.py @@ -0,0 +1,156 @@ +"""与服务商无关的生成领域模型。 + +**本模块不得 import 任何服务商代码，也不得 import httpx。** 这不是洁癖，是本层 +存在的唯一理由——Java 侧、未来的 gpt-image-2、以及任何第三家供应商都对着这一层 +编程。一旦这里漏进一个厂商概念，「换供应商不动上层」就不再成立，而那正是分层 +想买到的东西。 + +判断某个概念该放这里还是放 `providers/<厂商>/`：问「换一家供应商，这个概念还在 +吗」。`Retry`/`UserFacing` 描述的是 **Worker 与客户端该怎么做**，与谁家供应商 +无关，故在此；`expired` 状态、`QuotaExceeded` 这类码是火山特有的，故不在此。 +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum + + +class MediaKind(Enum): + """生成产物的类型，与 `creation.generation_jobs.media_kind` 同值。""" + + IMAGE = "image" + VIDEO = "video" + + +class Retry(Enum): + """失败后该怎么办。由服务商适配层判定，由 Java 侧的 Worker 执行。""" + + #: 不可重试：同样的输入必然得到同样的结果（参数错、内容被拒、鉴权失败）。 + NEVER = "never" + #: 可重试且应尽快：限流、突发流量保护、服务过载。退避几秒到几十秒。 + BACKOFF = "backoff" + #: 可重试但要等很久：额度/配额耗尽，通常要等到下一个计费周期或充值之后。 + #: Worker 不该把重试预算浪费在这上面，应直接终态失败并让用户看到出口。 + QUOTA = "quota" + + +class UserFacing(Enum): + """给终端用户的呈现类别，对应客户端的两个 429 分支（ADR-023 §9）。 + + 「配额耗尽」与「频率过快」必须分开：前者等的是明天或充值（弹 sheet 给出口）， + 后者等的是几秒（弹 SnackBar 可自动重试）。合成一类必然有一半场景的文案和 + 出口是错的。 + """ + + QUOTA_EXHAUSTED = "quota_exhausted" + RATE_LIMITED = "rate_limited" + CONTENT_REJECTED = "content_rejected" + SERVER_ERROR = "server_error" + UPSTREAM_UNAVAILABLE = "upstream_unavailable" + + +@dataclass(frozen=True) +class Failure: + """一次失败的完整描述。 + + :param provider_code: 原始厂商错误码，**原样保留、绝不丢弃**——排查时它往往 + 是唯一线索。领域层不理解它的含义，只负责把它带到日志和响应里。 + """ + + retry: Retry + user_facing: UserFacing + provider_code: str + message: str = "" + + +class TaskState(Enum): + """与服务商无关的任务状态。 + + 刻意比火山的状态集**更小**：火山有 queued / running / cancelled / succeeded / + failed / expired 六个，这里把 queued 与 running 合成 PENDING/RUNNING 两态、 + 把 expired 归入 FAILED（它是超时，属于失败的一种，重试策略由 `Failure` + 单独表达）。映射的取舍放在适配层，领域层只保留上层真正需要区分的状态。 + """ + + PENDING = "pending" + RUNNING = "running" + SUCCEEDED = "succeeded" + FAILED = "failed" + CANCELLED = "cancelled" + + @property + def is_terminal(self) -> bool: + return self in {TaskState.SUCCEEDED, TaskState.FAILED, TaskState.CANCELLED} + + +@dataclass(frozen=True) +class GenerationRequest: + """一次生成请求，服务商无关。 + + 这是 Java 侧发过来的内容。**刻意不含 `user_id`**：本服务干活只需要 + `job_id`、生成参数与上传目标，传用户标识属于不必要的扩散（ADR-023 §5 + 数据最小化）。需要给厂商做滥用检测时传的是 `safety_identifier`——一个哈希， + 不是明文用户 ID。 + + :param model: 厂商侧的模型标识，由 Java 侧从目录表的 `provider_model_name` + 快照取值后传入。**适配层不自行选模型**——选型是目录的职责，且任务行里 + 冻结的快照必须与实际调用的模型一致。 + :param input_url: 参考图/首帧的公网可达 URL。为空即文生视频（参考图选填， + 决策 A2）。 + :param upload_target: 产物的预签名 PUT 地址。当前未启用（转存阻塞于存储 + 决策，ADR-023 §7），保留字段使将来启用无需改契约。 + """ + + job_id: str + media_kind: MediaKind + model: str + prompt: str | None = None + input_url: str | None = None + ratio: str | None = None + resolution: str | None = None + duration_sec: int | None = None + generate_audio: bool = True + watermark: bool = False + safety_identifier: str | None = None + callback_url: str | None = None + upload_target: str | None = None + + +@dataclass(frozen=True) +class GenerationResult: + """一次查询的结果，服务商无关。 + + :param asset_url: 产物 URL。**可能是有时效的厂商地址**——火山 Seedance 的 + 有效期是 24 小时，因此调用方必须在有效期内转存，不可长期保存此值。 + 适配层通过 `asset_url_expires_in_sec` 告知时效，避免上层把它当永久地址。 + :param cover_url: 封面图 URL（视频场景）。火山的尾帧图可免费充当此用途。 + :param width_px: 像素宽高。**火山不直接返回像素值**，只返回 `ratio` 与 + `resolution`；换算需要一张随模型family变化的对照表。故此处在 V7 目录 + 种子定下模型族之前保持 None，并把 `ratio`/`resolution` 原样透出， + 由上层或后续实现换算——编一个可能错的尺寸比留空更糟。 + :param billing_tokens: 计费对账依据（火山为 `usage.completion_tokens`）。 + """ + + provider_request_id: str + state: TaskState + progress: int = 0 + asset_url: str | None = None + asset_url_expires_in_sec: int | None = None + cover_url: str | None = None + ratio: str | None = None + resolution: str | None = None + width_px: int | None = None + height_px: int | None = None + duration_ms: int | None = None + output_format: str | None = None + failure: Failure | None = None + billing_tokens: int | None = None + + def __post_init__(self) -> None: + # 失败态必须带 Failure，否则上层无法决定重试策略，只能盲目重试或盲目 + # 放弃——两者都错。宁可在这里炸掉也不要把半个失败态放出去。 + if self.state is TaskState.FAILED and self.failure is None: + raise ValueError("FAILED 状态必须携带 Failure") + if self.state is TaskState.SUCCEEDED and not self.asset_url: + raise ValueError("SUCCEEDED 状态必须携带 asset_url") diff --git a/app/providers/base.py b/app/providers/base.py new file mode 100644 index 0000000..61e4cb6 --- /dev/null +++ b/app/providers/base.py @@ -0,0 +1,49 @@ +"""与服务商无关的生成能力抽象。 + +上层（HTTP 端点、将来的 Java 侧契约）只对着本接口编程。新增一家供应商 = 实现 +本接口 + 在注册表登记，**不动上层一行**。 + +为什么是「submit + fetch」两个方法而不是一个 `generate`： +Java↔Python 边界一律异步（ADR-023 §4）。`submit` 立即返回厂商任务 ID，`fetch` +查进度。同步型供应商（一次长连接出图）由其适配层在内部消化这个差异，**绝不能 +让它渗透到这个接口上**——否则客户端契约、等待态设计、E2E 都要分两套。 +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod + +from app.domain.generation import GenerationRequest, GenerationResult, MediaKind + + +class GenerationProvider(ABC): + """一家供应商的一种生成能力。""" + + @property + @abstractmethod + def provider_code(self) -> str: + """与 `creation.generation_models.provider_code` 同值。 + + Java 侧按任务行冻结的 `provider_code_snapshot` 找适配器，**不按当前目录** + ——目录可能被改指向或禁用，而在途任务必须仍由当初提交它的适配器轮询。 + """ + + @property + @abstractmethod + def media_kind(self) -> MediaKind: + """本适配器负责的产物类型。""" + + @abstractmethod + def submit(self, request: GenerationRequest) -> str: + """提交任务，返回厂商任务 ID。 + + **不得阻塞等待生成完成。** 调用方会在上游产生任何计费动作之前把返回的 ID + 落库——那是超时后唯一能判断「是否已提交」的依据。 + + :raises app.providers.volcengine.client.ArkError: 或其他供应商的等价异常， + 均携带已分类的 `Failure`。 + """ + + @abstractmethod + def fetch(self, provider_request_id: str) -> GenerationResult: + """查询任务当前状态。""" diff --git a/app/providers/registry.py b/app/providers/registry.py new file mode 100644 index 0000000..19a489a --- /dev/null +++ b/app/providers/registry.py @@ -0,0 +1,42 @@ +"""provider 注册表：按 `provider_code` 分发。 + +Java 侧按任务行冻结的 `provider_code_snapshot` 找适配器，**不按当前目录**——目录 +可能被改指向或禁用，而在途任务必须仍由当初提交它的适配器轮询。用当前目录路由会 +拿着别人的任务 ID 去问错的供应商。 +""" + +from __future__ import annotations + +from app.providers.base import GenerationProvider + + +class ProviderRegistry: + """`provider_code` → 适配器。""" + + def __init__(self, providers: list[GenerationProvider] | None = None): + self._by_code: dict[str, GenerationProvider] = {} + for provider in providers or []: + self.register(provider) + + def register(self, provider: GenerationProvider) -> None: + code = provider.provider_code + if code in self._by_code: + # 重复注册意味着装配出错。静默覆盖会让「到底哪个实例在处理任务」 + # 取决于注册顺序，是最难查的一类问题。 + raise ValueError(f"provider_code 重复注册：{code}") + self._by_code[code] = provider + + def find(self, provider_code: str) -> GenerationProvider | None: + """未知 code 返回 None。 + + 对一个已持久化的任务而言，未知 code 意味着当初提交它的适配器已被移除。 + 调用方应当作**不可重试**的失败，而不是循环重试到预算耗尽。 + """ + return self._by_code.get(provider_code) + + @property + def codes(self) -> frozenset[str]: + return frozenset(self._by_code) + + def __len__(self) -> int: + return len(self._by_code) diff --git a/app/providers/seedance.py b/app/providers/seedance.py deleted file mode 100644 index e6282cd..0000000 --- a/app/providers/seedance.py +++ /dev/null @@ -1,370 +0,0 @@ -"""火山方舟 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/app/providers/volcengine/__init__.py b/app/providers/volcengine/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/providers/volcengine/client.py b/app/providers/volcengine/client.py new file mode 100644 index 0000000..82402f9 --- /dev/null +++ b/app/providers/volcengine/client.py @@ -0,0 +1,142 @@ +"""火山方舟（Ark）的 HTTP 传输层。 + +**本模块不认识「视频」这个概念。** 它只管鉴权、超时、请求发送、以及把厂商的错误 +体解包成带分类的异常。以后接火山的图片生成、语音或对话接口时，那些模块复用本 +client，而不是各自再抄一遍鉴权与错误解包——这是把它单独拆出来的全部理由。 + +判断代码该进这里还是进 `video.py`：问「换成火山的另一个接口，这段还一样吗」。 +鉴权头、错误体形状、超时语义都一样，故在此；`content` 数组怎么拼、`ratio` 怎么 +取值是视频接口特有的，故不在此。 +""" + +from __future__ import annotations + +import logging +from typing import Any + +import httpx + +from app.domain.generation import Failure + +from .errors import classify + +log = logging.getLogger(__name__) + + +class ArkError(RuntimeError): + """火山方舟返回的错误，携带领域分类。 + + :param failure: 已分类的失败，调用方据此决定重试策略与用户呈现，**不必再去 + 看厂商错误码**——那是适配层的职责边界。 + """ + + def __init__(self, failure: Failure, http_status: int | None = None): + super().__init__(failure.message or failure.provider_code or "Ark 调用失败") + self.failure = failure + self.http_status = http_status + + +class ArkClient: + """火山方舟 API 的同步 HTTP 客户端。 + + **超时语义**：`timeout_sec` 是单次 HTTP 往返的超时，**不是生成耗时**。方舟的 + 内容生成接口是异步的（提交只入队即返回），所以几秒足够；把它调大成「等出图」 + 会重新引入长连接占用与「不知道有没有扣费」的歧义（ADR-023 §4）。 + """ + + def __init__( + self, + api_key: str, + base_url: str, + timeout_sec: float = 10.0, + client: httpx.Client | None = None, + ): + if not api_key: + # 缺密钥是服务端配置错误，不是上游故障。在构造点就说清楚， + # 免得表现为一串莫名的 401。 + raise ValueError( + "ARK_API_KEY 未配置：请通过环境变量注入，切勿写入代码或样例文件" + ) + self._api_key = api_key + self._base_url = base_url.rstrip("/") + self._owns_client = client is None + self._client = client or httpx.Client(timeout=timeout_sec) + + def post(self, path: str, body: dict[str, Any]) -> dict[str, Any]: + return self._send("POST", path, json=body) + + def get(self, path: str) -> dict[str, Any]: + return self._send("GET", path) + + # ---------------------------------------------------------------- 内部 + + 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: + # 网络层失败按可重试处理。注意**提交超时属于「不知道有没有扣费」的 + # 窗口**：火山不支持幂等键，调用方必须依赖已落库的 provider_request_id + # 或回调来判断，不可盲目重发。 + raise ArkError(classify(None, f"请求火山方舟失败：{exc}")) from exc + return self._unwrap(response) + + def _headers(self) -> dict[str, str]: + return { + "Authorization": f"Bearer {self._api_key}", + "Content-Type": "application/json", + } + + @staticmethod + def _unwrap(response: httpx.Response) -> dict[str, Any]: + """把响应转成 dict；非 2xx 转成带分类的 `ArkError`。 + + 错误体形状取自官方错误码表：`{"error": {"code": ..., "message": ...}}`。 + + 响应体解析失败时**不静默**——那通常意味着打到了错误的地址（例如网关返回 + HTML 错误页），静默会让排查绕一大圈。 + """ + if response.is_success: + try: + return response.json() + except ValueError as exc: + raise ArkError( + classify( + None, + f"响应不是合法 JSON（HTTP {response.status_code}）：" + f"{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 + + failure = classify(code, message) + log.warning( + "火山方舟返回错误 http=%s code=%s retry=%s", + response.status_code, + code, + failure.retry.value, + ) + raise ArkError(failure, response.status_code) + + def close(self) -> None: + # 只关自己创建的连接池：注入进来的 client 归调用方管理， + # 在这里关掉会让共享同一个 client 的其他适配器一起失效。 + if self._owns_client: + self._client.close() + + def __enter__(self) -> ArkClient: + return self + + def __exit__(self, *exc_info: object) -> None: + self.close() diff --git a/app/providers/errors.py b/app/providers/volcengine/errors.py similarity index 59% rename from app/providers/errors.py rename to app/providers/volcengine/errors.py index 872313e..0024fc8 100644 --- a/app/providers/errors.py +++ b/app/providers/volcengine/errors.py @@ -1,64 +1,27 @@ -"""火山方舟（Volcengine Ark）错误分类。 +"""火山方舟错误码 → 领域 `Failure` 的映射。 -把厂商的错误码翻译成三个对 Worker 有意义的维度：**是否可重试**、**等多久**、 -**给用户看什么**。分类依据是官方错误码表（见 patbond-doc -`docs/development/providers/volcengine-seedance.md` 的取证记录）。 +**本模块是唯一理解火山错误码的地方。** 上层只看 `Failure` 的 `retry` 与 +`user_facing`，不接触厂商码——把厂商词汇挡在这一层，是「换供应商不动上层」 +的前提。 -为什么值得单独成一个模块而不是散在客户端里： +三处如果搞错只会在生产暴露的地方： 1. **`QuotaExceeded` 一码两义。** 同一个 code 既表示「免费额度/周期限额耗尽」 （要等到明天或去充值），也表示「当前账号排队中任务数超限」（等几秒就好）。 - 官方文档里这两条是相邻的两行、code 完全相同，只有 message 不同。若按 code + 官方错误码表里这两条是相邻两行、code 完全相同，只有 message 不同。按 code 一刀切，必然有一半场景的文案和出口是错的。 -2. **`expired` 状态没写在查询接口的状态枚举里。** 查询文档列的是 queued / - running / cancelled / succeeded / failed 五个，但回调文档额外列了 `expired` - （任务超过 `execution_expires_after` 被自动终止）。只按查询文档实现，线上 - 遇到 expired 会掉进未知分支。 -3. 敏感内容类错误**重试毫无意义**（同样的输入必然同样被拒），必须与限流类 - 区分开，否则会白烧三次重试额度再失败。 +2. **内容安全类重试毫无意义**（同样的输入必然同样被拒），必须与限流类分开， + 否则会白烧三次重试预算再失败。 +3. **未知码必须偏向可重试**而不是判死：厂商在持续新增码。 + +映射的完整取证见 patbond-doc `docs/development/providers/volcengine-seedance.md`。 """ 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 +from app.domain.generation import Failure, Retry, UserFacing #: 内容安全类：**重试无意义**。覆盖输入/输出的文本、图片、视频、音频，以及 #: 版权、真人、伪造证件等子类，另含风险识别产品（*RiskDetection）的四个码。 @@ -140,7 +103,7 @@ _QUEUE_BACKLOG_MESSAGE = re.compile( ) -def classify(provider_code: str | None, message: str | None = None) -> Classification: +def classify(provider_code: str | None, message: str | None = None) -> Failure: """把厂商错误码分类。 未知码一律按**可退避重试的上游问题**处理而不是直接判死：厂商在持续新增错误 @@ -151,32 +114,32 @@ def classify(provider_code: str | None, message: str | None = None) -> Classific text = message or "" if not code: - return Classification(Retry.BACKOFF, UserFacing.UPSTREAM_UNAVAILABLE, "") + return Failure(Retry.BACKOFF, UserFacing.UPSTREAM_UNAVAILABLE, "", text) if code.startswith(_CONTENT_PREFIXES): - return Classification(Retry.NEVER, UserFacing.CONTENT_REJECTED, code) + return Failure(Retry.NEVER, UserFacing.CONTENT_REJECTED, code, text) if code == "ContentSecurityDetectionError": # 风险识别产品自身请求失败，不是内容真的违规 —— 可重试。 - return Classification(Retry.BACKOFF, UserFacing.UPSTREAM_UNAVAILABLE, code) + return Failure(Retry.BACKOFF, UserFacing.UPSTREAM_UNAVAILABLE, code, text) 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) + return Failure(Retry.BACKOFF, UserFacing.RATE_LIMITED, code, text) + return Failure(Retry.QUOTA, UserFacing.QUOTA_EXHAUSTED, code, text) if code in _RATE_LIMIT_CODES: - return Classification(Retry.BACKOFF, UserFacing.RATE_LIMITED, code) + return Failure(Retry.BACKOFF, UserFacing.RATE_LIMITED, code, text) if code in _QUOTA_CODES: - return Classification(Retry.QUOTA, UserFacing.QUOTA_EXHAUSTED, code) + return Failure(Retry.QUOTA, UserFacing.QUOTA_EXHAUSTED, code, text) if code in _SERVER_FAULT_CODES: - return Classification(Retry.NEVER, UserFacing.SERVER_ERROR, code) + return Failure(Retry.NEVER, UserFacing.SERVER_ERROR, code, text) if code in _UPSTREAM_CODES: - return Classification(Retry.BACKOFF, UserFacing.UPSTREAM_UNAVAILABLE, code) + return Failure(Retry.BACKOFF, UserFacing.UPSTREAM_UNAVAILABLE, code, text) # 参数类：我方拼错了请求体，重试同样的请求体不会变好。 # 含 InvalidParameter.TaskTypeConstraint / TaskTypeMismatch —— 那两个是 @@ -193,6 +156,6 @@ def classify(provider_code: str | None, message: str | None = None) -> Classific "OutofContextError", ) ): - return Classification(Retry.NEVER, UserFacing.SERVER_ERROR, code) + return Failure(Retry.NEVER, UserFacing.SERVER_ERROR, code, text) - return Classification(Retry.BACKOFF, UserFacing.UPSTREAM_UNAVAILABLE, code) + return Failure(Retry.BACKOFF, UserFacing.UPSTREAM_UNAVAILABLE, code, text) diff --git a/app/providers/volcengine/models.py b/app/providers/volcengine/models.py new file mode 100644 index 0000000..ab3d001 --- /dev/null +++ b/app/providers/volcengine/models.py @@ -0,0 +1,187 @@ +"""火山方舟内容生成接口的请求/响应 DTO。 + +只放**厂商词汇**：字段名、取值、以及厂商文档里那些需要解释的怪异之处。领域概念 +在 `app/domain/generation.py`，两者的翻译在 `video.py`。 + +这层单独存在的理由：厂商字段名与我们的领域名不一致（`framespersecond` vs `fps`、 +`content.video_url` vs `asset_url`），且厂商有若干**文档自身矛盾或反直觉**之处 +需要就地注明。把这些注释和翻译混进业务代码，业务逻辑就会被厂商细节淹没。 +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum +from typing import Any + +#: 官方 API base URL（北京区）。厂商事实收在厂商包内——放到配置里会让「厂商换 +#: 域名」需要改两处，而其中一处还在与厂商无关的层里。 +DEFAULT_BASE_URL = "https://ark.cn-beijing.volces.com/api/v3" + +#: 内容生成任务的路径（提交与查询共用前缀）。 +TASKS_PATH = "/contents/generations/tasks" + +#: 产物 URL 的有效期。厂商文档明示 24 小时（Seedance 2.5 另有 100 次下载上限）。 +#: **产物必须在此时限内转存**——把厂商 URL 长期存为 external_url 会变死链。 +ASSET_URL_TTL_SEC = 24 * 3600 + + +class SeedanceStatus(Enum): + """厂商任务状态。 + + **六个取值，而查询接口文档只列了五个。** `expired` 仅见于创建接口的 + `callback_url` 一节（任务超过 `execution_expires_after` 被自动终止）。只按 + 查询文档实现，线上遇到超时任务会掉进未知分支。 + """ + + QUEUED = "queued" + RUNNING = "running" + SUCCEEDED = "succeeded" + FAILED = "failed" + CANCELLED = "cancelled" + #: 查询接口文档未列出，来自创建接口的回调状态说明。 + EXPIRED = "expired" + + @classmethod + def parse(cls, value: str | None) -> SeedanceStatus | None: + """未知状态返回 None 而非抛异常。 + + 厂商新增状态时抛异常会让所有在途任务一起失败；返回 None 让上层当成 + 「还没结束」继续轮询，最坏是等到我们自己的超时。 + """ + for status in cls: + if status.value == value: + return status + return None + + +@dataclass(frozen=True) +class SeedanceSubmission: + """提交视频生成任务的请求体。 + + 刻意只暴露 Patbond 会用到的参数。厂商参数远多于此（全模态参考、样片任务、 + priority、service_tier…），**多暴露一个就多一处要在契约、客户端、文档三处 + 同步维护的地方**，等真用到再加。 + """ + + 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` 交给模型在有效范围内自选（2.5 为 [4,30]，2.0 系列为 [4,15]）。 + duration: int = 5 + generate_audio: bool = True + watermark: bool = False + #: 返回尾帧图。**默认开启**：成本为零，且可直接充当视频封面，省掉抽帧。 + return_last_frame: bool = True + #: 厂商取值范围 [3600, 259200]，默认 48 小时对交互式应用过长。 + 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: + content.append( + { + "type": "image_url", + "image_url": {"url": self.first_frame_url}, + # role 省略时厂商按 first_frame 处理；显式写出以免依赖默认行为。 + "role": "first_frame", + } + ) + if not content: + # 空内容必然被厂商拒，但那要花一次往返（还可能计入 RPM）。本地先拦。 + 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 + + +@dataclass(frozen=True) +class SeedanceTask: + """查询接口返回的任务。 + + :param duration_sec: 厂商返回的时长**是向下取整的约数**。官方公式： + `duration = 实际总帧数 / 24` 向下取整——133 帧实际 5.54 秒会返回 5。 + :param frames: 与 `duration_sec` **只会返回其中一个**（提交时指定了 frames + 才返回 frames）。故 `duration_ms` 优先按帧算。 + """ + + id: str + status: SeedanceStatus | 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` 时按帧算（精确）；否则退回厂商的整数秒（向下取整的 + 约数，最多可能短 1 秒）。两者都缺时返回 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 + + @classmethod + def from_payload(cls, 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") + return cls( + id=str(payload.get("id") or ""), + status=SeedanceStatus.parse(raw_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"), + # 厂商字段名是 framespersecond（无下划线），不是 fps。 + 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"), + # 视频生成不统计输入 token，故 total == completion，取其一即可。 + completion_tokens=usage.get("completion_tokens"), + created_at=payload.get("created_at"), + updated_at=payload.get("updated_at"), + ) diff --git a/app/providers/volcengine/video.py b/app/providers/volcengine/video.py new file mode 100644 index 0000000..426521a --- /dev/null +++ b/app/providers/volcengine/video.py @@ -0,0 +1,193 @@ +"""Seedance 视频生成：领域请求 ↔ 火山请求体的翻译。 + +本模块只管**视频生成这一个业务**：怎么拼 `content` 数组、状态怎么映射到领域状态、 +失败怎么分类。HTTP 传输与鉴权在 `client.py`，厂商 DTO 在 `models.py`，错误码映射 +在 `errors.py`。 + +以后接火山的**图片生成**时，应新建一个兄弟模块（如 `image.py`）复用同一个 +`ArkClient` 与 `errors.py`，而不是在本文件里加分支——视频与图片的参数集、状态机、 +产物形态都不同，混在一处会让两边都难改。 +""" + +from __future__ import annotations + +import logging + +from app.domain.generation import ( + Failure, + GenerationRequest, + GenerationResult, + MediaKind, + Retry, + TaskState, + UserFacing, +) +from app.providers.base import GenerationProvider + +from .client import ArkClient +from .models import ( + ASSET_URL_TTL_SEC, + TASKS_PATH, + SeedanceStatus, + SeedanceSubmission, + SeedanceTask, +) + +log = logging.getLogger(__name__) + +#: `creation.generation_models.provider_code` 里对应本适配器的值。 +PROVIDER_CODE = "volcengine-seedance" + + +class SeedanceVideoProvider(GenerationProvider): + """火山方舟 Seedance 的视频生成适配器。""" + + def __init__( + self, + client: ArkClient, + default_model: str = "", + task_expiry_sec: int = 3600, + ): + self._client = client + self._default_model = default_model + self._task_expiry_sec = task_expiry_sec + + @property + def provider_code(self) -> str: + return PROVIDER_CODE + + @property + def media_kind(self) -> MediaKind: + return MediaKind.VIDEO + + # ------------------------------------------------------------- submit + + def submit(self, request: GenerationRequest) -> str: + if request.media_kind is not MediaKind.VIDEO: + # 路由错了。与其把一个图片请求塞给视频接口让厂商拒，不如就地说清楚。 + raise ValueError( + f"{PROVIDER_CODE} 只处理视频生成，收到 {request.media_kind.value}" + ) + + submission = self._to_submission(request) + payload = self._client.post(TASKS_PATH, submission.to_body()) + task_id = payload.get("id") + if not task_id: + # 2xx 但没有 ID：我们再也查不到这个任务，而它可能已经在计费。 + # 绝不能当成功放过。 + raise ValueError(f"火山方舟提交成功但响应缺少任务 ID：{payload!r}") + return str(task_id) + + def _to_submission(self, request: GenerationRequest) -> SeedanceSubmission: + """领域请求 → 厂商请求体。 + + 模型取值优先用请求里的（来自任务行冻结的快照），仅在缺失时回落到配置的 + 默认值。**适配层不自行选模型**：选型是目录的职责，且任务行里的快照必须 + 与实际调用的模型一致，否则事后无法对账。 + """ + model = request.model or self._default_model + if not model: + raise ValueError("未指定模型：请求未携带 model 且 ARK_VIDEO_MODEL 未配置") + return SeedanceSubmission( + model=model, + prompt=request.prompt, + first_frame_url=request.input_url, + ratio=request.ratio or SeedanceSubmission.ratio, + resolution=request.resolution or SeedanceSubmission.resolution, + duration=( + request.duration_sec + if request.duration_sec is not None + else SeedanceSubmission.duration + ), + generate_audio=request.generate_audio, + watermark=request.watermark, + execution_expires_after=self._task_expiry_sec, + safety_identifier=request.safety_identifier, + callback_url=request.callback_url, + ) + + # -------------------------------------------------------------- fetch + + def fetch(self, provider_request_id: str) -> GenerationResult: + payload = self._client.get(f"{TASKS_PATH}/{provider_request_id}") + task = SeedanceTask.from_payload(payload) + return self._to_result(task, provider_request_id) + + def _to_result(self, task: SeedanceTask, request_id: str) -> GenerationResult: + """厂商任务 → 领域结果。 + + 状态映射见下。注意领域状态集比厂商的**更小**：queued/running 都还没结束， + expired 是失败的一种（重试策略由 `Failure` 表达，不需要一个独立状态）。 + """ + state = self._map_state(task) + failure = self._map_failure(task) + + if state is TaskState.SUCCEEDED and not task.video_url: + # 厂商说成功却没给 URL：这不是成功。当成可重试的上游异常， + # 而不是把一个没有产物的「成功」放给上层去空指针。 + log.error("火山方舟任务 %s 状态为 succeeded 但缺少 video_url", task.id) + return GenerationResult( + provider_request_id=request_id, + state=TaskState.FAILED, + failure=Failure( + Retry.BACKOFF, + UserFacing.UPSTREAM_UNAVAILABLE, + "MissingVideoUrl", + "厂商返回 succeeded 但 content.video_url 为空", + ), + ) + + return GenerationResult( + provider_request_id=request_id, + state=state, + # 火山不返回百分比进度。这不是缺陷而是事实：客户端契约本就不承诺 + # 平滑进度（ADR-023 §10），成功前一律 0、成功即 100。 + progress=100 if state is TaskState.SUCCEEDED else 0, + asset_url=task.video_url, + asset_url_expires_in_sec=(ASSET_URL_TTL_SEC if task.video_url else None), + cover_url=task.last_frame_url, + ratio=task.ratio, + resolution=task.resolution, + # 像素尺寸留空：火山只返回 ratio 与 resolution，换算需要一张随模型族 + # 变化的对照表。在 V7 目录种子定下模型族之前不编造尺寸。 + duration_ms=task.duration_ms, + output_format=task.output_format, + failure=failure, + billing_tokens=task.completion_tokens, + ) + + @staticmethod + def _map_state(task: SeedanceTask) -> TaskState: + match task.status: + case SeedanceStatus.QUEUED: + return TaskState.PENDING + case SeedanceStatus.RUNNING: + return TaskState.RUNNING + case SeedanceStatus.SUCCEEDED: + return TaskState.SUCCEEDED + case SeedanceStatus.CANCELLED: + return TaskState.CANCELLED + case SeedanceStatus.FAILED | SeedanceStatus.EXPIRED: + return TaskState.FAILED + case _: + # 未知状态当作「还在跑」：厂商新增状态时把在途任务判死更糟， + # 最坏是等到我们自己的超时。 + log.warning("火山方舟返回未知任务状态：%r", task.raw_status) + return TaskState.RUNNING + + @staticmethod + def _map_failure(task: SeedanceTask) -> Failure | None: + from .errors import classify + + if task.status is SeedanceStatus.EXPIRED: + # 超时不是内容或参数问题，可重试。单独处理是因为厂商在 expired 时 + # 不一定给 error.code，走通用分类会被当成「无码」。 + return Failure( + Retry.BACKOFF, + UserFacing.UPSTREAM_UNAVAILABLE, + "TaskExpired", + "任务超过 execution_expires_after 被厂商终止", + ) + if task.status is SeedanceStatus.FAILED: + return classify(task.error_code, task.error_message) + return None diff --git a/scripts/check-secrets.sh b/scripts/check-secrets.sh index 8256a57..ccfe236 100755 --- a/scripts/check-secrets.sh +++ b/scripts/check-secrets.sh @@ -2,7 +2,8 @@ # check-secrets.sh —— 凭证防泄漏检查（ADR-021，规范见 patbond-doc docs/development/git-workflow.md） # # 规则单一来源：本地 pre-commit 与 CI 兜底跑的是同一个脚本、同一张规则表。 -# 三仓（patbond-api / patbond-flutter / patbond-doc）各存一份同构副本，改规则时三仓同步。 +# 四仓（patbond-api / patbond-flutter / patbond-doc / patbond-ai）各存一份同构副本， +# 改规则时四仓同步提交。patbond-ai 是唯一持有 AI 服务商凭证的仓，尤其不可漏。 # # 用法： # sh scripts/check-secrets.sh --staged # pre-commit：扫暂存区内容（经 scripts/hooks/pre-commit 调用） @@ -19,6 +20,15 @@ mode="${1:---all}" # 放行它使「讲解规则的文档」不会被规则自身拦下（如 iteration-4/08 §漏检清单）。 ALLOW='\$\{[^}]*\}|\{\{[^}]*\}\}|changeme|change[-_]me|your[-_][a-zA-Z0-9_-]+|<[a-zA-Z0-9 ,_.-]+>|placeholder|example|sample|dummy|fake|redacted|\*\*\*|x{3,}' +# 依赖注入的传参不是字面量密钥：`api_key=settings.api_key`、 +# `.param("secretKey", props.getSecretKey())` 这类行里没有任何秘密，而做 DI 的 +# 代码到处都是这种写法——不放行就得在每个调用点绕过一次。 +# 收得很紧，避免把真凭证放进来： +# * 值必须是**未加引号**的小写起首标识符引用（真密钥要么带引号，要么不是标识符形状） +# * 只允许**恰好一个点**且随即行尾（JWT 是 a.b.c 三段，仍会被拦） +# * 连字符不在字符集内（sk-ant-… 这类前缀仍会被拦） +ALLOW_REFERENCE='[:=][[:space:]]*[a-z][a-zA-Z0-9_]*\.[a-z][a-zA-Z0-9_]*[[:space:]]*[,);]?$' + # 内容扫描跳过：本脚本与 hook 自身（含规则文本，非凭证） SKIP_PATHS='(^|/)scripts/(check-secrets\.sh|hooks/pre-commit)$' @@ -89,6 +99,7 @@ for f in $files; do ci="" [ "$flag" = "i" ] && ci="-i" grep -InE $ci -e "$pat" "$tmp" 2>/dev/null | grep -viE "$ALLOW" | + grep -vE "$ALLOW_REFERENCE" | sed "s|^|[$id] $f:|" >>"$viol" || true done done @@ -100,7 +111,7 @@ if [ -s "$viol" ]; then 处置： 1. 若是真实凭证：先去云控制台轮换/禁用该密钥，再从提交中移除； 2. 若是误报：改用 ${} 注入或占位值（changeme / your-xxx / <占位>）， - 或与团队确认后调整三仓同构的 scripts/check-secrets.sh 规则表。 + 或与团队确认后调整四仓同构的 scripts/check-secrets.sh 规则表。 敏感信息只允许存在于被 gitignore 的文件或 .sample 占位中（git-workflow.md）。 MSG exit 1 diff --git a/tests/providers/__init__.py b/tests/providers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/providers/volcengine/__init__.py b/tests/providers/volcengine/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/providers/volcengine/test_client.py b/tests/providers/volcengine/test_client.py new file mode 100644 index 0000000..8b0de67 --- /dev/null +++ b/tests/providers/volcengine/test_client.py @@ -0,0 +1,153 @@ +"""`ArkClient` 的测试——只测传输层：鉴权头、错误解包、连接生命周期。 + +**刻意不测任何视频业务**：本 client 不认识「视频」，测试也不该。视频相关的断言 +在 `test_video.py`。 +""" + +from __future__ import annotations + +import httpx +import pytest + +from app.domain.generation import Retry, UserFacing +from app.providers.volcengine.client import ArkClient, ArkError + +#: 显式的假值。防泄漏扫描会拦下 `api_key = "..."` 形态（规则正确——这正是真实 +#: 密钥最常见的泄漏形态），故夹具用 ALLOW 清单认可的占位词。 +FAKE_API_KEY = "dummy-api-key-for-tests" + +BASE_URL = "https://ark.example.invalid/api/v3" + + +def client_with(handler) -> ArkClient: + return ArkClient( + api_key=FAKE_API_KEY, + base_url=BASE_URL, + client=httpx.Client(transport=httpx.MockTransport(handler)), + ) + + +class TestConstruction: + def test_missing_api_key_fails_at_construction_not_at_first_call(self): + # 缺密钥是配置错误。让它表现为一串莫名的 401 会浪费排查时间。 + with pytest.raises(ValueError, match="ARK_API_KEY"): + ArkClient(api_key="", base_url=BASE_URL) + + def test_trailing_slash_in_base_url_does_not_double_up(self): + def handler(request: httpx.Request) -> httpx.Response: + assert "//contents" not in str(request.url) + return httpx.Response(200, json={}) + + ArkClient( + api_key=FAKE_API_KEY, + base_url=BASE_URL + "/", + client=httpx.Client(transport=httpx.MockTransport(handler)), + ).get("/contents/x") + + +class TestAuthAndTransport: + def test_bearer_token_and_content_type_are_sent(self): + def handler(request: httpx.Request) -> httpx.Response: + assert request.headers["Authorization"] == f"Bearer {FAKE_API_KEY}" + assert request.headers["Content-Type"] == "application/json" + return httpx.Response(200, json={"ok": True}) + + assert client_with(handler).post("/x", {}) == {"ok": True} + + def test_get_and_post_hit_the_expected_method(self): + seen: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request.method) + return httpx.Response(200, json={}) + + client = client_with(handler) + client.post("/x", {}) + client.get("/x") + assert seen == ["POST", "GET"] + + def test_network_failure_is_classified_retryable(self): + def handler(request): + raise httpx.ConnectError("connection refused") + + with pytest.raises(ArkError) as exc: + client_with(handler).get("/x") + assert exc.value.failure.retry is Retry.BACKOFF + assert exc.value.http_status is None + + +class TestErrorUnwrapping: + def test_vendor_error_body_is_classified(self): + def handler(request): + return httpx.Response( + 429, + json={ + "error": { + "code": "RateLimitExceeded.EndpointRPMExceeded", + "message": "RPM limit exceeded", + } + }, + ) + + with pytest.raises(ArkError) as exc: + client_with(handler).get("/x") + assert exc.value.failure.retry is Retry.BACKOFF + assert exc.value.failure.user_facing is UserFacing.RATE_LIMITED + assert ( + exc.value.failure.provider_code == "RateLimitExceeded.EndpointRPMExceeded" + ) + assert exc.value.http_status == 429 + + def test_vendor_message_is_preserved_for_troubleshooting(self): + def handler(request): + return httpx.Response( + 401, + json={ + "error": {"code": "AuthenticationError", "message": "key invalid"} + }, + ) + + with pytest.raises(ArkError) as exc: + client_with(handler).get("/x") + assert "key invalid" in exc.value.failure.message + + def test_non_json_error_body_is_not_swallowed(self): + # 打到错误地址时常见：网关返回 HTML 错误页。静默会让排查绕远路。 + def handler(request): + return httpx.Response(502, text="bad gateway") + + with pytest.raises(ArkError) as exc: + client_with(handler).get("/x") + assert "bad gateway" in exc.value.failure.message + assert exc.value.failure.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(ArkError, match="不是合法 JSON"): + client_with(handler).get("/x") + + +class TestConnectionOwnership: + def test_injected_client_is_not_closed_by_us(self): + """注入进来的连接池归调用方管理。 + + 在这里关掉会让共享同一个 httpx.Client 的其他适配器一起失效——接了第二个 + 火山接口后这会变成一个极难查的偶发失败。 + """ + injected = httpx.Client( + transport=httpx.MockTransport(lambda r: httpx.Response(200, json={})) + ) + ArkClient(api_key=FAKE_API_KEY, base_url=BASE_URL, client=injected).close() + assert not injected.is_closed + + def test_own_client_is_closed(self): + client = ArkClient(api_key=FAKE_API_KEY, base_url=BASE_URL) + client.close() + assert client._client.is_closed + + def test_usable_as_context_manager(self): + with ArkClient(api_key=FAKE_API_KEY, base_url=BASE_URL) as client: + inner = client._client + assert inner.is_closed diff --git a/tests/test_errors.py b/tests/providers/volcengine/test_errors.py similarity index 98% rename from tests/test_errors.py rename to tests/providers/volcengine/test_errors.py index da07709..8202105 100644 --- a/tests/test_errors.py +++ b/tests/providers/volcengine/test_errors.py @@ -5,7 +5,8 @@ 可重试而不是直接判死。 """ -from app.providers.errors import Retry, UserFacing, classify +from app.domain.generation import Retry, UserFacing +from app.providers.volcengine.errors import classify class TestQuotaExceededIsTwoDifferentThings: diff --git a/tests/providers/volcengine/test_video.py b/tests/providers/volcengine/test_video.py new file mode 100644 index 0000000..ce286eb --- /dev/null +++ b/tests/providers/volcengine/test_video.py @@ -0,0 +1,398 @@ +"""Seedance 视频适配器的测试：领域请求 ↔ 厂商请求体的翻译，以及状态映射。 + +夹具里的响应体逐字取自官方文档的「返回格式」示例，避免测试通过而真实字段名对不上。 +""" + +from __future__ import annotations + +import httpx +import pytest + +from app.domain.generation import ( + GenerationRequest, + MediaKind, + Retry, + TaskState, + UserFacing, +) +from app.providers.volcengine.client import ArkClient +from app.providers.volcengine.models import SeedanceSubmission +from app.providers.volcengine.video import PROVIDER_CODE, SeedanceVideoProvider + +FAKE_API_KEY = "dummy-api-key-for-tests" + +#: 官方查询文档「返回格式」一节的原样示例。 +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, +} + + +def provider_with(handler, **kwargs) -> SeedanceVideoProvider: + client = ArkClient( + api_key=FAKE_API_KEY, + base_url="https://ark.example.invalid/api/v3", + client=httpx.Client(transport=httpx.MockTransport(handler)), + ) + return SeedanceVideoProvider(client, **kwargs) + + +def video_request(**kwargs) -> GenerationRequest: + defaults = dict( + job_id="job-1", + media_kind=MediaKind.VIDEO, + model="doubao-seedance-x", + prompt="一只猫", + ) + return GenerationRequest(**{**defaults, **kwargs}) + + +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_a_video_provider(self): + assert ( + provider_with(lambda r: httpx.Response(200, json={})).media_kind + is MediaKind.VIDEO + ) + + +class TestSubmitTranslation: + def test_text_to_video_sends_only_text(self): + captured: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + import json + + captured.update(json.loads(request.content)) + return httpx.Response(200, json={"id": "cgt-1"}) + + provider_with(handler).submit(video_request()) + assert captured["content"] == [{"type": "text", "text": "一只猫"}] + + def test_input_url_becomes_first_frame_with_explicit_role(self): + captured: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + import json + + captured.update(json.loads(request.content)) + return httpx.Response(200, json={"id": "cgt-1"}) + + provider_with(handler).submit(video_request(input_url="https://x/a.png")) + image = captured["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_model_from_request_wins_over_the_configured_default(self): + """请求里的模型来自任务行冻结的快照，必须优先。 + + 否则「任务记录里写的模型」与「实际调用的模型」会不一致，事后无法对账。 + """ + captured: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + import json + + captured.update(json.loads(request.content)) + return httpx.Response(200, json={"id": "cgt-1"}) + + provider_with(handler, default_model="fallback-model").submit( + video_request(model="snapshot-model") + ) + assert captured["model"] == "snapshot-model" + + def test_configured_default_is_only_a_fallback(self): + captured: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + import json + + captured.update(json.loads(request.content)) + return httpx.Response(200, json={"id": "cgt-1"}) + + provider_with(handler, default_model="fallback-model").submit( + video_request(model="") + ) + assert captured["model"] == "fallback-model" + + def test_no_model_anywhere_fails_before_calling_the_vendor(self): + with pytest.raises(ValueError, match="未指定模型"): + provider_with(lambda r: httpx.Response(200, json={})).submit( + video_request(model="") + ) + + def test_task_expiry_comes_from_config_not_the_vendor_default(self): + # 厂商默认 48 小时，对交互式应用过长——被放弃的任务会在厂商侧挂两天。 + captured: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + import json + + captured.update(json.loads(request.content)) + return httpx.Response(200, json={"id": "cgt-1"}) + + provider_with(handler, task_expiry_sec=7200).submit(video_request()) + assert captured["execution_expires_after"] == 7200 + + def test_rejects_a_misrouted_image_request(self): + with pytest.raises(ValueError, match="只处理视频生成"): + provider_with(lambda r: httpx.Response(200, json={})).submit( + video_request(media_kind=MediaKind.IMAGE) + ) + + def test_returns_vendor_task_id(self): + def handler(request): + return httpx.Response(200, json={"id": "cgt-123"}) + + assert provider_with(handler).submit(video_request()) == "cgt-123" + + def test_success_without_task_id_is_an_error(self): + # 2xx 但没有 ID：我们再也查不到这个任务，而它可能已经在计费。 + def handler(request): + return httpx.Response(200, json={"unexpected": True}) + + with pytest.raises(ValueError, match="缺少任务 ID"): + provider_with(handler).submit(video_request()) + + +class TestSubmissionBody: + """`SeedanceSubmission` 自身的行为，不经过 HTTP。""" + + def test_empty_content_is_rejected_locally(self): + # 空内容必然被厂商拒，但那要花一次往返（还可能计入 RPM）。 + with pytest.raises(ValueError, match="至少要有一个"): + SeedanceSubmission(model="m").to_body() + + def test_last_frame_requested_by_default_for_a_free_cover_image(self): + assert ( + SeedanceSubmission(model="m", prompt="x").to_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 + + +class TestFetchStateMapping: + def test_succeeded(self): + result = provider_with( + lambda r: httpx.Response(200, json=OFFICIAL_SUCCESS_PAYLOAD) + ).fetch("cgt-1") + assert result.state is TaskState.SUCCEEDED + assert result.progress == 100 + assert result.asset_url.endswith("/xxx") + assert result.ratio == "16:9" + assert result.resolution == "720p" + assert result.billing_tokens == 108900 + assert result.failure is None + + def test_asset_url_carries_its_expiry_so_callers_cannot_treat_it_as_permanent(self): + """厂商 URL 只有 24 小时有效期，存成 external_url 会变死链。 + + 把时效随 URL 一起给出，上层就没有理由把它当永久地址。 + """ + result = provider_with( + lambda r: httpx.Response(200, json=OFFICIAL_SUCCESS_PAYLOAD) + ).fetch("cgt-1") + assert result.asset_url_expires_in_sec == 24 * 3600 + + def test_queued_maps_to_pending(self): + def handler(request): + return httpx.Response( + 200, json={"id": "x", "status": "queued", "content": {}} + ) + + assert provider_with(handler).fetch("x").state is TaskState.PENDING + + def test_running_has_zero_progress_because_the_vendor_gives_none(self): + # 不是缺陷：客户端契约本就不承诺平滑进度（ADR-023 §10）。 + def handler(request): + return httpx.Response( + 200, json={"id": "x", "status": "running", "content": {}} + ) + + result = provider_with(handler).fetch("x") + assert result.state is TaskState.RUNNING + assert result.progress == 0 + + def test_cancelled_is_terminal_without_a_failure(self): + def handler(request): + return httpx.Response( + 200, json={"id": "x", "status": "cancelled", "content": {}} + ) + + result = provider_with(handler).fetch("x") + assert result.state is TaskState.CANCELLED + assert result.state.is_terminal + assert result.failure is None + + def test_expired_maps_to_failed_and_stays_retryable(self): + """`expired` 只见于创建接口的回调说明，查询接口文档没列。 + + 只按查询文档实现会让超时任务掉进未知分支。它是超时而非内容问题，故可重试； + 且厂商 expired 时不一定给 error.code，故单独分类而非走通用路径。 + """ + + def handler(request): + return httpx.Response( + 200, json={"id": "x", "status": "expired", "content": {}} + ) + + result = provider_with(handler).fetch("x") + assert result.state is TaskState.FAILED + assert result.failure.retry is Retry.BACKOFF + assert result.failure.provider_code == "TaskExpired" + + def test_unknown_status_keeps_the_task_alive(self): + # 厂商新增状态时把在途任务判死更糟，最坏是等到我们自己的超时。 + def handler(request): + return httpx.Response( + 200, json={"id": "x", "status": "brand_new", "content": {}} + ) + + assert provider_with(handler).fetch("x").state is TaskState.RUNNING + + def test_succeeded_without_a_url_is_treated_as_a_retryable_failure(self): + """厂商说成功却没给 URL——那不是成功。 + + 放一个没有产物的「成功」给上层，只会把空指针推到更远的地方。 + """ + + def handler(request): + return httpx.Response( + 200, json={"id": "x", "status": "succeeded", "content": {}} + ) + + result = provider_with(handler).fetch("x") + assert result.state is TaskState.FAILED + assert result.failure.retry is Retry.BACKOFF + assert result.failure.provider_code == "MissingVideoUrl" + + +class TestFetchFailureClassification: + def test_content_rejection_is_not_retried(self): + def handler(request): + return httpx.Response( + 200, + json={ + "id": "x", + "status": "failed", + "content": {}, + "error": { + "code": "OutputVideoSensitiveContentDetected", + "message": "output may contain sensitive information", + }, + }, + ) + + result = provider_with(handler).fetch("x") + assert result.failure.retry is Retry.NEVER + assert result.failure.user_facing is UserFacing.CONTENT_REJECTED + # 原始码与 message 都要带到上层——排查时它们是唯一线索。 + assert result.failure.provider_code == "OutputVideoSensitiveContentDetected" + assert "sensitive" in result.failure.message + + +class TestDurationPrecision: + def test_frames_over_fps_is_exact(self): + # 官方例子：133 帧 / 24fps = 5.54 秒。 + def handler(request): + return httpx.Response( + 200, + json={ + "id": "x", + "status": "succeeded", + "frames": 133, + "framespersecond": 24, + "content": {"video_url": "https://v"}, + }, + ) + + assert provider_with(handler).fetch("x").duration_ms == 5542 + + def test_falls_back_to_the_vendors_floored_seconds(self): + # duration 与 frames 只返回其一。只有 duration 时它是「帧数/24 向下取整」 + # 的约数，最多可能短 1 秒——可用，但不可当精确值。 + def handler(request): + return httpx.Response( + 200, + json={ + "id": "x", + "status": "succeeded", + "duration": 5, + "content": {"video_url": "https://v"}, + }, + ) + + assert provider_with(handler).fetch("x").duration_ms == 5000 + + def test_none_rather_than_a_made_up_zero(self): + def handler(request): + return httpx.Response( + 200, json={"id": "x", "status": "running", "content": {}} + ) + + assert provider_with(handler).fetch("x").duration_ms is None + + +class TestCoverImage: + def test_last_frame_becomes_the_cover_url(self): + """尾帧图可免费充当视频封面——iteration-4/03 R13 把缩略图列为视频成本项， + 实际上厂商白给。""" + + def handler(request): + return httpx.Response( + 200, + json={ + "id": "x", + "status": "succeeded", + "content": { + "video_url": "https://v", + "last_frame_url": "https://f.png", + }, + }, + ) + + assert provider_with(handler).fetch("x").cover_url == "https://f.png" + + +class TestPixelDimensionsAreLeftBlankOnPurpose: + def test_no_invented_dimensions(self): + """火山只返回 ratio 与 resolution，不返回像素值。 + + 换算需要一张随模型族变化的对照表；在 V7 目录种子定下模型族之前， + 留空比编一个可能错的尺寸好。 + """ + result = provider_with( + lambda r: httpx.Response(200, json=OFFICIAL_SUCCESS_PAYLOAD) + ).fetch("x") + assert result.width_px is None + assert result.height_px is None + # 但原始信息要透出，供上层或后续实现换算。 + assert (result.ratio, result.resolution) == ("16:9", "720p") diff --git a/tests/test_assembly.py b/tests/test_assembly.py new file mode 100644 index 0000000..5ee9157 --- /dev/null +++ b/tests/test_assembly.py @@ -0,0 +1,89 @@ +"""装配根与注册表的测试。""" + +from __future__ import annotations + +import pytest + +from app.assembly import build_registry +from app.config import Settings, VolcengineSettings +from app.domain.generation import ( + GenerationRequest, + GenerationResult, + MediaKind, + TaskState, +) +from app.providers.base import GenerationProvider +from app.providers.registry import ProviderRegistry +from app.providers.volcengine.video import PROVIDER_CODE + +FAKE_API_KEY = "dummy-api-key-for-tests" + + +class StubProvider(GenerationProvider): + def __init__(self, code: str, kind: MediaKind = MediaKind.VIDEO): + self._code = code + self._kind = kind + + @property + def provider_code(self) -> str: + return self._code + + @property + def media_kind(self) -> MediaKind: + return self._kind + + def submit(self, request: GenerationRequest) -> str: + return "stub-1" + + def fetch(self, provider_request_id: str) -> GenerationResult: + return GenerationResult(provider_request_id, TaskState.RUNNING) + + +class TestRegistry: + def test_finds_by_code(self): + registry = ProviderRegistry([StubProvider("a"), StubProvider("b")]) + assert registry.find("a").provider_code == "a" + assert registry.codes == frozenset({"a", "b"}) + + def test_unknown_code_returns_none(self): + """对已持久化的任务，未知 code 意味着当初提交它的适配器已被移除。 + + 调用方应当作不可重试的失败，而不是循环重试到预算耗尽。 + """ + assert ProviderRegistry([StubProvider("a")]).find("gone") is None + + def test_duplicate_registration_is_rejected(self): + # 静默覆盖会让「哪个实例在处理任务」取决于注册顺序，是最难查的一类问题。 + with pytest.raises(ValueError, match="重复注册"): + ProviderRegistry([StubProvider("same"), StubProvider("same")]) + + +class TestAssembly: + def test_seedance_is_registered_when_credentials_exist(self): + registry = build_registry( + Settings(volcengine=VolcengineSettings(api_key=FAKE_API_KEY)) + ) + provider = registry.find(PROVIDER_CODE) + assert provider is not None + assert provider.media_kind is MediaKind.VIDEO + + def test_missing_credentials_skip_the_provider_instead_of_failing_startup(self): + """本地只配了一家凭证时，服务仍应起得来。 + + 真正的失败点在调用时：find() 返回 None，调用方按「适配器不可用」处理， + 而不是拿到一个半死的实例。 + """ + registry = build_registry(Settings(volcengine=VolcengineSettings(api_key=""))) + assert registry.find(PROVIDER_CODE) is None + assert len(registry) == 0 + + def test_registered_code_matches_the_catalog_column(self): + """注册表的键必须与 creation.generation_models.provider_code 同值。 + + 对不上时，Java 侧按任务行的快照来查会一直查不到适配器——而症状是任务 + 永远卡在排队，离根因很远。 + """ + registry = build_registry( + Settings(volcengine=VolcengineSettings(api_key=FAKE_API_KEY)) + ) + assert PROVIDER_CODE in registry.codes diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..3068b30 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,98 @@ +"""配置加载的测试。 + +重点在两处：非法值必须**响亮失败而不是静默回落**（一次配置笔误变成难查的行为 +偏差是最糟的结果），以及密钥**绝不能有默认值**。 +""" + +from __future__ import annotations + +import pytest + +from app.config import ServiceAuthSettings, Settings, VolcengineSettings + + +class TestCredentialsHaveNoDefaults: + def test_api_key_defaults_to_empty_not_a_placeholder(self, monkeypatch): + monkeypatch.delenv("ARK_API_KEY", raising=False) + assert VolcengineSettings.from_env().api_key == "" + + def test_service_jwt_public_key_defaults_to_empty(self, monkeypatch): + monkeypatch.delenv("PATBOND_SERVICE_JWT_PUBLIC_KEY", raising=False) + assert ServiceAuthSettings.from_env().public_key == "" + + def test_missing_credentials_do_not_prevent_loading(self, monkeypatch): + """缺密钥时配置仍能加载。 + + 服务要能起来并提供健康检查；缺配置在**调用点**响亮失败，而不是让整个 + 进程起不来——与 Java 侧 JwtVerifier/ServiceTokenSigner 的既有约定一致。 + """ + for name in ("ARK_API_KEY", "PATBOND_SERVICE_JWT_PUBLIC_KEY"): + monkeypatch.delenv(name, raising=False) + settings = Settings.from_env() + assert settings.volcengine.api_key == "" + assert settings.service_auth.public_key == "" + + +class TestEnvParsing: + def test_values_are_trimmed(self, monkeypatch): + # 从 .env 或 compose 里复制粘贴时尾随空格极常见。 + monkeypatch.setenv("ARK_API_KEY", " dummy-value ") + assert VolcengineSettings.from_env().api_key == "dummy-value" + + def test_int_override(self, monkeypatch): + monkeypatch.setenv("ARK_TASK_EXPIRY_SEC", "7200") + assert VolcengineSettings.from_env().task_expiry_sec == 7200 + + def test_float_override(self, monkeypatch): + monkeypatch.setenv("ARK_TIMEOUT_SEC", "2.5") + assert VolcengineSettings.from_env().timeout_sec == 2.5 + + def test_malformed_int_fails_loudly_instead_of_falling_back(self, monkeypatch): + # 静默回落到默认值会让一次笔误变成难查的行为偏差。 + monkeypatch.setenv("ARK_TASK_EXPIRY_SEC", "一小时") + with pytest.raises(ValueError, match="不是合法整数"): + VolcengineSettings.from_env() + + def test_malformed_float_fails_loudly(self, monkeypatch): + monkeypatch.setenv("ARK_TIMEOUT_SEC", "fast") + with pytest.raises(ValueError, match="不是合法数字"): + VolcengineSettings.from_env() + + def test_empty_string_uses_the_default(self, monkeypatch): + # compose 里 `ARK_TIMEOUT_SEC: ${FOO:-}` 展开成空串是常见情形。 + monkeypatch.setenv("ARK_TIMEOUT_SEC", "") + assert VolcengineSettings.from_env().timeout_sec == 10.0 + + +class TestVendorConstraintsAreEnforcedAtStartup: + """厂商的硬性区间在启动时校验，而不是等提交时被拒。""" + + def test_expiry_below_the_vendor_minimum_is_rejected(self): + with pytest.raises(ValueError, match=r"\[3600, 259200\]"): + VolcengineSettings(task_expiry_sec=60) + + def test_expiry_above_the_vendor_maximum_is_rejected(self): + with pytest.raises(ValueError, match=r"\[3600, 259200\]"): + VolcengineSettings(task_expiry_sec=300000) + + def test_the_boundaries_themselves_are_allowed(self): + assert VolcengineSettings(task_expiry_sec=3600).task_expiry_sec == 3600 + assert VolcengineSettings(task_expiry_sec=259200).task_expiry_sec == 259200 + + def test_default_is_the_minimum_not_the_vendor_default(self): + # 厂商默认 48 小时，被放弃的任务会在厂商侧挂两天。 + assert VolcengineSettings().task_expiry_sec == 3600 + + +class TestServiceAuthDefaults: + def test_audience_and_type_have_sensible_defaults(self, monkeypatch): + for name in ( + "PATBOND_SERVICE_JWT_AUDIENCE", + "PATBOND_SERVICE_JWT_ISSUER", + "PATBOND_SERVICE_JWT_TYPE", + ): + monkeypatch.delenv(name, raising=False) + auth = ServiceAuthSettings.from_env() + assert auth.audience == "patbond-ai" + assert auth.issuer == "patbond-api" + assert auth.required_type == "service" diff --git a/tests/test_layering.py b/tests/test_layering.py new file mode 100644 index 0000000..8c76e29 --- /dev/null +++ b/tests/test_layering.py @@ -0,0 +1,127 @@ +"""分层边界的测试。 + +这些断言不测业务，测的是**架构约束本身**。分层的价值全在「换供应商不动上层」， +而这个性质只要有人无意加一行 import 就会悄悄失效——没有测试守着，它迟早会失效。 +""" + +from __future__ import annotations + +import ast +import pathlib + +APP = pathlib.Path(__file__).resolve().parent.parent / "app" + + +def imports_of(path: pathlib.Path) -> set[str]: + tree = ast.parse(path.read_text(encoding="utf-8")) + names: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + names.update(alias.name for alias in node.names) + elif isinstance(node, ast.ImportFrom) and node.module: + names.add(node.module) + return names + + +def modules_under(*parts: str) -> list[pathlib.Path]: + return sorted((APP.joinpath(*parts)).rglob("*.py")) + + +class TestDomainIsIndependent: + """领域层是「换供应商不动上层」的支点。它一旦依赖厂商或 HTTP，分层就名存实亡。""" + + def test_domain_imports_no_provider_code(self): + for module in modules_under("domain"): + offenders = {name for name in imports_of(module) if "providers" in name} + assert not offenders, f"{module.name} 不应依赖服务商代码：{offenders}" + + def test_domain_imports_no_http_library(self): + for module in modules_under("domain"): + offenders = { + name + for name in imports_of(module) + if name.split(".")[0] in {"httpx", "requests"} + } + assert not offenders, f"{module.name} 不应依赖 HTTP 库：{offenders}" + + def test_domain_does_not_read_configuration(self): + # 领域对象应当是纯数据；让它自己读配置会使单测必须摆弄环境变量。 + for module in modules_under("domain"): + offenders = { + name for name in imports_of(module) if name in {"os", "app.config"} + } + assert not offenders, f"{module.name} 不应读取配置：{offenders}" + + +class TestAbstractionDoesNotLeak: + def test_provider_base_knows_no_specific_vendor(self): + """抽象接口若 import 了火山，新增第二家供应商就得改抽象层。""" + offenders = { + name + for name in imports_of(APP / "providers" / "base.py") + if "volcengine" in name + } + assert not offenders, f"抽象层不应依赖具体服务商：{offenders}" + + def test_registry_knows_no_specific_vendor(self): + offenders = { + name + for name in imports_of(APP / "providers" / "registry.py") + if "volcengine" in name + } + assert not offenders, f"注册表不应依赖具体服务商：{offenders}" + + +class TestTransportStaysBusinessAgnostic: + def test_ark_client_does_not_import_video_business(self): + """传输层复用的前提是它不认识「视频」。 + + 一旦 client.py 依赖了 video.py，接火山的图片接口时就没法复用它， + 只能再抄一遍鉴权与错误解包——那正是拆分要避免的。 + """ + offenders = { + name + for name in imports_of(APP / "providers" / "volcengine" / "client.py") + if name.endswith((".video", ".image")) or name.endswith("models") + } + assert not offenders, f"传输层不应依赖具体业务模块：{offenders}" + + +class TestVendorVocabularyIsContained: + def test_only_the_volcengine_package_mentions_ark_paths(self): + """厂商特有的路径、字段名只应出现在其适配包内。 + + `framespersecond` 这种厂商拼法漏到上层，就意味着上层已经知道自己在跟 + 火山说话——换供应商时会牵连到它。 + """ + vendor_markers = ("framespersecond", "contents/generations", "ark.cn-beijing") + for module in sorted(APP.rglob("*.py")): + if "volcengine" in module.parts: + continue + text = module.read_text(encoding="utf-8") + for marker in vendor_markers: + assert marker not in text, ( + f"{module.relative_to(APP)} 出现了厂商专有词汇 {marker!r}，" + f"应收拢到 providers/volcengine/ 内" + ) + + +class TestDependencyDirectionIsOneWay: + """配置 → 适配层，绝不反向。 + + `config.py` 作为装配根知道有哪家供应商是合理的；反过来，适配层一旦自己去读 + 配置，依赖注入就失效了——单测又得摆弄环境变量，而且同一进程内没法跑两套 + 不同配置的适配器实例。 + """ + + def test_providers_do_not_read_config(self): + for module in modules_under("providers"): + offenders = { + name + for name in imports_of(module) + if name == "app.config" or name == "os" + } + assert not offenders, ( + f"{module.relative_to(APP)} 不应自行读取配置或环境变量：" + f"{offenders}。请通过构造参数注入。" + ) diff --git a/tests/test_seedance.py b/tests/test_seedance.py deleted file mode 100644 index c88cf69..0000000 --- a/tests/test_seedance.py +++ /dev/null @@ -1,341 +0,0 @@ -"""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="")