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

from datetime import datetime, timezone
from uuid import uuid4

from app.api.routes.permissions import require_workspace_member
from app.auth import AuthenticatedUser, get_current_user
from app.dependencies import (
    get_asset_repository,
    get_db_session,
    get_project_repository,
    get_workspace_member_repository,
)
from app.schemas.edit_plan import (
    AutoGenerateEditPlanRequest,
    CreateEditPlanRequest,
    EditPlanClipResponse,
    EditPlanResponse,
    EditTemplateResponse,
)
from packages.domain.edit_plan import EditingMode, SmartEditPlanGenerator
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session

from packages.adapters.sqlalchemy_impl.models import EditPlanClipModel, EditPlanModel, EditTemplateModel
from packages.domain import AssetStatus
from packages.ports.workspace_member_repository import WorkspaceMemberRepository

router = APIRouter(prefix="/projects/{project_id}/edit-plans", tags=["剪辑计划"])


def _ensure_project(project_id: str, workspace_id: str, user: AuthenticatedUser, project_repository, member_repository):
    project = project_repository.find_by_id(project_id)
    if project is None or project.workspace_id != workspace_id:
        raise HTTPException(status_code=404, detail="Project not found")
    require_workspace_member(workspace_id, user, member_repository)
    return project


def _default_template(session: Session, workspace_id: str, project_id: str, user_id: str) -> EditTemplateModel:
    template = (
        session.query(EditTemplateModel)
        .filter(
            EditTemplateModel.project_id == project_id,
            EditTemplateModel.is_active.is_(True),
        )
        .order_by(EditTemplateModel.created_at.asc())
        .first()
    )
    if template is not None:
        return template
    template = EditTemplateModel(
        id=uuid4().hex,
        workspace_id=workspace_id,
        project_id=project_id,
        name="基础节奏模板",
        description="自动选择可用视频素材，按上传顺序生成三段式剪辑计划。",
        target_duration=30,
        clip_count=3,
        created_by_user_id=user_id,
    )
    session.add(template)
    session.commit()
    return template


def _to_template_response(template: EditTemplateModel) -> EditTemplateResponse:
    return EditTemplateResponse(
        id=template.id,
        workspace_id=template.workspace_id,
        project_id=template.project_id,
        name=template.name,
        description=template.description,
        target_duration=float(template.target_duration or 0),
        clip_count=int(template.clip_count or 0),
        is_active=bool(template.is_active),
        created_at=template.created_at,
    )


def _to_plan_response(
    plan: EditPlanModel, clips: list[EditPlanClipModel], asset_names: dict[str, str]
) -> EditPlanResponse:
    return EditPlanResponse(
        id=plan.id,
        workspace_id=plan.workspace_id,
        project_id=plan.project_id,
        template_id=plan.template_id,
        asset_library_id=plan.asset_library_id,
        title_id=plan.title_id,
        status=plan.status,
        summary=plan.summary,
        editing_mode=plan.editing_mode,
        clips=[
            EditPlanClipResponse(
                id=clip.id,
                asset_id=clip.asset_id,
                asset_name=asset_names.get(clip.asset_id, clip.asset_id),
                sequence=clip.sequence,
                start_time=float(clip.start_time or 0),
                duration=float(clip.duration or 0),
                reason=clip.reason,
                layer=clip.layer,
            )
            for clip in clips
        ],
        created_at=plan.created_at,
        updated_at=plan.updated_at,
    )


@router.get("/templates/", response_model=list[EditTemplateResponse])
def list_edit_templates(
    project_id: str,
    workspace_id: str,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    project_repository=Depends(get_project_repository),
    workspace_member_repository: WorkspaceMemberRepository = Depends(get_workspace_member_repository),
    session: Session = Depends(get_db_session),
) -> list[EditTemplateResponse]:
    _ensure_project(project_id, workspace_id, authenticated_user, project_repository, workspace_member_repository)
    template = _default_template(session, workspace_id, project_id, authenticated_user.user.id)
    templates = (
        session.query(EditTemplateModel)
        .filter(EditTemplateModel.project_id == project_id, EditTemplateModel.is_active.is_(True))
        .all()
    )
    return [_to_template_response(item) for item in templates or [template]]


