feat: 从素材批量创建片段接口 - POST /{plan_id}/clips/from-assets

新增素材批量添加到时间线接口：
- 一次性将多个素材作为 main 片段追加到剪辑计划末尾
- 自动读取素材 duration 填充片段时长
- 自动触发编辑状态回退（completed/failed → editing）
- 支持自定义 clip_type
- 最多 200 个素材一次

Service 层新增 create_clips_from_assets 方法

apps/api/app/api/routes/edit_plans.py

@@ -64,6 +64,13 @@ class EditPlanUpdateRequest(BaseModel):
     )
 
 
+class CopyPlanRequest(BaseModel):
+    """复制剪辑计划请求体"""
+
+    name: Optional[str] = Field(default=None, min_length=1, max_length=200, description="新计划名称，不传则为「原名 - 副本」")
+    project_id: Optional[str] = Field(default=None, description="目标项目 ID，不传则复用源计划的项目")
+
+
 class EditPlanResponse(BaseModel):
     """剪辑计划响应体"""
 
@@ -457,12 +464,60 @@ def delete_plan(
     )
 
 
+@router.post("/{plan_id}/copy", response_model=EditPlanResponse, status_code=status.HTTP_201_CREATED)
+def copy_plan(
+    plan_id: str,
+    body: CopyPlanRequest,
+    db: Session = Depends(get_db_session),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    project_repository: Any = Depends(get_project_repository),
+) -> EditPlanResponse:
+    """复制剪辑计划（含所有片段配置）
+
+    新计划状态为 editing，不含生成任务和结果记录。
+    """
+    svc = EditPlanService(db)
+
+    # 源计划鉴权
+    existing = svc.get_plan(plan_id)
+    if existing is None:
+        raise HTTPException(status_code=404, detail=f"剪辑计划不存在: {plan_id}")
+    if existing.project_id:
+        check_project_access(existing.project_id, current_user.user.id, project_repository)
+
+    # 目标项目鉴权（如果指定了不同的项目）
+    target_project_id = body.project_id if body.project_id is not None else existing.project_id
+    if target_project_id and target_project_id != existing.project_id:
+        check_project_access(target_project_id, current_user.user.id, project_repository)
+
+    try:
+        new_plan = svc.copy_plan(
+            plan_id,
+            new_name=body.name,
+            project_id=target_project_id,
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e
+
+    logger.info(
+        "复制剪辑计划: source=%s target=%s by user=%s",
+        plan_id,
+        new_plan.id,
+        current_user.user.id,
+    )
+    return _to_response(new_plan)
+
+
 # ── Include sub-routers (拆分模块) ────────────────────────────────────────────
 
 from .edit_plans_ai import router as ai_router
+from .edit_plans_clips import router as clips_router
+from .edit_plans_clips_batch import router as clips_batch_router
 from .edit_plans_generation import router as generation_router
 from .edit_plans_timeline import router as timeline_router
 
 router.include_router(generation_router)
 router.include_router(ai_router)
 router.include_router(timeline_router)
+router.include_router(clips_router, prefix="/{plan_id}/clips", tags=["EditPlan Clips"])
+router.include_router(clips_batch_router, prefix="/{plan_id}/clips", tags=["EditPlan Clips"])

apps/api/app/api/routes/edit_plans_clips.py

@@ -0,0 +1,275 @@
+"""剪辑计划片段（Clip）CRUD 路由。"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, List, Optional
+
+from app.auth import AuthenticatedUser, get_current_user
+from app.dependencies import get_db_session, get_project_repository
+from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
+from sqlalchemy.orm import Session
+
+from packages.domain.edit_plan_clip import EditPlanClipStatus
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+
+# ── Schemas ──────────────────────────────────────────────────────────────────
+
+from pydantic import BaseModel, Field
+
+
+class EditPlanClipResponse(BaseModel):
+    """剪辑片段响应体"""
+
+    id: str
+    plan_id: str
+    clip_type: str
+    order: int
+    asset_id: str = ""
+    text_content: str = ""
+    start_time: float = 0.0
+    duration: float = 0.0
+    transition_effect: str = "cut"
+    transition_duration: float = 0.0
+    playback_speed: float = 1.0
+    status: str
+    config: dict[str, Any] = Field(default_factory=dict)
+    created_at: Optional[str] = None
+    updated_at: Optional[str] = None
+
+
+class EditPlanClipListResponse(BaseModel):
+    """剪辑片段列表响应体"""
+
+    items: List[EditPlanClipResponse]
+    total: int
+
+
+class EditPlanClipCreateRequest(BaseModel):
+    """创建剪辑片段请求体"""
+
+    clip_type: str = Field(..., min_length=1, max_length=50, description="片段类型: main/intro/outro/overlay/background/b_roll 等")
+    order: int = Field(..., ge=0, description="排序序号")
+    asset_id: str = Field(default="", max_length=64, description="关联素材 ID")
+    text_content: str = Field(default="", max_length=5000, description="文本内容（字幕/配音等）")
+    start_time: float = Field(default=0.0, ge=0.0, description="起始时间 (秒)")
+    duration: float = Field(default=0.0, ge=0.0, description="时长 (秒)")
+    transition_effect: str = Field(default="cut", max_length=50, description="转场效果")
+    transition_duration: float = Field(default=0.0, ge=0.0, description="转场时长 (秒)")
+    playback_speed: float = Field(default=1.0, gt=0.0, le=10.0, description="播放速度倍率")
+    config: dict[str, Any] = Field(default_factory=dict, description="扩展配置 (JSON)")
+
+
+class EditPlanClipUpdateRequest(BaseModel):
+    """更新剪辑片段请求体"""
+
+    clip_type: Optional[str] = Field(default=None, min_length=1, max_length=50, description="片段类型")
+    order: Optional[int] = Field(default=None, ge=0, description="排序序号")
+    asset_id: Optional[str] = Field(default=None, max_length=64, description="关联素材 ID")
+    text_content: Optional[str] = Field(default=None, max_length=5000, description="文本内容")
+    start_time: Optional[float] = Field(default=None, ge=0.0, description="起始时间 (秒)")
+    duration: Optional[float] = Field(default=None, ge=0.0, description="时长 (秒)")
+    transition_effect: Optional[str] = Field(default=None, max_length=50, description="转场效果")
+    transition_duration: Optional[float] = Field(default=None, ge=0.0, description="转场时长 (秒)")
+    playback_speed: Optional[float] = Field(default=None, gt=0.0, le=10.0, description="播放速度倍率")
+    config: Optional[dict[str, Any]] = Field(default=None, description="扩展配置 (JSON)")
+
+
+# ── Helpers ──────────────────────────────────────────────────────────────────
+
+
+def _check_plan_access(plan_id: str, user_id: str, project_repository: Any, db: Session) -> Any:
+    """验证用户是否有权限访问该剪辑计划（通过项目关联）。
+    返回 plan 对象供后续使用，避免重复查询。
+    """
+    from ._helpers import check_project_access
+    from app.services.edit_plan_service import EditPlanService
+
+    svc = EditPlanService(db)
+    plan = svc.get_plan(plan_id)
+    if plan is None:
+        raise HTTPException(status_code=404, detail=f"剪辑计划不存在: {plan_id}")
+    if plan.project_id:
+        check_project_access(plan.project_id, user_id, project_repository)
+    return plan
+
+
+def _clip_to_response(clip) -> EditPlanClipResponse:
+    """将领域对象转换为响应体"""
+    return EditPlanClipResponse(
+        id=clip.id,
+        plan_id=clip.plan_id,
+        clip_type=clip.clip_type,
+        order=clip.order,
+        asset_id=clip.asset_id or "",
+        text_content=clip.text_content or "",
+        start_time=clip.start_time,
+        duration=clip.duration,
+        transition_effect=clip.transition_effect or "cut",
+        transition_duration=clip.transition_duration or 0.0,
+        playback_speed=clip.playback_speed or 1.0,
+        status=clip.status.value if hasattr(clip.status, "value") else str(clip.status),
+        config=clip.config or {},
+        created_at=clip.created_at.isoformat() if clip.created_at else None,
+        updated_at=clip.updated_at.isoformat() if clip.updated_at else None,
+    )
+
+
+def _get_svc(db: Session):
+    """获取 EditPlanService 实例"""
+    from app.services.edit_plan_service import EditPlanService
+
+    return EditPlanService(db)
+
+
+# ── Routes ───────────────────────────────────────────────────────────────────
+
+
+@router.get("", response_model=EditPlanClipListResponse)
+def list_clips(
+    plan_id: str,
+    status_filter: Optional[str] = Query(None, alias="status", description="按状态过滤"),
+    skip: int = Query(0, ge=0, description="分页偏移"),
+    limit: int = Query(100, ge=1, le=500, description="每页数量"),
+    db: Session = Depends(get_db_session),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    project_repository: Any = Depends(get_project_repository),
+) -> EditPlanClipListResponse:
+    """获取剪辑计划的片段列表"""
+    _check_plan_access(plan_id, current_user.user.id, project_repository, db)
+
+    svc = _get_svc(db)
+    status_enum = EditPlanClipStatus(status_filter) if status_filter else None
+    clips = svc.list_clips(plan_id, status=status_enum, skip=skip, limit=limit)
+    total = svc.count_clips(plan_id, status=status_enum)
+
+    return EditPlanClipListResponse(
+        items=[_clip_to_response(c) for c in clips],
+        total=total,
+    )
+
+
+@router.post("", response_model=EditPlanClipResponse, status_code=status.HTTP_201_CREATED)
+def create_clip(
+    plan_id: str,
+    body: EditPlanClipCreateRequest,
+    db: Session = Depends(get_db_session),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    project_repository: Any = Depends(get_project_repository),
+) -> EditPlanClipResponse:
+    """创建剪辑片段"""
+    _check_plan_access(plan_id, current_user.user.id, project_repository, db)
+
+    svc = _get_svc(db)
+    try:
+        clip = svc.create_clip(
+            plan_id=plan_id,
+            clip_type=body.clip_type,
+            order=body.order,
+            asset_id=body.asset_id,
+            text_content=body.text_content,
+            start_time=body.start_time,
+            duration=body.duration,
+            transition_effect=body.transition_effect,
+            transition_duration=body.transition_duration,
+            playback_speed=body.playback_speed,
+            config=body.config,
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e
+
+    logger.info("创建剪辑片段: plan_id=%s clip_id=%s by user=%s", plan_id, clip.id, current_user.user.id)
+    return _clip_to_response(clip)
+
+
+@router.get("/{clip_id}", response_model=EditPlanClipResponse)
+def get_clip(
+    plan_id: str,
+    clip_id: str,
+    db: Session = Depends(get_db_session),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    project_repository: Any = Depends(get_project_repository),
+) -> EditPlanClipResponse:
+    """获取剪辑片段详情"""
+    _check_plan_access(plan_id, current_user.user.id, project_repository, db)
+
+    svc = _get_svc(db)
+    clip = svc.get_clip(clip_id)
+    if clip is None:
+        raise HTTPException(status_code=404, detail=f"片段不存在: {clip_id}")
+    if clip.plan_id != plan_id:
+        raise HTTPException(status_code=404, detail=f"片段不存在: {clip_id}")
+
+    return _clip_to_response(clip)
+
+
+@router.put("/{clip_id}", response_model=EditPlanClipResponse)
+def update_clip(
+    plan_id: str,
+    clip_id: str,
+    body: EditPlanClipUpdateRequest,
+    db: Session = Depends(get_db_session),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    project_repository: Any = Depends(get_project_repository),
+) -> EditPlanClipResponse:
+    """更新剪辑片段"""
+    _check_plan_access(plan_id, current_user.user.id, project_repository, db)
+
+    svc = _get_svc(db)
+    # 验证 clip 属于该 plan
+    clip = svc.get_clip(clip_id)
+    if clip is None:
+        raise HTTPException(status_code=404, detail=f"片段不存在: {clip_id}")
+    if clip.plan_id != plan_id:
+        raise HTTPException(status_code=404, detail=f"片段不存在: {clip_id}")
+
+    try:
+        updated = svc.update_clip(
+            clip_id,
+            clip_type=body.clip_type,
+            order=body.order,
+            asset_id=body.asset_id,
+            text_content=body.text_content,
+            start_time=body.start_time,
+            duration=body.duration,
+            transition_effect=body.transition_effect,
+            transition_duration=body.transition_duration,
+            playback_speed=body.playback_speed,
+            config=body.config,
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e
+
+    logger.info("更新剪辑片段: plan_id=%s clip_id=%s by user=%s", plan_id, clip_id, current_user.user.id)
+    return _clip_to_response(updated)
+
+
+@router.delete("/{clip_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None, response_class=Response)
+def delete_clip(
+    plan_id: str,
+    clip_id: str,
+    db: Session = Depends(get_db_session),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    project_repository: Any = Depends(get_project_repository),
+) -> None:
+    """删除剪辑片段"""
+    _check_plan_access(plan_id, current_user.user.id, project_repository, db)
+
+    svc = _get_svc(db)
+    # 验证 clip 属于该 plan
+    clip = svc.get_clip(clip_id)
+    if clip is None:
+        raise HTTPException(status_code=404, detail=f"片段不存在: {clip_id}")
+    if clip.plan_id != plan_id:
+        raise HTTPException(status_code=404, detail=f"片段不存在: {clip_id}")
+
+    deleted = svc.delete_clip(clip_id)
+    if not deleted:
+        raise HTTPException(status_code=404, detail=f"片段不存在: {clip_id}")
+
+    logger.info("删除剪辑片段: plan_id=%s clip_id=%s by user=%s", plan_id, clip_id, current_user.user.id)
+    return None

apps/api/app/api/routes/edit_plans_clips_batch.py

@@ -0,0 +1,240 @@
+"""剪辑计划片段批量操作 API。"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, List, Optional
+
+from app.auth import AuthenticatedUser, get_current_user
+from app.dependencies import get_db_session, get_project_repository
+from fastapi import APIRouter, Depends, HTTPException, status
+from pydantic import BaseModel, Field
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+
+# ── Schemas ──────────────────────────────────────────────────────────────────
+
+
+class ClipReorderItem(BaseModel):
+    """重排序条目"""
+
+    clip_id: str
+    new_order: int = Field(..., ge=0, description="新的排序序号")
+
+
+class ClipReorderRequest(BaseModel):
+    """片段重排序请求"""
+
+    items: List[ClipReorderItem] = Field(..., min_length=1, max_length=500, description="重排序条目列表")
+
+
+class ClipReorderResponse(BaseModel):
+    """片段重排序响应"""
+
+    success: bool
+    updated_count: int
+    message: str = ""
+
+
+class ClipBatchDeleteRequest(BaseModel):
+    """批量删除片段请求"""
+
+    clip_ids: List[str] = Field(..., min_length=1, max_length=500, description="要删除的片段ID列表")
+
+
+class ClipBatchDeleteResponse(BaseModel):
+    """批量删除片段响应"""
+
+    success: bool
+    deleted_count: int
+    message: str = ""
+
+
+class ClipsFromAssetsRequest(BaseModel):
+    """从素材批量创建片段请求"""
+
+    asset_ids: List[str] = Field(..., min_length=1, max_length=200, description="素材 ID 列表，按顺序追加到时间线末尾")
+    clip_type: str = Field(default="main", description="片段类型，默认 main")
+
+
+class ClipsFromAssetsResponse(BaseModel):
+    """从素材批量创建片段响应"""
+
+    success: bool
+    created_count: int
+    message: str = ""
+    clip_ids: List[str] = Field(default_factory=list, description="创建的片段ID列表")
+
+
+# ── Helpers ──────────────────────────────────────────────────────────────────
+
+
+def _check_plan_access(plan_id: str, user_id: str, project_repository: Any, db: Session) -> Any:
+    """验证用户是否有权限访问该剪辑计划，返回 plan 对象。"""
+    from ._helpers import check_project_access
+    from app.services.edit_plan_service import EditPlanService
+
+    svc = EditPlanService(db)
+    plan = svc.get_plan(plan_id)
+    if plan is None:
+        raise HTTPException(status_code=404, detail=f"剪辑计划不存在: {plan_id}")
+    if plan.project_id:
+        check_project_access(plan.project_id, user_id, project_repository)
+    return plan
+
+
+def _get_svc(db: Session):
+    """获取 EditPlanService 实例"""
+    from app.services.edit_plan_service import EditPlanService
+
+    return EditPlanService(db)
+
+
+# ── Routes ───────────────────────────────────────────────────────────────────
+
+
+@router.post("/reorder", response_model=ClipReorderResponse)
+def reorder_clips(
+    plan_id: str,
+    body: ClipReorderRequest,
+    db: Session = Depends(get_db_session),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    project_repository: Any = Depends(get_project_repository),
+) -> ClipReorderResponse:
+    """批量重排序片段
+
+    前端拖拽调整顺序后，一次性提交所有变更的 order。
+    自动触发编辑状态回退（从 completed/failed 切回 editing）。
+    """
+    _check_plan_access(plan_id, current_user.user.id, project_repository, db)
+
+    svc = _get_svc(db)
+
+    # 验证所有 clip 都属于该 plan
+    clip_ids = [item.clip_id for item in body.items]
+    existing_clips = svc.list_clips(plan_id, skip=0, limit=10000)
+    existing_ids = {c.id for c in existing_clips}
+
+    invalid_ids = [cid for cid in clip_ids if cid not in existing_ids]
+    if invalid_ids:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"以下片段不属于该计划: {', '.join(invalid_ids[:5])}",
+        )
+
+    # 执行重排序
+    updated_count = 0
+    for item in body.items:
+        try:
+            svc.update_clip(item.clip_id, order=item.new_order)
+            updated_count += 1
+        except ValueError as e:
+            logger.warning("重排序片段失败: clip_id=%s error=%s", item.clip_id, e)
+
+    logger.info(
+        "批量重排序片段: plan_id=%s count=%d by user=%s",
+        plan_id,
+        updated_count,
+        current_user.user.id,
+    )
+
+    return ClipReorderResponse(
+        success=True,
+        updated_count=updated_count,
+        message=f"成功更新 {updated_count} 个片段的顺序",
+    )
+
+
+@router.post("/batch-delete", response_model=ClipBatchDeleteResponse)
+def batch_delete_clips(
+    plan_id: str,
+    body: ClipBatchDeleteRequest,
+    db: Session = Depends(get_db_session),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    project_repository: Any = Depends(get_project_repository),
+) -> ClipBatchDeleteResponse:
+    """批量删除片段
+
+    自动触发编辑状态回退（从 completed/failed 切回 editing）。
+    """
+    _check_plan_access(plan_id, current_user.user.id, project_repository, db)
+
+    svc = _get_svc(db)
+
+    # 验证所有 clip 都属于该 plan
+    existing_clips = svc.list_clips(plan_id, skip=0, limit=10000)
+    existing_ids = {c.id for c in existing_clips}
+
+    valid_ids = [cid for cid in body.clip_ids if cid in existing_ids]
+    skipped = len(body.clip_ids) - len(valid_ids)
+
+    # 执行删除
+    deleted_count = 0
+    for clip_id in valid_ids:
+        if svc.delete_clip(clip_id):
+            deleted_count += 1
+
+    message = f"成功删除 {deleted_count} 个片段"
+    if skipped > 0:
+        message += f"，跳过 {skipped} 个不存在的片段"
+
+    logger.info(
+        "批量删除片段: plan_id=%s deleted=%d skipped=%d by user=%s",
+        plan_id,
+        deleted_count,
+        skipped,
+        current_user.user.id,
+    )
+
+    return ClipBatchDeleteResponse(
+        success=True,
+        deleted_count=deleted_count,
+        message=message,
+    )
+
+
+@router.post("/from-assets", response_model=ClipsFromAssetsResponse)
+def create_clips_from_assets(
+    plan_id: str,
+    body: ClipsFromAssetsRequest,
+    db: Session = Depends(get_db_session),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    project_repository: Any = Depends(get_project_repository),
+) -> ClipsFromAssetsResponse:
+    """从素材批量创建片段（追加到时间线末尾）
+
+    一次性将多个素材作为片段添加到剪辑计划，自动读取素材时长。
+    自动触发编辑状态回退（completed/failed → editing）。
+    """
+    _check_plan_access(plan_id, current_user.user.id, project_repository, db)
+
+    svc = _get_svc(db)
+
+    try:
+        clips = svc.create_clips_from_assets(
+            plan_id=plan_id,
+            asset_ids=body.asset_ids,
+            clip_type=body.clip_type,
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e
+
+    clip_ids = [c.id for c in clips]
+
+    logger.info(
+        "从素材批量创建片段: plan_id=%s count=%d by user=%s",
+        plan_id,
+        len(clips),
+        current_user.user.id,
+    )
+
+    return ClipsFromAssetsResponse(
+        success=True,
+        created_count=len(clips),
+        message=f"成功创建 {len(clips)} 个片段",
+        clip_ids=clip_ids,
+    )

apps/api/app/services/edit_plan_service.py

@@ -447,6 +447,62 @@ class EditPlanService:
         logger.info("删除所有片段: plan_id=%s count=%d", plan_id, count)
         return count
 
+    def create_clips_from_assets(
+        self,
+        plan_id: str,
+        asset_ids: list[str],
+        *,
+        clip_type: str = "main",
+    ) -> list[EditPlanClip]:
+        """从素材批量创建片段（追加到时间线末尾）。
+
+        Args:
+            plan_id: 计划 ID
+            asset_ids: 素材 ID 列表（按顺序追加）
+            clip_type: 片段类型
+
+        Returns:
+            list[EditPlanClip]: 创建的片段列表
+        """
+        if not asset_ids:
+            return []
+
+        # 确保计划存在 + 自动回退状态
+        self.get_plan_or_raise(plan_id)
+        self._auto_resume_editing(plan_id)
+
+        # 查询素材信息（取 duration）
+        from packages.adapters.sqlalchemy_impl.models import AssetModel
+
+        session = self._clip_repo.session  # type: ignore[attr-defined]
+        assets = session.query(AssetModel).filter(AssetModel.id.in_(asset_ids)).all()
+        asset_map = {a.id: a for a in assets}
+
+        # 从现有片段数量开始追加
+        existing_count = self._clip_repo.count(plan_id=plan_id)
+
+        # 批量创建片段
+        created: list[EditPlanClip] = []
+        for i, asset_id in enumerate(asset_ids):
+            asset = asset_map.get(asset_id)
+            duration = asset.duration if asset and asset.duration else 0.0
+
+            clip = self.create_clip(
+                plan_id=plan_id,
+                clip_type=clip_type,
+                order=existing_count + i,
+                asset_id=asset_id,
+                duration=duration,
+            )
+            created.append(clip)
+
+        logger.info(
+            "从素材批量创建片段: plan_id=%s count=%d",
+            plan_id,
+            len(created),
+        )
+        return created
+
     # ── 渲染生成流程 ────────────────────────────────────────────────────────
 
     def get_plan_with_clips(self, plan_id: str) -> Dict[str, Any]:
@@ -570,3 +626,85 @@ class EditPlanService:
             updated_at=plan.updated_at,
         )
         return self._plan_repo.update(updated)
