素材云批量删除API

video-gen-api/app/services/module_generation_flow_base_service.py

@@ -9,14 +9,17 @@ from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.enums.common import ModuleEventTypeEnum, ModuleStepStatusEnum
+from app.enums.generation_task import ChatGenerationPipelineStage, ChatGenerationTaskStatus
 from app.models.chat_generation_task import ChatGenerationTask
 from app.models.module_generation_project import ModuleGenerationProject
 from app.models.module_generation_step import ModuleGenerationStep
 from app.models.user import User
-from app.services.generation_refund_service import mark_chat_generation_task_failed_and_refund_once
 from app.enums.module_generation_flow import ModuleGenerationFlowConfig
 from app.services.module_generation_step_common_service import build_step_input, build_step_output, utc_now
-from app.services.resource_accounting_service import soft_delete_chat_task_resources
+from app.services.resource_accounting_service import (
+    SOURCE_MODEL_CHAT_TASK,
+    soft_delete_resources_by_source,
+)
 from app.utils.id_gen import generate_id
 
 LogModuleEventCallable = Callable[..., Awaitable[None]]
@@ -224,6 +227,106 @@ def _clear_project_final_resources_by_deleted_steps(
     return cleared_fields
 
 
+ACTIVE_CHAT_TASK_BLOCK_STATUSES = {
+    ChatGenerationTaskStatus.PENDING.value,
+    ChatGenerationTaskStatus.GENERATING.value,
+}
+
+ACTIVE_CHAT_TASK_BLOCK_STAGES = {
+    ChatGenerationPipelineStage.QUEUED.value,
+    ChatGenerationPipelineStage.PREPARING.value,
+    ChatGenerationPipelineStage.CREATING_PROVIDER_TASK.value,
+    ChatGenerationPipelineStage.WAITING_REMOTE.value,
+    ChatGenerationPipelineStage.POLLING.value,
+    ChatGenerationPipelineStage.RESULT_READY.value,
+    ChatGenerationPipelineStage.DOWNLOAD_QUEUED.value,
+    ChatGenerationPipelineStage.DOWNLOADING.value,
+    ChatGenerationPipelineStage.RETRY_WAITING.value,
+}
+
+
+def is_active_chat_generation_task(task: ChatGenerationTask) -> bool:
+    """判断 ChatGenerationTask 是否仍处于不可主动删除/废弃的处理中状态。"""
+    return bool(
+        task.status in ACTIVE_CHAT_TASK_BLOCK_STATUSES
+        or (task.pipeline_stage in ACTIVE_CHAT_TASK_BLOCK_STAGES)
+    )
+
+
+def _unique_ids(values: list[str | None] | tuple[str | None, ...]) -> list[str]:
+    return list(dict.fromkeys(str(value) for value in values if value))
+
+
+async def load_chat_tasks_for_steps(
+    db: AsyncSession,
+    steps: list[ModuleGenerationStep],
+    *,
+    for_update: bool = False,
+) -> dict[str, ChatGenerationTask]:
+    ids = _unique_ids([step.chat_task_id for step in steps])
+    if not ids:
+        return {}
+    stmt = select(ChatGenerationTask).where(
+        ChatGenerationTask.id.in_(ids),
+        ChatGenerationTask.deleted_at.is_(None),
+    )
+    if for_update:
+        stmt = stmt.with_for_update()
+    result = await db.execute(stmt)
+    return {task.id: task for task in result.scalars().all()}
+
+
+async def assert_no_active_chat_tasks_for_steps(
+    db: AsyncSession,
+    steps: list[ModuleGenerationStep],
+    *,
+    detail_message: str = "当前存在生成中任务，请等待生成完成或失败后再操作",
+    for_update: bool = True,
+) -> dict[str, ChatGenerationTask]:
+    """批量校验步骤关联任务是否有进行中任务。
+
+    用户主动修改上游步骤、删除项目、删除历史时，统一采用“生成中拦截、不退款”。
+    返回已批量加载的未删除 ChatGenerationTask，调用方可继续复用，避免重复查询。
+    """
+    task_map = await load_chat_tasks_for_steps(db, steps, for_update=for_update)
+    active_task_ids = [task.id for task in task_map.values() if is_active_chat_generation_task(task)]
+    if active_task_ids:
+        raise HTTPException(
+            status_code=409,
+            detail={
+                "message": detail_message,
+                "active_count": len(active_task_ids),
+            },
+        )
+    return task_map
+
+
+async def assert_project_has_no_active_chat_tasks(
+    db: AsyncSession,
+    *,
+    project: ModuleGenerationProject,
+    config: ModuleGenerationFlowConfig,
+    detail_message: str = "当前存在生成中任务，请等待生成完成或失败后再操作",
+) -> dict[str, ChatGenerationTask]:
+    result = await db.execute(
+        select(ModuleGenerationStep)
+        .where(
+            ModuleGenerationStep.project_id == project.id,
+            ModuleGenerationStep.module == config.module,
+            ModuleGenerationStep.deleted_at.is_(None),
+            ModuleGenerationStep.chat_task_id.is_not(None),
+        )
+        .with_for_update()
+    )
+    steps = list(result.scalars().all())
+    return await assert_no_active_chat_tasks_for_steps(
+        db,
+        steps,
+        detail_message=detail_message,
+        for_update=True,
+    )
+
+
 async def soft_delete_steps_from_index(
     db: AsyncSession,
     *,
@@ -232,9 +335,17 @@ async def soft_delete_steps_from_index(
     config: ModuleGenerationFlowConfig,
     log_module_event: LogModuleEventCallable,
     deleted_at: datetime | None = None,
-    refund_unfinished: bool = True,
+    refund_unfinished: bool = False,
+    block_active_tasks: bool = True,
     release_stats: dict[str, int] | None = None,
 ) -> list[ModuleGenerationStep]:
+    """软删指定步骤及其后续当前版本步骤。
+
+    说明：
+    - 兼容旧调用方保留 refund_unfinished 参数，但用户主动修改/删除链路不再退款。
+    - 存在 pending/generating/下载中/轮询中等任务时直接 409 拦截。
+    - 已完成任务只做软删任务与 generated_resources，释放容量统计；失败任务只软删任务。
+    """
     deleted_at = deleted_at or utc_now()
     result = await db.execute(
         select(ModuleGenerationStep)
@@ -248,6 +359,21 @@ async def soft_delete_steps_from_index(
         .with_for_update()
     )
     steps = list(result.scalars().all())
+    if not steps:
+        return []
+
+    task_map = await load_chat_tasks_for_steps(db, steps, for_update=True)
+    if block_active_tasks:
+        active_task_ids = [task.id for task in task_map.values() if is_active_chat_generation_task(task)]
+        if active_task_ids:
+            raise HTTPException(
+                status_code=409,
+                detail={
+                    "message": "当前存在生成中任务，请等待生成完成或失败后再操作",
+                    "active_count": len(active_task_ids),
+                },
+            )
+
     deleted_step_codes = {step.step_code for step in steps}
     cleared_project_fields = _clear_project_final_resources_by_deleted_steps(
         project,
@@ -255,30 +381,27 @@ async def soft_delete_steps_from_index(
         config=config,
     )
 
+    completed_task_ids = [
+        task.id
+        for task in task_map.values()
+        if task.status == ChatGenerationTaskStatus.COMPLETED.value
+    ]
+    if completed_task_ids:
+        released_size = await soft_delete_resources_by_source(
+            db,
+            source_model=SOURCE_MODEL_CHAT_TASK,
+            source_ids=completed_task_ids,
+            deleted_at=deleted_at,
+        )
+        if release_stats is not None:
+            release_stats["released_size_bytes"] = int(release_stats.get("released_size_bytes", 0)) + int(released_size or 0)
+
     for step in steps:
         step.is_current = False
         step.deleted_at = deleted_at
-        if step.chat_task_id:
-            chat_result = await db.execute(
-                select(ChatGenerationTask)
-                .where(ChatGenerationTask.id == step.chat_task_id, ChatGenerationTask.deleted_at.is_(None))
-                .with_for_update()
-                .limit(1)
-            )
-            chat_task = chat_result.scalar_one_or_none()
-            if chat_task:
-                if chat_task.status == "completed":
-                    released_size = await soft_delete_chat_task_resources(db, chat_task.id, deleted_at=deleted_at)
-                    if release_stats is not None:
-                        release_stats["released_size_bytes"] = int(release_stats.get("released_size_bytes", 0)) + int(released_size or 0)
-                elif refund_unfinished and chat_task.status != "failed":
-                    await mark_chat_generation_task_failed_and_refund_once(
-                        db,
-                        task=chat_task,
-                        error_message=config.cancel_chat_task_error_message,
-                        pipeline_stage="failed",
-                    )
-                chat_task.deleted_at = deleted_at
+        if step.chat_task_id and step.chat_task_id in task_map:
+            task_map[step.chat_task_id].deleted_at = deleted_at
+
     if steps:
         await log_module_event(
             db,
@@ -289,14 +412,12 @@ async def soft_delete_steps_from_index(
                 "step_ids": [step.id for step in steps],
                 "step_codes": [step.step_code for step in steps],
                 "cleared_project_fields": cleared_project_fields,
+                "block_active_tasks": block_active_tasks,
+                "refund_unfinished": False,
             },
         )
     return steps
 
 
 async def chat_tasks_by_id(db: AsyncSession, steps: list[ModuleGenerationStep]) -> dict[str, ChatGenerationTask]:
-    ids = [step.chat_task_id for step in steps if step.chat_task_id]
-    if not ids:
-        return {}
-    result = await db.execute(select(ChatGenerationTask).where(ChatGenerationTask.id.in_(ids)))
-    return {task.id: task for task in result.scalars().all()}
+    return await load_chat_tasks_for_steps(db, steps, for_update=False)