From 0fba6fdf0935d340d9ff1ea3397f35c8d521f56d Mon Sep 17 00:00:00 2001 From: GinHa <15201596918@163.com> Date: Thu, 2 Jul 2026 11:37:32 +0800 Subject: [PATCH] =?UTF-8?q?=E7=B4=A0=E6=9D=90=E4=BA=91=E6=89=B9=E9=87=8F?= =?UTF-8?q?=E5=88=A0=E9=99=A4API?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- video-gen-api/app/api/v1/generation_ai.py | 50 ++ video-gen-api/app/enums/generation_history.py | 3 + video-gen-api/app/schemas/generation_ai.py | 92 +++- .../generation_history_delete_service.py | 514 ++++++++++++++++++ .../module_generation_flow_base_service.py | 179 +++++- .../services/shot_replicate_flow_service.py | 53 +- 6 files changed, 815 insertions(+), 76 deletions(-) create mode 100644 video-gen-api/app/services/generation_history_delete_service.py diff --git a/video-gen-api/app/api/v1/generation_ai.py b/video-gen-api/app/api/v1/generation_ai.py index 28db9b24..17842a22 100644 --- a/video-gen-api/app/api/v1/generation_ai.py +++ b/video-gen-api/app/api/v1/generation_ai.py @@ -9,6 +9,8 @@ from app.models.chat_generation_task import ChatGenerationTask from app.models.user import User from app.schemas.generation_ai import ( GenerationAIEngineOptionsOut, + GenerationAIHistoryBatchDeleteOut, + GenerationAIHistoryBatchDeleteRequest, GenerationAIHistoryDayItemsOut, GenerationAIHistoryGroupedOut, GenerationAIRetryOut, @@ -31,6 +33,7 @@ from app.services.generation_billing_service import ( charge_generation_media_by_params, get_next_credit_attempt_no, ) +from app.services.generation_history_delete_service import batch_delete_generation_history_items from app.services.generation_log_service import log_task_event from app.services.generation_refund_service import mark_chat_generation_task_failed_and_refund_once from app.services.resource_capacity_service import assert_user_resource_capacity_available @@ -332,6 +335,53 @@ async def list_history_grouped_days( ) +@router.delete( + "/history/batch", + response_model=GenerationAIHistoryBatchDeleteOut, + summary="批量删除素材云历史记录", + description=( + "按 history_source 批量软删除素材云历史记录，单次最多30条。" + "generation_record 和 chat_task 入参 ids 为对应记录ID，且只有生成完成后才能删除；" + "hot_opening_replicate 入参 ids 为 module_project_id；" + "shot_replicate 入参 ids 为 shot_segment_id。" + "爆款开头复刻和拆镜复刻会联动软删 ModuleGenerationProject、ModuleGenerationStep、关联 ChatGenerationTask 和 generated_resources；" + "拆镜复刻还会联动软删 ShotReplicateSegment。" + "如果存在生成中、轮询中、下载中等任务，接口直接拦截，不做失败标记、不退款。" + ), + responses={ + 200: { + "description": "批量软删除成功", + }, + 400: { + "description": "参数错误，例如 history_source 不支持、ids 为空、超过30条或重复", + }, + 401: { + "description": "未登录或 Token 无效", + }, + 404: { + "description": "部分 ID 不存在、不属于当前用户或已删除", + }, + 409: { + "description": "存在未完成或生成中的记录，当前不能删除", + }, + }, +) +async def batch_delete_history_items( + req: GenerationAIHistoryBatchDeleteRequest = Body( + ..., + description="素材云历史批量删除参数。不同 history_source 对应不同 ID 语义，详见字段说明", + ), + current_user: User = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +): + return await batch_delete_generation_history_items( + db=db, + current_user=current_user, + history_source=req.history_source, + ids=req.ids, + ) + + @router.get( "/history/{generated_date}", response_model=GenerationAIHistoryDayItemsOut, diff --git a/video-gen-api/app/enums/generation_history.py b/video-gen-api/app/enums/generation_history.py index 974cd185..b53f863e 100644 --- a/video-gen-api/app/enums/generation_history.py +++ b/video-gen-api/app/enums/generation_history.py @@ -118,3 +118,6 @@ def is_generation_history_module_source(source: GenerationHistorySourceEnum) -> """判断当前来源是否需要回填模块项目信息。""" return source in GENERATION_HISTORY_MODULE_SOURCES + + +MAX_BATCH_DELETE_COUNT = 30 diff --git a/video-gen-api/app/schemas/generation_ai.py b/video-gen-api/app/schemas/generation_ai.py index 168ec56a..f18d2629 100644 --- a/video-gen-api/app/schemas/generation_ai.py +++ b/video-gen-api/app/schemas/generation_ai.py @@ -1,8 +1,9 @@ from typing import Annotated, Literal -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, field_validator from app.schemas.common import NaiveDatetimeOptional +from app.enums.generation_history import normalize_generation_history_source class GenerationAIReference(BaseModel): @@ -492,6 +493,95 @@ class GenerationAITaskDeleteOut(BaseModel): freed_size_bytes: int = Field(0, description="本次软删联动释放的有效资源空间字节数") +class GenerationAIHistoryBatchDeleteRequest(BaseModel): + """素材云历史批量删除请求体。""" + + model_config = ConfigDict( + json_schema_extra={ + "example": { + "history_source": "shot_replicate", + "ids": ["shot_segment_id_1", "shot_segment_id_2"], + } + } + ) + + history_source: str = Field( + ..., + description=( + "素材云历史来源。" + "generation_record=项目生成，chat_task=AI创作，" + "hot_opening_replicate=爆款开头复刻，shot_replicate=拆镜复刻。" + "注意：项目生成/AI创作传记录ID；爆款开头复刻传 module_project_id；拆镜复刻传 shot_segment_id。" + ), + examples=["shot_replicate"], + ) + ids: list[str] = Field( + ..., + min_length=1, + max_length=30, + description="需要删除的ID数组，最多30个；不允许重复或空字符串", + examples=[["shot_segment_id_1", "shot_segment_id_2"]], + ) + + @field_validator("history_source") + @classmethod + def validate_history_source(cls, value: str) -> str: + try: + return normalize_generation_history_source(value).value + except ValueError as exc: + raise ValueError("history_source 不支持") from exc + + @field_validator("ids") + @classmethod + def validate_ids(cls, values: list[str]) -> list[str]: + normalized = [str(item).strip() for item in values if str(item or "").strip()] + if not normalized: + raise ValueError("ids 不能为空") + if len(normalized) > 30: + raise ValueError("单次最多删除30条记录") + if len(normalized) != len(set(normalized)): + raise ValueError("ids 不允许重复") + return normalized + + +class GenerationAIHistoryBatchDeleteOut(BaseModel): + """素材云历史批量删除响应体。""" + + model_config = ConfigDict( + json_schema_extra={ + "example": { + "message": "删除成功", + "history_source": "shot_replicate", + "history_source_label": "拆镜复刻", + "requested_count": 2, + "deleted_count": 2, + "requested_ids": ["shot_segment_id_1", "shot_segment_id_2"], + "deleted_ids": ["shot_segment_id_1", "shot_segment_id_2"], + "generation_record_ids": [], + "chat_task_ids": ["chat_task_id_1", "chat_task_id_2"], + "module_project_ids": ["module_project_id_1", "module_project_id_2"], + "shot_segment_ids": ["shot_segment_id_1", "shot_segment_id_2"], + "deleted": True, + "freed_size_bytes": 123456, + } + } + ) + + message: str = Field(..., description="操作结果提示信息") + history_source: str = Field(..., description="素材云历史来源") + history_source_label: str | None = Field(None, description="素材云历史来源中文名称") + requested_count: int = Field(..., description="请求删除数量") + deleted_count: int = Field(..., description="实际删除数量") + requested_ids: list[str] = Field(default_factory=list, description="请求删除的原始ID列表") + deleted_ids: list[str] = Field(default_factory=list, description="已删除的原始ID列表") + generation_record_ids: list[str] = Field(default_factory=list, description="联动软删除的 GenerationRecord ID") + chat_task_ids: list[str] = Field(default_factory=list, description="联动软删除的 ChatGenerationTask ID") + module_project_ids: list[str] = Field(default_factory=list, description="联动软删除的 ModuleGenerationProject ID") + shot_segment_ids: list[str] = Field(default_factory=list, description="联动软删除的 ShotReplicateSegment ID") + deleted: bool = Field(..., description="是否已完成软删除") + freed_size_bytes: int = Field(0, description="本次软删联动释放的有效资源空间字节数") + + class GenerationAIRetryOut(BaseModel): """AI生成任务重试响应体。""" diff --git a/video-gen-api/app/services/generation_history_delete_service.py b/video-gen-api/app/services/generation_history_delete_service.py new file mode 100644 index 00000000..47204b12 --- /dev/null +++ b/video-gen-api/app/services/generation_history_delete_service.py @@ -0,0 +1,514 @@ +from __future__ import annotations + +import json +from datetime import datetime, timezone +from typing import Iterable, Sequence + +from fastapi import HTTPException +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.enums.common import ModuleEventTypeEnum +from app.enums.generation_history import ( + GenerationHistorySourceEnum, + get_generation_history_source_label, + normalize_generation_history_source, + MAX_BATCH_DELETE_COUNT, +) +from app.enums.generation_task import ChatGenerationTaskStatus, GenerationMode +from app.enums.shot_replicate import ShotSegmentAnalysisStatusEnum, ShotSegmentReplicateStatusEnum, ShotSplitStatusEnum +from app.models.chat_generation_task import ChatGenerationTask +from app.models.generation_record import GenerationRecord +from app.models.module_generation_project import ModuleGenerationProject +from app.models.module_generation_step import ModuleGenerationStep +from app.models.shot_replicate_segment import ShotReplicateSegment +from app.models.user import User +from app.schemas.generation_ai import GenerationAIHistoryBatchDeleteOut +from app.services.module_generation_flow_base_service import is_active_chat_generation_task +from app.services.module_generation_log_service import log_module_event_file +# from app.services.operation_log import log_operation +from app.services.resource_accounting_service import ( + SOURCE_MODEL_CHAT_TASK, + SOURCE_MODEL_GENERATION_RECORD, + SOURCE_MODEL_SHOT_SEGMENT, + soft_delete_generation_record_resources, + soft_delete_resources_by_source, +) + + +COMPLETED_STATUS = ChatGenerationTaskStatus.COMPLETED.value + +_ACTIVE_SPLIT_STATUSES = { + ShotSplitStatusEnum.PENDING.value, + ShotSplitStatusEnum.PROCESSING.value, + ShotSplitStatusEnum.RETRY_WAITING.value, +} +_ACTIVE_SEGMENT_ANALYSIS_STATUSES = { + ShotSegmentAnalysisStatusEnum.PENDING.value, + ShotSegmentAnalysisStatusEnum.PROCESSING.value, +} +_ACTIVE_SEGMENT_REPLICATE_STATUSES = { + ShotSegmentReplicateStatusEnum.PROCESSING.value, +} + + +def _normalize_ids(ids: Sequence[str] | Iterable[str]) -> list[str]: + normalized = [str(item).strip() for item in ids if str(item or "").strip()] + if not normalized: + raise HTTPException(status_code=400, detail="ids 不能为空") + if len(normalized) > MAX_BATCH_DELETE_COUNT: + raise HTTPException(status_code=400, detail=f"单次最多删除 {MAX_BATCH_DELETE_COUNT} 条记录") + if len(normalized) != len(set(normalized)): + raise HTTPException(status_code=400, detail="ids 不允许重复") + return normalized + + +def _missing_ids(request_ids: list[str], actual_ids: Iterable[str]) -> list[str]: + actual_set = {str(item) for item in actual_ids if item} + return [item for item in request_ids if item not in actual_set] + + +def _raise_missing_if_any(*, ids: list[str], found_ids: Iterable[str], message: str) -> None: + missing = _missing_ids(ids, found_ids) + if missing: + raise HTTPException( + status_code=404, + detail={ + "message": message, + "missing_ids": missing, + "missing_count": len(missing), + }, + ) + + +def _raise_invalid_if_any(*, invalid_ids: list[str], message: str, status_code: int = 409) -> None: + if invalid_ids: + raise HTTPException( + status_code=status_code, + detail={ + "message": message, + "invalid_ids": invalid_ids, + "invalid_count": len(invalid_ids), + }, + ) + + +def _task_id_list(steps: list[ModuleGenerationStep]) -> list[str]: + return list(dict.fromkeys(step.chat_task_id for step in steps if step.chat_task_id)) + + +async def _load_chat_tasks_for_steps( + db: AsyncSession, + steps: list[ModuleGenerationStep], +) -> dict[str, ChatGenerationTask]: + task_ids = _task_id_list(steps) + if not task_ids: + return {} + result = await db.execute( + select(ChatGenerationTask) + .where( + ChatGenerationTask.id.in_(task_ids), + ChatGenerationTask.deleted_at.is_(None), + ) + .with_for_update() + ) + return {task.id: task for task in result.scalars().all()} + + +def _assert_no_active_chat_tasks(tasks: Iterable[ChatGenerationTask]) -> None: + active_task_ids = [task.id for task in tasks if is_active_chat_generation_task(task)] + if active_task_ids: + raise HTTPException( + status_code=409, + detail={ + "message": "当前存在生成中任务，请等待生成完成或失败后再操作", + "active_count": len(active_task_ids), + }, + ) + + +# async def _log_history_batch_delete( +# db: AsyncSession, +# *, +# current_user: User, +# result: GenerationAIHistoryBatchDeleteOut, +# ) -> None: +# detail = { +# "history_source": result.history_source, +# "history_source_label": result.history_source_label, +# "requested_count": result.requested_count, +# "deleted_count": result.deleted_count, +# "requested_ids": result.requested_ids, +# "deleted_ids": result.deleted_ids, +# "generation_record_ids": result.generation_record_ids, +# "chat_task_ids": result.chat_task_ids, +# "module_project_ids": result.module_project_ids, +# "shot_segment_ids": result.shot_segment_ids, +# "freed_size_bytes": result.freed_size_bytes, +# } +# await log_operation( +# db, +# current_user.id, +# current_user.username, +# f"批量删除素材云历史-{result.history_source_label or result.history_source}", +# "DELETE", +# "/generation-ai/history/batch", +# detail=json.dumps(detail, ensure_ascii=False, default=str), +# ) + + +def _build_out( + *, + source: GenerationHistorySourceEnum, + requested_ids: list[str], + deleted_ids: list[str], + generation_record_ids: list[str] | None = None, + chat_task_ids: list[str] | None = None, + module_project_ids: list[str] | None = None, + shot_segment_ids: list[str] | None = None, + freed_size_bytes: int = 0, +) -> GenerationAIHistoryBatchDeleteOut: + return GenerationAIHistoryBatchDeleteOut( + message="删除成功", + history_source=source.value, + history_source_label=get_generation_history_source_label(source), + requested_count=len(requested_ids), + deleted_count=len(deleted_ids), + requested_ids=requested_ids, + deleted_ids=deleted_ids, + generation_record_ids=generation_record_ids or [], + chat_task_ids=chat_task_ids or [], + module_project_ids=module_project_ids or [], + shot_segment_ids=shot_segment_ids or [], + deleted=True, + freed_size_bytes=int(freed_size_bytes or 0), + ) + + +async def _delete_generation_records( + db: AsyncSession, + *, + current_user: User, + source: GenerationHistorySourceEnum, + ids: list[str], + deleted_at: datetime, +) -> GenerationAIHistoryBatchDeleteOut: + result = await db.execute( + select(GenerationRecord) + .where( + GenerationRecord.id.in_(ids), + GenerationRecord.user_id == current_user.id, + GenerationRecord.deleted_at.is_(None), + ) + .with_for_update() + ) + records = list(result.scalars().all()) + _raise_missing_if_any(ids=ids, found_ids=[record.id for record in records], message="项目生成记录不存在或已删除") + + invalid_ids = [ + record.id + for record in records + if record.status != COMPLETED_STATUS or record.generated_at is None + ] + _raise_invalid_if_any(invalid_ids=invalid_ids, message="项目生成记录只有生成完成后才能删除") + + freed_size = await soft_delete_generation_record_resources(db, [record.id for record in records], deleted_at=deleted_at) + for record in records: + record.deleted_at = deleted_at + + return _build_out( + source=source, + requested_ids=ids, + deleted_ids=ids, + generation_record_ids=ids, + freed_size_bytes=freed_size, + ) + + +async def _delete_chat_tasks( + db: AsyncSession, + *, + current_user: User, + source: GenerationHistorySourceEnum, + ids: list[str], + deleted_at: datetime, +) -> GenerationAIHistoryBatchDeleteOut: + result = await db.execute( + select(ChatGenerationTask) + .where( + ChatGenerationTask.id.in_(ids), + ChatGenerationTask.user_id == current_user.id, + ChatGenerationTask.generation_mode == GenerationMode.CHATAPI_ASYNC.value, + ChatGenerationTask.deleted_at.is_(None), + ) + .with_for_update() + ) + tasks = list(result.scalars().all()) + _raise_missing_if_any(ids=ids, found_ids=[task.id for task in tasks], message="AI 创作记录不存在或已删除") + + invalid_ids = [ + task.id + for task in tasks + if task.status != COMPLETED_STATUS or task.generated_at is None + ] + _raise_invalid_if_any(invalid_ids=invalid_ids, message="AI 创作记录只有生成完成后才能删除") + + freed_size = await soft_delete_resources_by_source( + db, + source_model=SOURCE_MODEL_CHAT_TASK, + source_ids=[task.id for task in tasks], + deleted_at=deleted_at, + ) + for task in tasks: + task.deleted_at = deleted_at + + return _build_out( + source=source, + requested_ids=ids, + deleted_ids=ids, + chat_task_ids=ids, + freed_size_bytes=freed_size, + ) + + +async def _load_module_projects_by_ids( + db: AsyncSession, + *, + current_user: User, + source: GenerationHistorySourceEnum, + project_ids: list[str], +) -> list[ModuleGenerationProject]: + result = await db.execute( + select(ModuleGenerationProject) + .where( + ModuleGenerationProject.id.in_(project_ids), + ModuleGenerationProject.user_id == current_user.id, + ModuleGenerationProject.module == source.value, + ModuleGenerationProject.deleted_at.is_(None), + ) + .with_for_update() + ) + projects = list(result.scalars().all()) + _raise_missing_if_any(ids=project_ids, found_ids=[project.id for project in projects], message="模块生成项目不存在或已删除") + return projects + + +async def _soft_delete_module_projects( + db: AsyncSession, + *, + source: GenerationHistorySourceEnum, + projects: list[ModuleGenerationProject], + deleted_at: datetime, +) -> tuple[list[str], list[str], int]: + project_ids = list(dict.fromkeys(project.id for project in projects)) + if not project_ids: + return [], [], 0 + + step_result = await db.execute( + select(ModuleGenerationStep) + .where( + ModuleGenerationStep.project_id.in_(project_ids), + ModuleGenerationStep.module == source.value, + ModuleGenerationStep.deleted_at.is_(None), + ModuleGenerationStep.is_current == True, + ) + .with_for_update() + ) + steps = list(step_result.scalars().all()) + task_map = await _load_chat_tasks_for_steps(db, steps) + _assert_no_active_chat_tasks(task_map.values()) + + chat_task_ids = list(task_map.keys()) + freed_size = await soft_delete_resources_by_source( + db, + source_model=SOURCE_MODEL_CHAT_TASK, + source_ids=chat_task_ids, + deleted_at=deleted_at, + ) + + steps_by_project: dict[str, list[ModuleGenerationStep]] = {} + for step in steps: + steps_by_project.setdefault(step.project_id, []).append(step) + step.deleted_at = deleted_at + step.is_current = False + + for task in task_map.values(): + task.deleted_at = deleted_at + + for project in projects: + project.deleted_at = deleted_at + project.final_image_url = None + project.final_video_url = None + project.final_video_cover_url = None + project.completed_at = None + project_steps = steps_by_project.get(project.id, []) + log_module_event_file( + module=source.value, + event_type=ModuleEventTypeEnum.PROJECT_DELETED.value, + project_id=project.id, + user_id=project.user_id, + message="素材云历史批量删除模块项目", + detail={ + "source": "generation_history_batch_delete", + "step_ids": [step.id for step in project_steps], + "step_codes": [step.step_code for step in project_steps], + "chat_task_ids": [step.chat_task_id for step in project_steps if step.chat_task_id], + "refund_unfinished": False, + }, + ) + + return project_ids, chat_task_ids, int(freed_size or 0) + + +async def _delete_hot_opening_projects( + db: AsyncSession, + *, + current_user: User, + source: GenerationHistorySourceEnum, + ids: list[str], + deleted_at: datetime, +) -> GenerationAIHistoryBatchDeleteOut: + projects = await _load_module_projects_by_ids(db, current_user=current_user, source=source, project_ids=ids) + module_project_ids, chat_task_ids, freed_size = await _soft_delete_module_projects( + db, + source=source, + projects=projects, + deleted_at=deleted_at, + ) + return _build_out( + source=source, + requested_ids=ids, + deleted_ids=ids, + module_project_ids=module_project_ids, + chat_task_ids=chat_task_ids, + freed_size_bytes=freed_size, + ) + + +def _assert_segments_not_active(segments: list[ShotReplicateSegment]) -> None: + invalid_ids = [ + segment.id + for segment in segments + if segment.split_status in _ACTIVE_SPLIT_STATUSES + or segment.analysis_status in _ACTIVE_SEGMENT_ANALYSIS_STATUSES + or segment.replicate_status in _ACTIVE_SEGMENT_REPLICATE_STATUSES + ] + _raise_invalid_if_any(invalid_ids=invalid_ids, message="拆镜片段仍在分割、分析或复刻处理中，暂不能删除") + + +async def _delete_shot_segments( + db: AsyncSession, + *, + current_user: User, + source: GenerationHistorySourceEnum, + ids: list[str], + deleted_at: datetime, +) -> GenerationAIHistoryBatchDeleteOut: + result = await db.execute( + select(ShotReplicateSegment) + .where( + ShotReplicateSegment.id.in_(ids), + ShotReplicateSegment.user_id == current_user.id, + ShotReplicateSegment.deleted_at.is_(None), + ) + .with_for_update() + ) + segments = list(result.scalars().all()) + _raise_missing_if_any(ids=ids, found_ids=[segment.id for segment in segments], message="拆镜复刻片段不存在或已删除") + _assert_segments_not_active(segments) + + missing_project_segment_ids = [segment.id for segment in segments if not segment.module_project_id] + _raise_invalid_if_any(invalid_ids=missing_project_segment_ids, message="拆镜复刻片段尚未关联复刻项目，不能按素材云历史删除") + + module_project_ids = list(dict.fromkeys(segment.module_project_id for segment in segments if segment.module_project_id)) + projects = await _load_module_projects_by_ids( + db, + current_user=current_user, + source=source, + project_ids=module_project_ids, + ) + deleted_project_ids, chat_task_ids, project_freed_size = await _soft_delete_module_projects( + db, + source=source, + projects=projects, + deleted_at=deleted_at, + ) + + segment_freed_size = await soft_delete_resources_by_source( + db, + source_model=SOURCE_MODEL_SHOT_SEGMENT, + source_ids=[segment.id for segment in segments], + deleted_at=deleted_at, + ) + for segment in segments: + segment.deleted_at = deleted_at + + return _build_out( + source=source, + requested_ids=ids, + deleted_ids=ids, + module_project_ids=deleted_project_ids, + chat_task_ids=chat_task_ids, + shot_segment_ids=ids, + freed_size_bytes=int(project_freed_size or 0) + int(segment_freed_size or 0), + ) + + +async def batch_delete_generation_history_items( + db: AsyncSession, + *, + current_user: User, + history_source: str, + ids: Sequence[str] | Iterable[str], +) -> GenerationAIHistoryBatchDeleteOut: + """ + 按素材云 history_source 批量软删除历史记录。 + 事务由 get_db 统一提交/回滚；本服务只 flush，不主动 commit。 + 所有分支均为“先批量查询校验，再统一软删”，任何校验失败都会整体回滚。 + """ + try: + source = normalize_generation_history_source(history_source) + except ValueError as exc: + raise HTTPException(status_code=400, detail="history_source 不支持") from exc + + normalized_ids = _normalize_ids(ids) + deleted_at = datetime.now(timezone.utc) + + if source == GenerationHistorySourceEnum.GENERATION_RECORD: + result = await _delete_generation_records( + db, + current_user=current_user, + source=source, + ids=normalized_ids, + deleted_at=deleted_at, + ) + elif source == GenerationHistorySourceEnum.CHAT_TASK: + result = await _delete_chat_tasks( + db, + current_user=current_user, + source=source, + ids=normalized_ids, + deleted_at=deleted_at, + ) + elif source == GenerationHistorySourceEnum.HOT_OPENING_REPLICATE: + result = await _delete_hot_opening_projects( + db, + current_user=current_user, + source=source, + ids=normalized_ids, + deleted_at=deleted_at, + ) + elif source == GenerationHistorySourceEnum.SHOT_REPLICATE: + result = await _delete_shot_segments( + db, + current_user=current_user, + source=source, + ids=normalized_ids, + deleted_at=deleted_at, + ) + else: + raise HTTPException(status_code=400, detail="history_source 不支持") + + # await _log_history_batch_delete(db, current_user=current_user, result=result) + await db.flush() + return result diff --git a/video-gen-api/app/services/module_generation_flow_base_service.py b/video-gen-api/app/services/module_generation_flow_base_service.py index 4be57cb9..0db913e5 100644 --- a/video-gen-api/app/services/module_generation_flow_base_service.py +++ b/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) diff --git a/video-gen-api/app/services/shot_replicate_flow_service.py b/video-gen-api/app/services/shot_replicate_flow_service.py index 2fd5d3e2..fd5869dc 100644 --- a/video-gen-api/app/services/shot_replicate_flow_service.py +++ b/video-gen-api/app/services/shot_replicate_flow_service.py @@ -12,7 +12,6 @@ from sqlalchemy.orm.attributes import flag_modified from app.config import settings from app.enums.common import ModuleEventTypeEnum, ModuleProjectStatusEnum, ModulePromptTypeEnum, ModuleStepStatusEnum -from app.enums.generation_task import ChatGenerationPipelineStage, ChatGenerationTaskStatus from app.enums.shot_replicate import ShotReplicateGenerationModeEnum, ShotReplicateStepCodeEnum, ModuleCodeEnum from app.models.chat_generation_task import ChatGenerationTask from app.models.module_generation_project import ModuleGenerationProject @@ -55,6 +54,7 @@ from app.services.module_generation_log_service import log_module_error, log_mod from app.services.llm import optimize_prompt from app.services.resource_accounting_service import soft_delete_chat_task_resources from app.services.module_generation_flow_base_service import ( + assert_project_has_no_active_chat_tasks as _base_assert_project_has_no_active_chat_tasks, chat_tasks_by_id as _base_chat_tasks_by_id, create_module_step as _base_create_step, get_current_step_by_code as _base_get_current_step_by_code, @@ -336,7 +336,7 @@ async def _soft_delete_steps_from_index( project: ModuleGenerationProject, start_index: int, deleted_at: datetime | None = None, - refund_unfinished: bool = True, + refund_unfinished: bool = False, release_stats: dict[str, int] | None = None, ) -> None: await _base_soft_delete_steps_from_index( @@ -1412,57 +1412,18 @@ async def handle_chat_generation_task_failed(db: AsyncSession, task: ChatGenerat -_ACTIVE_DELETE_BLOCK_STATUSES = { - ChatGenerationTaskStatus.PENDING.value, - ChatGenerationTaskStatus.GENERATING.value, -} -_ACTIVE_DELETE_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, -} - - async def _assert_project_has_no_active_chat_tasks_for_delete( db: AsyncSession, *, project: ModuleGenerationProject, ) -> None: """用户主动删除项目/切片时不退款；如仍有异步生成任务进行中，直接拦截。""" - step_result = await db.execute( - select(ModuleGenerationStep.chat_task_id) - .where( - ModuleGenerationStep.project_id == project.id, - ModuleGenerationStep.module == MODULE, - ModuleGenerationStep.deleted_at.is_(None), - ModuleGenerationStep.chat_task_id.is_not(None), - ) + await _base_assert_project_has_no_active_chat_tasks( + db, + project=project, + config=FLOW_CONFIG, + detail_message="当前拆镜复刻项目仍有生成中任务，暂不能删除", ) - chat_task_ids = [task_id for task_id in step_result.scalars().all() if task_id] - if not chat_task_ids: - return - - task_result = await db.execute( - select(ChatGenerationTask) - .where( - ChatGenerationTask.id.in_(chat_task_ids), - ChatGenerationTask.deleted_at.is_(None), - ) - .with_for_update() - ) - active_tasks = [] - for task in task_result.scalars().all(): - if task.status in _ACTIVE_DELETE_BLOCK_STATUSES or (task.pipeline_stage in _ACTIVE_DELETE_BLOCK_STAGES): - active_tasks.append(task.id) - - if active_tasks: - raise HTTPException(status_code=400, detail="当前拆镜复刻项目仍有生成中任务，暂不能删除") async def mark_shot_replicate_step_dispatch_failed(