xiaoxia-saas/apps/api/app/api/routes/edit_plans_cover.py

"""封面管理 API.

- GET    /{plan_id}/cover              获取封面配置
- PUT    /{plan_id}/cover              更新封面配置
- POST   /{plan_id}/cover/extract      从指定片段抽帧生成封面
- POST   /{plan_id}/cover/smart        智能选帧生成封面
"""

from __future__ import annotations

import logging
from typing import Any, Optional

from app.auth import AuthenticatedUser, get_current_user
from app.core.storage import get_storage_service
from app.dependencies import (
    get_asset_repository,
    get_db_session,
    get_project_repository,
)
from app.services import EditPlanService
from app.services.cover_service import CoverService
from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session

from packages.domain.config_schemas import normalize_plan_config

from ._helpers import check_project_access

logger = logging.getLogger(__name__)

router = APIRouter()


# ── Schemas ──────────────────────────────────────────────────────────────────


class CoverConfigResponse(BaseModel):
    """封面配置响应"""

    type: str = Field(..., description="封面类型: ai_frame / manual / upload")
    image_url: str = Field(default="", description="封面图片 URL")
    frame_time: Optional[float] = Field(default=None, description="抽帧时间点（秒）")


class CoverUpdateRequest(BaseModel):
    """更新封面配置请求"""

    type: Optional[str] = Field(default=None, description="封面类型")
    image_url: Optional[str] = Field(default=None, description="封面图片 URL")
    frame_time: Optional[float] = Field(default=None, ge=0.0, description="抽帧时间点（秒）")


class CoverExtractRequest(BaseModel):
    """从片段抽帧生成封面请求"""

    clip_id: str = Field(..., description="片段 ID")
    frame_time: float = Field(1.0, ge=0.0, description="抽帧时间点（秒）")


class CoverSmartRequest(BaseModel):
    """智能选帧请求"""

    clip_id: Optional[str] = Field(default=None, description="指定片段 ID（不传则用第一个视频片段）")


class CoverGenerateResponse(BaseModel):
    """封面生成响应"""

    type: str = Field(..., description="封面类型")
    image_url: str = Field(..., description="封面图片 URL")
    frame_time: Optional[float] = Field(default=None, description="抽帧时间点（秒）")


# ── Routes ───────────────────────────────────────────────────────────────────


@router.get("/{plan_id}/cover", response_model=CoverConfigResponse)
def get_cover(
    plan_id: str,
    db: Session = Depends(get_db_session),
    current_user: AuthenticatedUser = Depends(get_current_user),
    project_repository: Any = Depends(get_project_repository),
) -> CoverConfigResponse:
    """获取封面配置"""
    svc = EditPlanService(db)
    plan = svc.get_plan(plan_id)
    if not plan:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"剪辑计划不存在: {plan_id}",
        )

    if plan.project_id:
        check_project_access(plan.project_id, current_user.user.id, project_repository)

    cover = CoverService.get_cover_config(plan.config or {})
    return CoverConfigResponse(**cover)


@router.put("/{plan_id}/cover", response_model=CoverConfigResponse)
def update_cover(
    plan_id: str,
    body: CoverUpdateRequest,
    db: Session = Depends(get_db_session),
    current_user: AuthenticatedUser = Depends(get_current_user),
    project_repository: Any = Depends(get_project_repository),
) -> CoverConfigResponse:
    """更新封面配置

    用于：设置上传的封面图片 URL、切换封面类型、调整时间点等。
    """
    svc = EditPlanService(db)
    plan = svc.get_plan(plan_id)
    if not plan:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"剪辑计划不存在: {plan_id}",
        )

    if plan.project_id:
        check_project_access(plan.project_id, current_user.user.id, project_repository)

    # 合并更新
    current_cover = CoverService.get_cover_config(plan.config or {})
    updates = body.model_dump(exclude_none=True)
    new_cover = {**current_cover, **updates}

    # 验证 type 值
    valid_types = {"ai_frame", "manual", "upload", "ai_regenerate"}
    if "type" in updates and updates["type"] not in valid_types:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"无效的封面类型: {updates['type']}，有效值: {valid_types}",
        )

    # 更新到 plan.config
    current_config = dict(plan.config or {})
    current_config["cover"] = new_cover
    normalized = normalize_plan_config(current_config)
    updated_plan = svc.update_plan_config(plan_id, {"cover": normalized["cover"]})

    result = CoverService.get_cover_config(updated_plan.config or {})
    logger.info("更新封面配置: plan_id=%s type=%s by user=%s", plan_id, result["type"], current_user.user.id)
    return CoverConfigResponse(**result)