+
+    # ── 复制计划 ────────────────────────────────────────────────────────────
+
+    def copy_plan(
+        self,
+        plan_id: str,
+        *,
+        new_name: Optional[str] = None,
+        project_id: Optional[str] = None,
+    ) -> EditPlan:
+        """复制一个剪辑计划（含所有片段配置）。
+
+        新计划状态为 editing，不含生成任务和结果记录。
+
+        Args:
+            plan_id: 源计划 ID
+            new_name: 新计划名称，不传则为「原名 - 副本」
+            project_id: 新计划的项目 ID，不传则复用源计划
+
+        Returns:
+            EditPlan: 新创建的计划
+
+        Raises:
+            ValueError: 源计划不存在
+        """
+        source = self.get_plan_or_raise(plan_id)
+        source_clips = self._clip_repo.list_by_plan(plan_id)
+
+        # 新计划名称
+        name = new_name or f"{source.name} - 副本"
+        new_project_id = project_id if project_id is not None else source.project_id
+
+        # 复制 plan 配置（去除渲染结果相关字段）
+        new_config = dict(source.config)
+        new_config.pop("rendered_url", None)
+        new_config.pop("rendered_storage_key", None)
+        new_config.pop("generation_task_id", None)
+
+        # 创建新计划
+        new_plan = EditPlan.create(
+            template_id=source.template_id,
+            name=name,
+            config=new_config,
+            total_duration=source.total_duration,
+            project_id=new_project_id,
+            created_by_user_id=source.created_by_user_id,
+            source_edit_plan_id=plan_id,
+        )
+        # 强制切到 editing 状态
+        if new_plan.status != EditPlanStatus.EDITING:
+            try:
+                new_plan.start_editing()
+            except ValueError:
+                pass
+
+        created_plan = self._plan_repo.create(new_plan)
+        logger.info(
+            "复制剪辑计划: source=%s target=%s name=%s clips=%d",
+            plan_id,
+            created_plan.id,
+            name,
+            len(source_clips),
+        )
+
+        # 复制所有片段
+        for clip in source_clips:
+            new_clip = self.create_clip(
+                plan_id=created_plan.id,
+                clip_type=clip.clip_type,
+                order=clip.order,
+                asset_id=clip.asset_id or "",
+                text_content=clip.text_content or "",
+                start_time=clip.start_time,
+                duration=clip.duration,
+                transition_effect=clip.transition_effect or "cut",
+                transition_duration=clip.transition_duration or 0.0,
+                playback_speed=clip.playback_speed or 1.0,
+                config=dict(clip.config) if clip.config else None,
+            )
+            logger.debug("复制片段: source=%s target=%s order=%d", clip.id, new_clip.id, clip.order)
+
+        return self.get_plan_or_raise(created_plan.id)

