video-gen/video-gen-api/app/services/generation/history_delete_service.py

from __future__ import annotations

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.generation.pipeline.db_lock_service import execute_with_lock_timeout
from app.services.generation.ai.task_group_service import soft_delete_child_tasks_batch
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_service import log_operation_event
# from app.services.operation_log import log_operation
from app.services.video_upscale.guard_service import assert_no_recoverable_failed_upscale_tasks
from app.services.resource_accounting_service import (
    SOURCE_MODEL_CHAT_TASK,
    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 execute_with_lock_timeout(
        db,

        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 execute_with_lock_timeout(
        db,

        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="项目生成记录不存在或已删除")
    await assert_no_recoverable_failed_upscale_tasks(
        db,
        generation_record_ids=[str(record.id) for record in records],
    )

    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 execute_with_lock_timeout(
        db,

        select(ChatGenerationTask)
        .where(
            ChatGenerationTask.id.in_(ids),
            ChatGenerationTask.user_id == current_user.id,
            ChatGenerationTask.generation_mode.in_([
                GenerationMode.CHATAPI_ASYNC.value,
                GenerationMode.CHATAPI_CHILD.value,
            ]),
        )
        .with_for_update()
    )
    tasks = list(result.scalars().all())
    _raise_missing_if_any(ids=ids, found_ids=[task.id for task in tasks], message="AI 创作记录不存在或已删除")

    already_deleted_ids = [str(task.id) for task in tasks if task.deleted_at is not None]
    _raise_invalid_if_any(invalid_ids=already_deleted_ids, message="AI 创作记录不存在或已删除", status_code=404)

    await assert_no_recoverable_failed_upscale_tasks(db, [str(task.id) for task in tasks])

    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 创作记录只有生成完成后才能删除")

    async_ids = [str(task.id) for task in tasks if task.generation_mode == GenerationMode.CHATAPI_ASYNC.value]
    child_ids = [str(task.id) for task in tasks if task.generation_mode == GenerationMode.CHATAPI_CHILD.value]

    freed_size = 0
    if async_ids:
        freed_size += int(await soft_delete_resources_by_source(
            db,
            source_model=SOURCE_MODEL_CHAT_TASK,
            source_ids=async_ids,
            deleted_at=deleted_at,
        ) or 0)
        async_id_set = set(async_ids)
        for task in tasks:
            if str(task.id) in async_id_set:
                task.deleted_at = deleted_at

    if child_ids:
        freed_size += await soft_delete_child_tasks_batch(
            db,
            child_task_ids=child_ids,
            user_id=current_user.id,
            deleted_at=deleted_at,
            require_completed=True,
        )

    parent_task_ids = list(dict.fromkeys(
        str(task.parent_task_id) for task in tasks if task.parent_task_id
    ))
    log_operation_event(
        domain="generation_ai_batch",
        event_type="CHILD_RESOURCE_DELETE_SUCCESS",
        event_status="success",
        source="service",
        user_id=current_user.id,
        group_id=parent_task_ids[0] if len(parent_task_ids) == 1 else None,
        detail={
            "batch": True,
            "task_ids": [str(task.id) for task in tasks],
            "parent_task_ids": parent_task_ids,
            "freed_size_bytes": int(freed_size or 0),
            "physical_files_deleted": False,
        },
    )

    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 execute_with_lock_timeout(
        db,

        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 execute_with_lock_timeout(
        db,
        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())
    await assert_no_recoverable_failed_upscale_tasks(db, chat_task_ids)
    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 execute_with_lock_timeout(
        db,

        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