@router.post("/{plan_id}/cover/extract", response_model=CoverGenerateResponse)
def extract_cover(
    plan_id: str,
    body: CoverExtractRequest,
    db: Session = Depends(get_db_session),
    current_user: AuthenticatedUser = Depends(get_current_user),
    project_repository: Any = Depends(get_project_repository),
    storage_service: Any = Depends(get_storage_service),
    asset_repository: Any = Depends(get_asset_repository),
) -> CoverGenerateResponse:
    """从指定片段的指定时间点抽帧生成封面"""
    svc = EditPlanService(db)
    plan = svc.get_plan(plan_id)
    if not plan:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"剪辑计划不存在: {plan_id}",
        )

    if plan.project_id:
        check_project_access(plan.project_id, current_user.user.id, project_repository)

    # 获取片段对应的素材
    clip = svc.get_clip(body.clip_id)
    if not clip:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"片段不存在: {body.clip_id}",
        )

    if clip.plan_id != plan_id:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="片段不属于该剪辑计划",
        )

    if not clip.asset_id:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="片段没有关联素材，无法抽帧",
        )

    # 抽帧生成封面
    cover_svc = CoverService(storage_service, asset_repository)
    try:
        cover_data = cover_svc.extract_cover_from_clip(
            plan_id=plan_id,
            asset_id=clip.asset_id,
            frame_time=body.frame_time,
        )
    except ValueError as e:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e
    except RuntimeError as e:
        logger.error("封面抽帧失败: plan_id=%s clip_id=%s error=%s", plan_id, body.clip_id, e)
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"封面抽帧失败: {e}",
        ) from e

    # 更新到 plan.config.cover
    current_config = dict(plan.config or {})
    current_config["cover"] = cover_data
    normalized = normalize_plan_config(current_config)
    svc.update_plan_config(plan_id, {"cover": normalized["cover"]})

    logger.info(
        "封面抽帧完成: plan_id=%s clip_id=%s time=%.2fs by user=%s",
        plan_id,
        body.clip_id,
        body.frame_time,
        current_user.user.id,
    )
    return CoverGenerateResponse(**cover_data)


@router.post("/{plan_id}/cover/smart", response_model=CoverGenerateResponse)
def smart_cover(
    plan_id: str,
    body: CoverSmartRequest,
    db: Session = Depends(get_db_session),
    current_user: AuthenticatedUser = Depends(get_current_user),
    project_repository: Any = Depends(get_project_repository),
    storage_service: Any = Depends(get_storage_service),
    asset_repository: Any = Depends(get_asset_repository),
) -> CoverGenerateResponse:
    """智能选帧生成封面

    从指定片段（或第一个视频片段）中智能选取一帧作为封面。
    当前实现：取片段第3秒帧（后续可优化为多帧选最清晰）。
    """
    svc = EditPlanService(db)
    plan = svc.get_plan(plan_id)
    if not plan:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"剪辑计划不存在: {plan_id}",
        )

    if plan.project_id:
        check_project_access(plan.project_id, current_user.user.id, project_repository)

    # 确定使用哪个片段
    clip_id = body.clip_id
    asset_id = ""

    if clip_id:
        clip = svc.get_clip(clip_id)
        if not clip:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"片段不存在: {clip_id}",
            )
        if clip.plan_id != plan_id:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="片段不属于该剪辑计划",
            )
        if not clip.asset_id:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="片段没有关联素材",
            )
        asset_id = clip.asset_id
    else:
        # 找第一个有素材的视频片段
        clips = svc.list_clips(plan_id, limit=50, skip=0)
        for c in clips:
            if c.asset_id and c.clip_type == "video":
                asset_id = c.asset_id
                clip_id = c.id
                break

        if not asset_id:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="没有找到可用的视频片段",
            )

    # 智能选帧
    cover_svc = CoverService(storage_service, asset_repository)
    try:
        cover_data = cover_svc.generate_smart_cover(
            plan_id=plan_id,
            asset_id=asset_id,
        )
    except ValueError as e:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e
    except RuntimeError as e:
        logger.error("智能封面生成失败: plan_id=%s error=%s", plan_id, e)
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"智能封面生成失败: {e}",
        ) from e

    # 更新到 plan.config.cover
    current_config = dict(plan.config or {})
    current_config["cover"] = cover_data
    normalized = normalize_plan_config(current_config)
    svc.update_plan_config(plan_id, {"cover": normalized["cover"]})

    logger.info(
        "智能封面生成完成: plan_id=%s clip_id=%s by user=%s",
        plan_id,
        clip_id,
        current_user.user.id,
    )
    return CoverGenerateResponse(**cover_data)