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

"""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