scripts/check_migration_safety.py

@@ -35,6 +35,7 @@ import os
 import re
 import subprocess
 import sys
+import time
 import urllib.error
 import urllib.request
 from pathlib import Path
@@ -78,6 +79,15 @@ SAFE_PATTERNS = [
 ]
 
 
+def _get_env(*names: str, default: str = "") -> str:
+    """按优先级尝试多个环境变量名，返回第一个非空值。"""
+    for name in names:
+        val = os.environ.get(name, "")
+        if val:
+            return val
+    return default
+
+
 def extract_upgrade_content(content: str) -> str:
     """
     从迁移文件中提取 upgrade 函数的内容。
@@ -100,25 +110,60 @@ def extract_upgrade_content(content: str) -> str:
     return content[upgrade_start:upgrade_end]
 
 
-def get_new_migrations_via_diff(diff_target: str) -> List[Path]:
+def _api_get_with_retry(url: str, token: str, max_retries: int = 3) -> dict | list:
     """
-    通过 Gitea API 对比目标分支，找出 alembic/versions/ 下新增的迁移文件。
-    不依赖本地 git，避免 CI 环境下 git 操作不稳定的问题。
+    带重试的 API 调用。
+    指数退避：1s, 2s, 4s
     """
-    api_url = os.environ.get("GITHUB_API_URL", "")
-    repo = os.environ.get("GITHUB_REPOSITORY", "")
-    token = os.environ.get("GITHUB_TOKEN", "")
+    last_error = None
+    for attempt in range(max_retries):
+        try:
+            req = urllib.request.Request(url, headers={"Authorization": f"token {token}"})
+            with urllib.request.urlopen(req, timeout=15) as resp:
+                return json.loads(resp.read().decode())
+        except urllib.error.HTTPError as e:
+            # 404 说明目录不存在或分支不存在，直接抛
+            if e.code == 404:
+                raise
+            last_error = e
+            if attempt < max_retries - 1:
+                wait = 2**attempt
+                print(f"   （API 请求失败，{wait}s 后重试 {attempt + 1}/{max_retries}：{e}）")
+                time.sleep(wait)
+        except Exception as e:
+            last_error = e
+            if attempt < max_retries - 1:
+                wait = 2**attempt
+                print(f"   （API 请求失败，{wait}s 后重试 {attempt + 1}/{max_retries}：{e}）")
+                time.sleep(wait)
+    raise last_error  # type: ignore[misc]
+
+
+def get_new_migrations_via_api(diff_target: str) -> List[Path] | None:
+    """
+    通过 Gitea/GitHub Contents API 对比目标分支，找出 alembic/versions/ 下新增的迁移文件。
+    返回 None 表示 API 方式不可用，调用方应尝试其他方式。
+    """
+    # 同时支持 Gitea 和 GitHub 的环境变量命名
+    api_url = _get_env("GITEA_API_URL", "GITHUB_API_URL", "CI_API_V4_URL")
+    repo = _get_env("GITEA_REPOSITORY", "GITHUB_REPOSITORY", "CI_PROJECT_PATH")
+    token = _get_env("GITEA_TOKEN", "GITHUB_TOKEN", "CI_JOB_TOKEN")
     branch = diff_target.replace("origin/", "")
 
     if not api_url or not repo or not token:
-        print("⚠️  CI 环境变量不完整，降级为检查所有迁移文件")
-        return sorted(ALEMBIC_VERSIONS_DIR.glob("*.py"))
+        print(
+            f"   （API 环境变量不完整：api_url={'✓' if api_url else '✗'} repo={'✓' if repo else '✗'} token={'✓' if token else '✗'}）"
+        )
+        return None
 
     try:
         url = f"{api_url}/repos/{repo}/contents/alembic/versions?ref={branch}"
-        req = urllib.request.Request(url, headers={"Authorization": f"token {token}"})
-        with urllib.request.urlopen(req, timeout=15) as resp:
-            data = json.loads(resp.read().decode())
+        data = _api_get_with_retry(url, token)
+
+        if isinstance(data, dict):
+            # Gitea 目录不存在时返回 404，不会到这里；如果返回 dict 可能是错误信息
+            print(f"   （API 返回异常：{str(data)[:100]}）")
+            return None
 
         remote_files = {item["name"] for item in data if item["name"].endswith(".py")}
         local_files = {f.name for f in ALEMBIC_VERSIONS_DIR.glob("*.py")}
@@ -132,9 +177,70 @@ def get_new_migrations_via_diff(diff_target: str) -> List[Path]:
             print(f"   （API 对比 {branch} 分支，无新增迁移）")
             return []
     except Exception as e:
-        print(f"⚠️  API 获取迁移列表失败：{e}")
-        print("   降级为检查所有迁移文件")
-        return sorted(ALEMBIC_VERSIONS_DIR.glob("*.py"))
+        print(f"   （API 获取迁移列表失败：{e}）")
+        return None
+
+
+def get_new_migrations_via_git(diff_target: str) -> List[Path] | None:
+    """
+    Fallback：通过本地 git diff 找出新增的迁移文件。
+    CI 环境中 git 可用时作为 API 失败后的兜底方案。
+    """
+    try:
+        # 确保目标分支存在
+        subprocess.run(
+            ["git", "fetch", "origin", diff_target.replace("origin/", ""), "--depth=50"],
+            capture_output=True,
+            cwd=str(REPO_ROOT),
+            timeout=30,
+        )
+
+        result = subprocess.run(
+            ["git", "diff", "--name-only", "--diff-filter=A", f"{diff_target}...HEAD"],
+            capture_output=True,
+            text=True,
+            cwd=str(REPO_ROOT),
+            timeout=10,
+        )
+        if result.returncode != 0:
+            print(f"   （git diff 失败：{result.stderr.strip()}）")
+            return None
+
+        new_migrations = []
+        for line in result.stdout.strip().split("\n"):
+            line = line.strip()
+            if line.startswith("alembic/versions/") and line.endswith(".py"):
+                new_migrations.append(REPO_ROOT / line)
+
+        new_migrations.sort()
+        print(f"   （git diff 对比 {diff_target}，发现 {len(new_migrations)} 个新增迁移）")
+        return new_migrations
+    except Exception as e:
+        print(f"   （git diff 方式失败：{e}）")
+        return None
+
+
+def get_new_migrations_via_diff(diff_target: str) -> List[Path]:
+    """
+    找出相对目标分支新增的迁移文件，按优先级尝试多种方式：
+    1. Gitea/GitHub Contents API（最可靠，不受本地 checkout 深度影响）
+    2. git diff（API 失败时的兜底）
+    3. 全量扫描（以上都失败时的最后兜底，会输出警告）
+    """
+    print("🔍 尝试通过 API 获取新增迁移列表...")
+    result = get_new_migrations_via_api(diff_target)
+    if result is not None:
+        return result
+
+    print("🔍 API 不可用，尝试 git diff 方式...")
+    result = get_new_migrations_via_git(diff_target)
+    if result is not None:
+        return result
+
+    print("⚠️  所有增量方式均失败，降级为检查所有迁移文件")
+    print("   这可能导致历史迁移中的破坏性操作被误报")
+    print("   建议检查 CI 环境变量配置（GITHUB_API_URL / GITHUB_REPOSITORY / GITHUB_TOKEN）")
+    return sorted(ALEMBIC_VERSIONS_DIR.glob("*.py"))
 
 
 def find_new_migrations(since_revision: str | None = None, diff_against: str | None = None) -> List[Path]: