ozon-seller-kit/server/api/video.py

"""商品图片轮播视频：复用后台任务状态机，FFmpeg 在独立 Worker 中执行。"""
from __future__ import annotations

import uuid
from pathlib import Path

from fastapi import APIRouter, HTTPException
from fastapi.responses import FileResponse, RedirectResponse
from sqlalchemy import select

from db import get_session_factory
from models import Product, ProductAsset
from models.enums import AssetStatus
from schemas.video import VideoCreateRequest, VideoCreateResponse, VideoDeleteResponse, VideoJobOut
from services.jobs import create_image_job, delete_job, get_job, list_product_jobs, retry_failed_items, video_job_view
from services.storage import get_storage
from services.storage import local_path
from services.video_renderer import slide_seconds_for

router = APIRouter(prefix="/api", tags=["video"])
VIDEO_JOB_KIND = "slideshow_video"


def _job_out(result) -> VideoJobOut:
    return VideoJobOut(**video_job_view(*result))


@router.post("/products/{product_id}/video-jobs", response_model=VideoCreateResponse)
async def create_video_job(product_id: str, req: VideoCreateRequest) -> VideoCreateResponse:
    try:
        product_uuid = uuid.UUID(product_id)
        asset_ids = [uuid.UUID(asset_id) for asset_id in req.asset_ids]
    except ValueError as exc:
        raise HTTPException(status_code=400, detail="商品或图片 ID 不合法") from exc

    async with get_session_factory()() as db:
        product = await db.get(Product, product_uuid)
        if product is None:
            raise HTTPException(status_code=404, detail="商品不存在")
        assets = (await db.scalars(
            select(ProductAsset).where(
                ProductAsset.product_id == product_uuid,
                ProductAsset.id.in_(asset_ids),
                ProductAsset.type == "img",
                ProductAsset.status == AssetStatus.uploaded,
            )
        )).all()
    by_id = {asset.id: asset for asset in assets}
    ordered = [by_id.get(asset_id) for asset_id in asset_ids]
    if any(asset is None or not asset.stored_url for asset in ordered):
        raise HTTPException(status_code=400, detail="所选图片必须属于当前商品且已完成转存")

    seconds = slide_seconds_for(len(ordered))
    task = await create_image_job(
        product_id=product_id,
        kind=VIDEO_JOB_KIND,
        priority=5,
        provider="ffmpeg",
        model=None,
        platform="wb_ozon",
        lang="ru",
        ratio="3:4",
        style_set=0,
        request_payload={
            "preset": "wb_ozon",
            "slide_seconds": seconds,
            "sources": [
                {"asset_id": str(asset.id), "url": asset.stored_url, "storage_key": asset.current_storage_key}
                for asset in ordered if asset is not None
            ],
        },
        item_payloads=[{"title": "商品轮播视频", "kind": VIDEO_JOB_KIND}],
    )
    return VideoCreateResponse(job_id=str(task.id))


@router.get("/products/{product_id}/video-jobs", response_model=list[VideoJobOut])
async def list_video_jobs(product_id: str, active_only: bool = False):
    jobs = await list_product_jobs(product_id, active_only=active_only, kinds=(VIDEO_JOB_KIND,))
    views: list[VideoJobOut] = []
    for job in jobs:
        result = await get_job(str(job.id))
        if result is not None:
            views.append(_job_out(result))
    return views


@router.get("/video-jobs/{job_id}", response_model=VideoJobOut)
async def get_video_job(job_id: str):
    result = await get_job(job_id)
    if result is None or result[0].kind != VIDEO_JOB_KIND:
        raise HTTPException(status_code=404, detail="视频任务不存在")
    return _job_out(result)


@router.post("/video-jobs/{job_id}/cancel", response_model=VideoDeleteResponse)
async def cancel_video_job(job_id: str) -> VideoDeleteResponse:
    # 视频任务不像生图任务需要保留诊断历史：取消后立即从页面和数据库消失。
    return await _purge_video_job(job_id)


@router.post("/video-jobs/{job_id}/retry", response_model=VideoJobOut)
async def retry_video_job(job_id: str):
    job = await retry_failed_items(job_id)
    result = await get_job(str(job.id)) if job and job.kind == VIDEO_JOB_KIND else None
    if result is None:
        raise HTTPException(status_code=404, detail="视频任务不存在")
    return _job_out(result)


async def _purge_video_job(job_id: str) -> VideoDeleteResponse:
    result = await get_job(job_id)
    if result is None or result[0].kind != VIDEO_JOB_KIND:
        raise HTTPException(status_code=404, detail="视频任务不存在")
    job, items = result
    # 先删该任务专属目录；即使 FFmpeg 正在运行，Worker 下一次检查也会发现任务不存在并终止。
    if job.product_id:
        await get_storage().delete_prefix(f"products/{job.product_id}/videos/{job.id}/")
    if not await delete_job(job_id):
        raise HTTPException(status_code=404, detail="视频任务不存在")
    return VideoDeleteResponse()


@router.delete("/video-jobs/{job_id}", response_model=VideoDeleteResponse)
async def delete_video_job(job_id: str) -> VideoDeleteResponse:
    """删除完成/失败视频及其 MP4；任务记录不保留。"""
    return await _purge_video_job(job_id)


@router.get("/video-jobs/{job_id}/download")
async def download_video(job_id: str):
    result = await get_job(job_id)
    if result is None or result[0].kind != VIDEO_JOB_KIND or not result[1] or not result[1][0].result_url:
        raise HTTPException(status_code=404, detail="视频文件不存在")
    url = result[1][0].result_url
    path = local_path(url)
    filename = f"product-video-{job_id}.mp4"
    if path is not None and Path(path).is_file():
        return FileResponse(path, media_type="video/mp4", filename=filename)
    return RedirectResponse(url, status_code=307)