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

"""火山方舟内容生成接口的请求/响应 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"),
        )