@router.post("", response_model=EditPlanResponse)
def create_edit_plan(
    project_id: str,
    request: CreateEditPlanRequest,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    project_repository=Depends(get_project_repository),
    asset_repository=Depends(get_asset_repository),
    workspace_member_repository: WorkspaceMemberRepository = Depends(get_workspace_member_repository),
    session: Session = Depends(get_db_session),
) -> EditPlanResponse:
    _ensure_project(
        project_id, request.workspace_id, authenticated_user, project_repository, workspace_member_repository
    )
    template = (
        session.query(EditTemplateModel).filter(EditTemplateModel.id == request.template_id).first()
        if request.template_id
        else None
    )
    if template is None:
        template = _default_template(session, request.workspace_id, project_id, authenticated_user.user.id)
    assets = [
        asset
        for asset in asset_repository.list_by_library(request.asset_library_id)
        if asset.status == AssetStatus.READY and asset.mime_type.startswith("video/")
    ]
    if not assets:
        raise HTTPException(status_code=422, detail="素材库暂无可用于剪辑计划的视频素材")
    selected = sorted(assets, key=lambda asset: (-(asset.quality_score or 0), asset.created_at))[
        : max(1, int(template.clip_count or 3))
    ]
    plan = EditPlanModel(
        id=uuid4().hex,
        workspace_id=request.workspace_id,
        project_id=project_id,
        template_id=template.id,
        asset_library_id=request.asset_library_id,
        title_id=request.title_id,
        status="draft",
        summary=f"按《{template.name}》自动选择 {len(selected)} 段素材，预计生成约 {int(template.target_duration or 30)} 秒成片。",
        created_by_user_id=authenticated_user.user.id,
        updated_at=datetime.now(timezone.utc),
    )
    session.add(plan)
    clips: list[EditPlanClipModel] = []
    clip_duration = max(1, float(template.target_duration or 30) / len(selected))
    for index, asset in enumerate(selected, start=1):
        clip = EditPlanClipModel(
            id=uuid4().hex,
            edit_plan_id=plan.id,
            asset_id=asset.id,
            sequence=index,
            start_time=0,
            duration=min(float(asset.duration or clip_duration), clip_duration),
            reason="优先选择已就绪、质量分较高的视频素材。",
        )
        session.add(clip)
        clips.append(clip)
    session.commit()
    return _to_plan_response(plan, clips, {asset.id: asset.name for asset in selected})


@router.post("/auto-generate", response_model=EditPlanResponse)
def auto_generate_edit_plan(
    project_id: str,
    request: AutoGenerateEditPlanRequest,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    project_repository=Depends(get_project_repository),
    asset_repository=Depends(get_asset_repository),
    workspace_member_repository: WorkspaceMemberRepository = Depends(get_workspace_member_repository),
    session: Session = Depends(get_db_session),
) -> EditPlanResponse:
    """
    智能生成剪辑计划

    根据素材的分类结果和质量评分，自动编排剪辑计划。
    支持多种剪辑模式：
    - one_take: 按分类分组，组内按质量排序，顺序拼接
    - pip: 第一个高质量素材为主画面，其余为画中画
    - voice_over: person 类素材为主播口播，其余穿插为 B-roll
    - voice_pip: 结合 voice_over 和 pip，第一个高质量 person 素材为主画面
    """
    _ensure_project(
        project_id, request.workspace_id, authenticated_user, project_repository, workspace_member_repository
    )

    # 获取素材库中的所有素材
    assets = asset_repository.list_by_library(request.asset_library_id)
    
    if not assets:
        raise HTTPException(status_code=422, detail="素材库中暂无素材")

    # 使用智能生成器
    generator = SmartEditPlanGenerator(project_id, assets)
    
    try:
        plan_result = generator.generate_plan(
            editing_mode=request.editing_mode,
            target_duration=request.target_duration
        )
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

    if not plan_result.clips:
        raise HTTPException(status_code=422, detail="无符合条件的视频素材")

    # 获取模板
    template = (
        session.query(EditTemplateModel).filter(EditTemplateModel.id == request.template_id).first()
        if request.template_id
        else None
    )
    if template is None:
        template = _default_template(session, request.workspace_id, project_id, authenticated_user.user.id)

    # 创建剪辑计划
    plan = EditPlanModel(
        id=uuid4().hex,
        workspace_id=request.workspace_id,
        project_id=project_id,
        template_id=template.id,
        asset_library_id=request.asset_library_id,
        title_id=request.title_id or "",
        status="draft",
        editing_mode=request.editing_mode,
        summary=plan_result.summary,
        created_by_user_id=authenticated_user.user.id,
        updated_at=datetime.now(timezone.utc),
    )
    session.add(plan)

    # 创建剪辑片段
    clips: list[EditPlanClipModel] = []
    asset_name_map = {asset.id: asset.name for asset in assets}
    
    for clip_plan in plan_result.clips:
        clip = EditPlanClipModel(
            id=uuid4().hex,
            edit_plan_id=plan.id,
            asset_id=clip_plan.asset_id,
            sequence=clip_plan.sequence,
            start_time=clip_plan.start_time,
            duration=clip_plan.duration,
            reason=clip_plan.reason,
            layer=clip_plan.layer,
        )
        session.add(clip)
        clips.append(clip)

    session.commit()

    return _to_plan_response(plan, clips, asset_name_map)


@router.get("/{plan_id}", response_model=EditPlanResponse)
def get_edit_plan(
    project_id: str,
    plan_id: str,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    project_repository=Depends(get_project_repository),
    asset_repository=Depends(get_asset_repository),
    workspace_member_repository: WorkspaceMemberRepository = Depends(get_workspace_member_repository),
    session: Session = Depends(get_db_session),
) -> EditPlanResponse:
    plan = (
        session.query(EditPlanModel).filter(EditPlanModel.id == plan_id, EditPlanModel.project_id == project_id).first()
    )
    if plan is None:
        raise HTTPException(status_code=404, detail="Edit plan not found")
    _ensure_project(project_id, plan.workspace_id, authenticated_user, project_repository, workspace_member_repository)
    clips = (
        session.query(EditPlanClipModel)
        .filter(EditPlanClipModel.edit_plan_id == plan.id)
        .order_by(EditPlanClipModel.sequence.asc())
        .all()
    )
    assets = asset_repository.list_by_library(plan.asset_library_id)
    return _to_plan_response(plan, clips, {asset.id: asset.name for asset in assets})