from __future__ import annotations import uuid from datetime import datetime, timezone from pathlib import Path from typing import Any from fastapi import HTTPException from app.config import settings from app.enums.celery_queue import CeleryQueue from app.enums.credit_record import ( CreditRecordBillingScene, CreditRecordChargeKind, CreditRecordOwnerType, CreditRecordSourceModule, CreditRecordSourceStepCode, ) from sqlalchemy import String, case, cast, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.enums.shot_replicate import ( ModuleCodeEnum, ShotAnalysisStatusEnum, ShotSegmentAnalysisStatusEnum, ShotSegmentReplicateStatusEnum, ShotReplicateLogEventEnum, ShotSegmentSourceModeEnum, ShotSplitStatusEnum, ShotTaskSetStatusEnum, ) from app.models.module_generation_project import ModuleGenerationProject from app.models.shot_replicate_segment import ShotReplicateSegment from app.models.shot_replicate_task_set import ShotReplicateTaskSet from app.models.user import User from app.schemas.shot_replicate import ( ShotAISuggestionOut, ShotSegmentDeleteOut, ShotSegmentDetailOut, ShotSegmentListOut, ShotSegmentSplitRetryOut, ShotReanalyzeOut, ShotReplicateDeleteOut, ShotSegmentOut, ShotSplitByAIOut, ShotSplitByAIRequest, ShotSplitCustomOut, ShotSplitCustomRequest, ShotTaskSetCreate, ShotTaskSetDeleteOut, ShotTaskSetDetailOut, ShotTaskSetListOut, ShotTaskSetOut, ) from app.services.module_generation_log_service import log_module_event_file from app.services.llm_billing import ( LlmBillingContext, refund_on_final_failure, charge_llm_credits, validate_retryable_previous_attempt, ) from app.services.resource_accounting_service import SOURCE_MODEL_SHOT_SEGMENT, soft_delete_resources_by_source from app.enums.upload_resource import UploadResourceModuleEnum, UploadResourceSourceModelEnum from app.services.upload_resource import release_upload_resources_by_source from app.services.upload_video_asset_service import ( build_time_node, validate_split_range, validate_upload_video_asset, ) from app.tasks.celery_app import celery_app from app.utils.id_gen import generate_id MODULE = ModuleCodeEnum.SHOT_REPLICATE.value def _now() -> datetime: return datetime.now(timezone.utc) def build_task_set_analysis_billing_context(task_set: ShotReplicateTaskSet) -> LlmBillingContext: return LlmBillingContext( user_id=str(task_set.user_id), owner_type=CreditRecordOwnerType.SHOT_REPLICATE_TASK_SET.value, owner_id=str(task_set.id), attempt_no=int(task_set.analysis_attempt_no or 1), charge_kind=CreditRecordChargeKind.VIDEO_ANALYSIS.value, billing_scene=CreditRecordBillingScene.SHOT_ORIGINAL_VIDEO_ANALYSIS.value, source_module=CreditRecordSourceModule.SHOT_REPLICATE.value, source_project_id=str(task_set.id), source_step_id=str(task_set.id), source_step_code=CreditRecordSourceStepCode.VIDEO_ANALYSIS.value, related_id=str(task_set.id), description_prefix="拆镜复刻原视频AI分析", trace_id=f"shot-task-set-analysis:{task_set.id}:attempt:{int(task_set.analysis_attempt_no or 1)}", ) def build_segment_analysis_billing_context(segment: ShotReplicateSegment) -> LlmBillingContext: return LlmBillingContext( user_id=str(segment.user_id), owner_type=CreditRecordOwnerType.SHOT_REPLICATE_SEGMENT.value, owner_id=str(segment.id), attempt_no=int(segment.analysis_attempt_no or 1), charge_kind=CreditRecordChargeKind.VIDEO_ANALYSIS.value, billing_scene=CreditRecordBillingScene.SHOT_SEGMENT_VIDEO_ANALYSIS.value, source_module=CreditRecordSourceModule.SHOT_REPLICATE.value, source_project_id=str(segment.task_set_id), source_step_id=str(segment.id), source_step_code=CreditRecordSourceStepCode.VIDEO_ANALYSIS.value, related_id=str(segment.id), description_prefix="拆镜复刻片段视频AI分析", trace_id=f"shot-segment-analysis:{segment.id}:attempt:{int(segment.analysis_attempt_no or 1)}", ) def _normalize_suggestions(value: Any) -> list[dict[str, Any]]: if not isinstance(value, list): return [] normalized: list[dict[str, Any]] = [] for idx, item in enumerate(value, start=1): if not isinstance(item, dict): continue start = item.get("拆镜开始秒") end = item.get("拆镜结束秒") try: start_f = float(start) end_f = float(end) except Exception: continue if start_f < 0 or end_f <= start_f: continue normalized.append( { "index": idx, "start_second": start_f, "end_second": end_f, "duration_seconds": round(end_f - start_f, 3), "time_node": str(item.get("拆镜时间节点") or build_time_node(start_f, end_f)), "content": str(item.get("对应时间节点内的内容") or "无"), "category": str(item.get("分类") or "无"), "audience": str(item.get("受众人群") or "无"), "raw": item, } ) return normalized def _task_set_to_out(task_set: ShotReplicateTaskSet, user_name: str | None = None) -> ShotTaskSetOut: data = ShotTaskSetOut.model_validate(task_set) data.user_name = user_name return data def _task_set_to_detail_out(task_set: ShotReplicateTaskSet, user_name: str | None = None) -> ShotTaskSetDetailOut: suggestions = [ShotAISuggestionOut(**{k: v for k, v in item.items() if k != "raw"}) for item in _normalize_suggestions(task_set.ai_suggestion_json)] base = ShotTaskSetDetailOut.model_validate(task_set) base.user_name = user_name base.ai_suggestions = suggestions return base def _segment_to_out(segment: ShotReplicateSegment, project: ModuleGenerationProject | None = None) -> ShotSegmentOut: data = ShotSegmentOut.model_validate(segment) data.segment_name = f"片段{segment.segment_index}" if project: data.module_project_title = project.title data.module_project_status = project.status data.module_project_current_step_code = project.current_step_code data.module_project_flow_version = str(getattr(project, "flow_version", None) or "v1") return data def _segment_to_detail_out(segment: ShotReplicateSegment, project: ModuleGenerationProject | None = None) -> ShotSegmentDetailOut: data = ShotSegmentDetailOut.model_validate(segment) data.segment_name = f"片段{segment.segment_index}" if project: data.module_project_title = project.title data.module_project_status = project.status data.module_project_current_step_code = project.current_step_code data.module_project_flow_version = str(getattr(project, "flow_version", None) or "v1") return data async def get_task_set_for_user( db: AsyncSession, *, task_set_id: str, user: User, for_update: bool = False, ) -> ShotReplicateTaskSet: query = select(ShotReplicateTaskSet).where( ShotReplicateTaskSet.id == task_set_id, ShotReplicateTaskSet.deleted_at.is_(None), ) if not user.is_admin: query = query.where(ShotReplicateTaskSet.user_id == user.id) if for_update: query = query.with_for_update() result = await db.execute(query.limit(1)) task_set = result.scalar_one_or_none() if not task_set: raise HTTPException(status_code=404, detail="拆镜总任务集不存在") return task_set async def get_segment_for_user( db: AsyncSession, *, segment_id: str, user: User, for_update: bool = False, ) -> ShotReplicateSegment: query = select(ShotReplicateSegment).where( ShotReplicateSegment.id == segment_id, ShotReplicateSegment.deleted_at.is_(None), ) if not user.is_admin: query = query.where(ShotReplicateSegment.user_id == user.id) if for_update: query = query.with_for_update() result = await db.execute(query.limit(1)) segment = result.scalar_one_or_none() if not segment: raise HTTPException(status_code=404, detail="拆镜片段不存在") return segment async def create_task_set( db: AsyncSession, *, current_user: User, req: ShotTaskSetCreate, ) -> tuple[ShotReplicateTaskSet, bool]: if req.idempotency_key: existing_result = await db.execute( select(ShotReplicateTaskSet).where( ShotReplicateTaskSet.user_id == current_user.id, ShotReplicateTaskSet.idempotency_key == req.idempotency_key, ShotReplicateTaskSet.deleted_at.is_(None), ).limit(1) ) existing = existing_result.scalar_one_or_none() if existing: # 幂等命中只返回已有任务，不重复消费积分、绑定资源或投递 Celery。 return existing, False asset = validate_upload_video_asset(req.video_url, req.video_duration_seconds) task_set = ShotReplicateTaskSet( id=generate_id(), user_id=current_user.id, title=req.title or "拆镜复刻任务", video_url=asset.url, video_path=str(asset.path), video_duration_seconds=asset.duration_seconds, status=ShotTaskSetStatusEnum.PENDING_ANALYSIS.value, analysis_status=ShotAnalysisStatusEnum.PENDING.value, split_status=ShotSplitStatusEnum.NONE.value, segment_count=0, completed_segment_count=0, failed_segment_count=0, idempotency_key=req.idempotency_key, ) db.add(task_set) await db.flush() await charge_llm_credits(db, build_task_set_analysis_billing_context(task_set)) log_module_event_file( module=MODULE, event_type="SHOT_TASK_SET_CREATED", project_id=task_set.id, user_id=task_set.user_id, message="创建拆镜总任务集", detail={ "task_set_id": task_set.id, "title": task_set.title, "video_url": task_set.video_url, "video_path": task_set.video_path, "video_duration_seconds": task_set.video_duration_seconds, "idempotency_key": task_set.idempotency_key, }, ) return task_set, True def _user_name_filter_subquery(value: str): """后台按用户名筛选时使用的子查询。 只在传入 user_name 时查询 users 表；keyword 不再关联 users，避免后台关键词搜索扩大查询范围。 """ like = f"%{value.strip()}%" return select(User.id).where(User.username.ilike(like)) async def _user_name_map_by_ids(db: AsyncSession, user_ids: set[str]) -> dict[str, str | None]: """一次性查询当前页涉及的用户，避免列表逐条查询用户表。""" if not user_ids: return {} result = await db.execute(select(User.id, User.username).where(User.id.in_(list(user_ids)))) return {user_id: username for user_id, username in result.all()} async def list_task_sets( db: AsyncSession, *, current_user: User, status: str | None = None, analysis_status: str | None = None, split_status: str | None = None, keyword: str | None = None, user_id: str | None = None, user_name: str | None = None, created_start: datetime | None = None, created_end: datetime | None = None, page: int = 1, page_size: int = 20, ) -> ShotTaskSetListOut: """ 拆镜总任务集列表查询。 性能策略： 1. 主列表不 join User，避免 count/list 复杂化。 2. 管理员按 user_name 搜索时使用 IN (SELECT users.id ...) 子查询；keyword 不查询 users 表。 3. 当前页数据取出后，再按 user_id 去重批量查 user_name，用于后台渲染。 """ query = select(ShotReplicateTaskSet).where(ShotReplicateTaskSet.deleted_at.is_(None)) if not current_user.is_admin: query = query.where(ShotReplicateTaskSet.user_id == current_user.id) else: if user_id and user_id.strip(): query = query.where(ShotReplicateTaskSet.user_id == user_id.strip()) if user_name and user_name.strip(): query = query.where(ShotReplicateTaskSet.user_id.in_(_user_name_filter_subquery(user_name))) if created_start: query = query.where(ShotReplicateTaskSet.created_at >= created_start) if created_end: query = query.where(ShotReplicateTaskSet.created_at <= created_end) if status: query = query.where(ShotReplicateTaskSet.status == status) if analysis_status: query = query.where(ShotReplicateTaskSet.analysis_status == analysis_status) if split_status: query = query.where(ShotReplicateTaskSet.split_status == split_status) if keyword and keyword.strip(): like = f"%{keyword.strip()}%" conditions = [ ShotReplicateTaskSet.id.ilike(like), ShotReplicateTaskSet.title.ilike(like), ShotReplicateTaskSet.original_video_content.ilike(like), ShotReplicateTaskSet.original_video_category.ilike(like), ShotReplicateTaskSet.original_video_audience.ilike(like), cast(ShotReplicateTaskSet.ai_suggestion_json, String).ilike(like), ] query = query.where(or_(*conditions)) total_result = await db.execute(select(func.count()).select_from(query.subquery())) total = int(total_result.scalar() or 0) result = await db.execute( query.order_by(ShotReplicateTaskSet.created_at.desc()) .offset((page - 1) * page_size) .limit(page_size) ) task_sets = list(result.scalars().unique().all()) user_name_map: dict[str, str | None] = {} if current_user.is_admin: user_ids = {task_set.user_id for task_set in task_sets if task_set.user_id} user_name_map = await _user_name_map_by_ids(db, user_ids) return ShotTaskSetListOut( total=total, page=page, page_size=page_size, items=[ _task_set_to_out( task_set, user_name_map.get(task_set.user_id) if current_user.is_admin and task_set.user_id else None, ) for task_set in task_sets ], ) async def task_set_detail(db: AsyncSession, *, current_user: User, task_set_id: str) -> ShotTaskSetDetailOut: task_set = await get_task_set_for_user(db, task_set_id=task_set_id, user=current_user) user_result = await db.execute(select(User.username).where(User.id == task_set.user_id).limit(1)) return _task_set_to_detail_out(task_set, user_result.scalar_one_or_none()) async def _next_segment_index(db: AsyncSession, task_set_id: str) -> int: result = await db.execute( select(func.max(ShotReplicateSegment.segment_index)).where( ShotReplicateSegment.task_set_id == task_set_id, ShotReplicateSegment.deleted_at.is_(None), ) ) return int(result.scalar() or 0) + 1 async def refresh_task_set_split_summaries( db: AsyncSession, task_set_ids: set[str] | list[str], *, log_changes: bool = True, ) -> list[dict[str, Any]]: ids = sorted({str(item) for item in task_set_ids if item}) if not ids: return [] task_set_result = await db.execute( select(ShotReplicateTaskSet) .where(ShotReplicateTaskSet.id.in_(ids)) .order_by(ShotReplicateTaskSet.id.asc()) .with_for_update() ) task_sets = list(task_set_result.scalars().all()) if not task_sets: return [] # 项目 AsyncSession 关闭了 autoflush。聚合查询前必须把当前事务中刚修改的 # segment.split_status/deleted_at 等字段落到数据库，否则最后一个片段会少统计一次。 await db.flush() count_result = await db.execute( select( ShotReplicateSegment.task_set_id, func.count(ShotReplicateSegment.id).label("total"), func.sum( case( (ShotReplicateSegment.split_status == ShotSplitStatusEnum.COMPLETED.value, 1), else_=0, ) ).label("completed"), func.sum( case( (ShotReplicateSegment.split_status == ShotSplitStatusEnum.FAILED.value, 1), else_=0, ) ).label("failed"), ) .where( ShotReplicateSegment.task_set_id.in_(ids), ShotReplicateSegment.deleted_at.is_(None), ) .group_by(ShotReplicateSegment.task_set_id) ) count_map = { str(row.task_set_id): (int(row.total or 0), int(row.completed or 0), int(row.failed or 0)) for row in count_result.all() } changes: list[dict[str, Any]] = [] for task_set in task_sets: total, completed, failed = count_map.get(str(task_set.id), (0, 0, 0)) old_status = task_set.status old_split_status = task_set.split_status old_total = int(task_set.segment_count or 0) old_completed = int(task_set.completed_segment_count or 0) old_failed = int(task_set.failed_segment_count or 0) task_set.segment_count = total task_set.completed_segment_count = completed task_set.failed_segment_count = failed if total <= 0: task_set.split_status = ShotSplitStatusEnum.NONE.value if task_set.analysis_status == ShotAnalysisStatusEnum.COMPLETED.value: task_set.status = ShotTaskSetStatusEnum.ANALYSIS_COMPLETED.value elif completed == total: task_set.split_status = ShotSplitStatusEnum.COMPLETED.value task_set.status = ShotTaskSetStatusEnum.SPLIT_COMPLETED.value elif failed == total: task_set.split_status = ShotSplitStatusEnum.FAILED.value task_set.status = ShotTaskSetStatusEnum.FAILED.value elif failed > 0: task_set.split_status = ShotSplitStatusEnum.FAILED.value task_set.status = ShotTaskSetStatusEnum.PARTIAL_FAILED.value else: task_set.split_status = ShotSplitStatusEnum.PROCESSING.value task_set.status = ShotTaskSetStatusEnum.SPLITTING.value changed = ( old_status != task_set.status or old_split_status != task_set.split_status or old_total != total or old_completed != completed or old_failed != failed ) if changed: change = { "task_set_id": str(task_set.id), "user_id": str(task_set.user_id), "from_status": old_status, "to_status": task_set.status, "from_split_status": old_split_status, "to_split_status": task_set.split_status, "from_segment_count": old_total, "segment_count": total, "from_completed_segment_count": old_completed, "completed_segment_count": completed, "from_failed_segment_count": old_failed, "failed_segment_count": failed, } changes.append(change) if log_changes: log_module_event_file( module=MODULE, event_type=ShotReplicateLogEventEnum.SPLIT_STATUS_CHANGED.value, project_id=task_set.id, user_id=task_set.user_id, message="拆镜总任务集拆分状态变更", detail=change, ) return changes async def refresh_task_set_split_summary(db: AsyncSession, task_set_id: str) -> None: await refresh_task_set_split_summaries(db, {task_set_id}) async def create_segments_by_ai( db: AsyncSession, *, current_user: User, task_set_id: str, req: ShotSplitByAIRequest, ) -> ShotSplitByAIOut: task_set = await get_task_set_for_user(db, task_set_id=task_set_id, user=current_user, for_update=True) if task_set.analysis_status != ShotAnalysisStatusEnum.COMPLETED.value: raise HTTPException(status_code=400, detail="原视频分析未完成，不能按 AI 建议拆镜") suggestions = _normalize_suggestions(task_set.ai_suggestion_json) if not suggestions: raise HTTPException(status_code=400, detail="当前没有可用 AI 建议拆镜方案，请使用自定义拆镜") if req.selected_indices: selected_set = {int(x) for x in req.selected_indices} suggestions = [item for item in suggestions if int(item["index"]) in selected_set] if not suggestions: raise HTTPException(status_code=400, detail="selected_indices 没有匹配到可用 AI 建议") # old_result = await db.execute( # select(ShotReplicateSegment).where( # ShotReplicateSegment.task_set_id == task_set.id, # ShotReplicateSegment.source_mode == ShotSegmentSourceModeEnum.AI_SUGGESTION.value, # ShotReplicateSegment.deleted_at.is_(None), # ) # ) # old_segments = list(old_result.scalars().all()) # if old_segments and not req.replace_existing: # raise HTTPException(status_code=409, detail="已存在 AI 建议拆镜片段，如需重拆请传 replace_existing=true") # if old_segments and req.replace_existing: # now = _now() # for segment in old_segments: # segment.deleted_at = now created: list[ShotReplicateSegment] = [] next_index = await _next_segment_index(db, task_set.id) for item in suggestions: start, end, duration = validate_split_range( start_second=item["start_second"], end_second=item["end_second"], video_duration_seconds=task_set.video_duration_seconds, ) segment = ShotReplicateSegment( id=generate_id(), task_set_id=task_set.id, user_id=task_set.user_id, segment_index=next_index, source_mode=ShotSegmentSourceModeEnum.AI_SUGGESTION.value, start_second=start, end_second=end, duration_seconds=duration, time_node=build_time_node(start, end), split_status=ShotSplitStatusEnum.PENDING.value, analysis_status=ShotSegmentAnalysisStatusEnum.NOT_REQUIRED.value, replicate_status=ShotSegmentReplicateStatusEnum.NOT_STARTED.value, original_video_content=task_set.original_video_content, original_video_category=task_set.original_video_category, original_video_audience=task_set.original_video_audience, segment_content=item.get("content"), segment_category=item.get("category"), segment_audience=item.get("audience"), ai_suggestion_json=item.get("raw") or item, split_enqueued_at=_now(), split_celery_task_id=f"shot-split:{uuid.uuid4().hex}", ) db.add(segment) created.append(segment) next_index += 1 task_set.status = ShotTaskSetStatusEnum.SPLITTING.value task_set.split_status = ShotSplitStatusEnum.PROCESSING.value await db.flush() await refresh_task_set_split_summary(db, task_set.id) await db.flush() log_module_event_file( module=MODULE, event_type="SHOT_SPLIT_BY_AI_SUBMITTED", project_id=task_set.id, user_id=task_set.user_id, message="按 AI 建议创建拆镜片段", detail={ "task_set_id": task_set.id, "selected_indices": req.selected_indices, # "replace_existing": req.replace_existing, "created_segment_count": len(created), "segment_ids": [segment.id for segment in created], }, ) return ShotSplitByAIOut( task_set_id=task_set.id, status=task_set.status, split_status=task_set.split_status, created_segment_count=len(created), segments=[_segment_to_out(segment) for segment in created], ) async def prepare_retry_split_segment( db: AsyncSession, *, current_user: User, segment_id: str, force: bool = False, reason: str | None = None, ) -> ShotSegmentSplitRetryOut: """重置失败切片片段，commit 成功后由 API 投递现有 split_one_segment 任务。""" segment = await get_segment_for_user(db, segment_id=segment_id, user=current_user, for_update=True) task_set = await get_task_set_for_user(db, task_set_id=segment.task_set_id, user=current_user, for_update=True) from_split_status = segment.split_status allowed = {ShotSplitStatusEnum.FAILED.value, ShotSplitStatusEnum.RETRY_WAITING.value} reject_reason: str | None = None if task_set.deleted_at is not None or task_set.status == ShotTaskSetStatusEnum.DELETED.value: reject_reason = "拆镜总任务集已删除，不能重试切片" elif segment.deleted_at is not None: reject_reason = "拆镜片段已删除，不能重试切片" elif from_split_status == ShotSplitStatusEnum.PROCESSING.value: reject_reason = "拆镜片段正在切片处理中，不能重复投递" elif from_split_status == ShotSplitStatusEnum.COMPLETED.value: reject_reason = "拆镜片段已切片完成，不支持重切，避免旧切片资源覆盖" elif from_split_status not in allowed and not force: reject_reason = "仅允许失败或等待重试的切片片段重新投递" elif not task_set.video_path: reject_reason = "原视频本地路径为空，不能重试切片" elif not Path(str(task_set.video_path)).exists(): reject_reason = "原视频本地文件不存在，不能重试切片" if reject_reason: log_module_event_file( module=MODULE, event_type=ShotReplicateLogEventEnum.SEGMENT_SPLIT_RETRY_REJECTED.value, project_id=task_set.id, step_id=segment.id, message=reject_reason, detail={ "segment_id": segment.id, "task_set_id": task_set.id, "from_split_status": from_split_status, "force": force, "reason": reason, "status": "rejected", }, ) raise HTTPException(status_code=400, detail=reject_reason) now = _now() segment.split_status = ShotSplitStatusEnum.PENDING.value segment.split_enqueued_at = now segment.split_started_at = None segment.split_lease_until = None segment.split_next_retry_at = None segment.split_retry_count = 0 segment.split_last_error = None segment.split_celery_task_id = f"shot-split:{uuid.uuid4().hex}" task_set.split_error_message = None await refresh_task_set_split_summary(db, task_set.id) await db.flush() # 切片重试只重放本地视频切割，不创建新的 LLM attempt，也不重复消费积分。 # 切片成功后仍会继续原 attempt 的片段分析；显式重新分析才走 reanalyze_segment。 log_module_event_file( module=MODULE, event_type=ShotReplicateLogEventEnum.SEGMENT_SPLIT_RETRY_RECEIVED.value, project_id=task_set.id, step_id=segment.id, message="拆镜片段切片失败重试已重置，等待投递 Celery", detail={ "segment_id": segment.id, "task_set_id": task_set.id, "from_split_status": from_split_status, "to_split_status": segment.split_status, "force": force, "reason": reason, "source_path": task_set.video_path, "celery_task_name": "shot_replicate.split_one_segment", "queue": CeleryQueue.GEN_SHOT_SPLIT.value, "status": "pending", }, ) return ShotSegmentSplitRetryOut( message="切片重试已提交，正在重新切割视频片段", task_set_id=task_set.id, segment_id=segment.id, split_status=segment.split_status, celery_task_name="shot_replicate.split_one_segment", ) async def create_custom_segment( db: AsyncSession, *, current_user: User, task_set_id: str, req: ShotSplitCustomRequest, ) -> ShotSplitCustomOut: task_set = await get_task_set_for_user(db, task_set_id=task_set_id, user=current_user, for_update=True) start, end, duration = validate_split_range( start_second=req.start_second, end_second=req.end_second, video_duration_seconds=task_set.video_duration_seconds, ) next_index = await _next_segment_index(db, task_set.id) segment = ShotReplicateSegment( id=generate_id(), task_set_id=task_set.id, user_id=task_set.user_id, segment_index=next_index, source_mode=ShotSegmentSourceModeEnum.CUSTOM.value, start_second=start, end_second=end, duration_seconds=duration, time_node=build_time_node(start, end), split_status=ShotSplitStatusEnum.PENDING.value, analysis_status=ShotSegmentAnalysisStatusEnum.PENDING.value, replicate_status=ShotSegmentReplicateStatusEnum.NOT_STARTED.value, split_enqueued_at=_now(), split_celery_task_id=f"shot-split:{uuid.uuid4().hex}", ) db.add(segment) task_set.status = ShotTaskSetStatusEnum.SPLITTING.value task_set.split_status = ShotSplitStatusEnum.PROCESSING.value await db.flush() await charge_llm_credits(db, build_segment_analysis_billing_context(segment)) await refresh_task_set_split_summary(db, task_set.id) await db.flush() log_module_event_file( module=MODULE, event_type="SHOT_SPLIT_CUSTOM_SUBMITTED", project_id=task_set.id, step_id=segment.id, user_id=task_set.user_id, message="按用户自定义时间创建拆镜片段", detail={ "task_set_id": task_set.id, "segment_id": segment.id, "start_second": start, "end_second": end, "duration_seconds": duration, "time_node": segment.time_node, }, ) return ShotSplitCustomOut(task_set_id=task_set.id, segment=_segment_to_out(segment)) async def enqueue_segment_split(segment_id: str, *, countdown: int | None = None, recover: bool = False) -> None: if not celery_app: return from app.tasks.shot_replicate_tasks import split_one_segment split_one_segment.apply_async( args=[segment_id], queue=CeleryQueue.GEN_SHOT_SPLIT.value, countdown=countdown, priority=settings.DOWNLOAD_TASK_PRIORITY_RECOVER if recover else settings.DOWNLOAD_TASK_PRIORITY_NORMAL, ) async def list_segments( db: AsyncSession, *, current_user: User, task_set_id: str, source_mode: str | None = None, split_status: str | None = None, analysis_status: str | None = None, replicate_status: str | None = None, page: int = 1, page_size: int = 20, ) -> ShotSegmentListOut: await get_task_set_for_user(db, task_set_id=task_set_id, user=current_user) query = ( select(ShotReplicateSegment, ModuleGenerationProject) .outerjoin(ModuleGenerationProject, ModuleGenerationProject.id == ShotReplicateSegment.module_project_id) .where( ShotReplicateSegment.task_set_id == task_set_id, ShotReplicateSegment.deleted_at.is_(None), ) ) if not current_user.is_admin: query = query.where(ShotReplicateSegment.user_id == current_user.id) if source_mode: query = query.where(ShotReplicateSegment.source_mode == source_mode) if split_status: query = query.where(ShotReplicateSegment.split_status == split_status) if analysis_status: query = query.where(ShotReplicateSegment.analysis_status == analysis_status) if replicate_status: query = query.where(ShotReplicateSegment.replicate_status == replicate_status) total_result = await db.execute(select(func.count()).select_from(query.subquery())) total = int(total_result.scalar() or 0) rows = await db.execute( query.order_by(ShotReplicateSegment.segment_index.asc()) .offset((page - 1) * page_size) .limit(page_size) ) return ShotSegmentListOut( total=total, page=page, page_size=page_size, items=[_segment_to_out(segment, project) for segment, project in rows.all()], ) async def segment_detail(db: AsyncSession, *, current_user: User, segment_id: str) -> ShotSegmentDetailOut: segment = await get_segment_for_user(db, segment_id=segment_id, user=current_user) project: ModuleGenerationProject | None = None if segment.module_project_id: project_result = await db.execute(select(ModuleGenerationProject).where(ModuleGenerationProject.id == segment.module_project_id).limit(1)) project = project_result.scalar_one_or_none() return _segment_to_detail_out(segment, project) async def _delete_linked_replication_project( db: AsyncSession, *, current_user: User, project_id: str, ) -> ShotReplicateDeleteOut: """按项目保存的流程版本分发 V1/V2 删除，供片段和任务集联动删除复用。""" result = await db.execute( select(ModuleGenerationProject) .where( ModuleGenerationProject.id == project_id, ModuleGenerationProject.deleted_at.is_(None), ) .limit(1) ) project = result.scalar_one_or_none() if project and str(getattr(project, "flow_version", None) or "v1") == "v2": from app.services.module_generation_v2.config import SHOT_REPLICATE_V2 from app.services.module_generation_v2.flow_service import delete_project_v2 payload = await delete_project_v2( db, config=SHOT_REPLICATE_V2, current_user=current_user, project_id=project_id, ) return ShotReplicateDeleteOut(**payload) from app.services.shot_replicate_flow_service import delete_shot_replicate_project return await delete_shot_replicate_project( db, current_user=current_user, project_id=project_id, refund_unfinished=False, ) async def delete_segment( db: AsyncSession, *, current_user: User, segment_id: str, ) -> ShotSegmentDeleteOut: """软删除拆镜片段。 只释放用户容量账本记录，不删除 segment_video_path 指向的物理文件。 如果片段已创建复刻项目，则联动调用项目删除逻辑，但用户主动删除不退款； 项目仍有生成中任务时会拒绝删除，避免异步任务继续写回软删数据。 """ query = select(ShotReplicateSegment).where(ShotReplicateSegment.id == segment_id) if not current_user.is_admin: query = query.where(ShotReplicateSegment.user_id == current_user.id) result = await db.execute(query.with_for_update().limit(1)) segment = result.scalar_one_or_none() if not segment: raise HTTPException(status_code=404, detail="拆镜片段不存在") task_set_id = segment.task_set_id module_project_id = segment.module_project_id if segment.deleted_at is not None: return ShotSegmentDeleteOut( message="拆镜片段已删除", segment_id=segment.id, task_set_id=task_set_id, deleted=True, deleted_module_project_id=module_project_id, released_size_bytes=0, ) if segment.split_status == ShotSplitStatusEnum.PROCESSING.value: raise HTTPException(status_code=409, detail="当前拆镜片段正在切割处理中，暂不能删除") if segment.analysis_status == ShotSegmentAnalysisStatusEnum.PROCESSING.value: raise HTTPException(status_code=409, detail="当前拆镜片段正在分析处理中，暂不能删除") if segment.replicate_status == ShotSegmentReplicateStatusEnum.PROCESSING.value: raise HTTPException(status_code=409, detail="当前拆镜片段关联的复刻流程正在处理中，暂不能删除") if segment.source_mode == ShotSegmentSourceModeEnum.CUSTOM.value: await refund_on_final_failure( db, build_segment_analysis_billing_context(segment), error="用户删除自定义拆镜片段，退回片段分析消费积分", ) deleted_at = _now() released_size_bytes = await soft_delete_resources_by_source( db, source_model=SOURCE_MODEL_SHOT_SEGMENT, source_ids=[segment.id], deleted_at=deleted_at, ) upload_release = await release_upload_resources_by_source( db, source_model=UploadResourceSourceModelEnum.SHOT_REPLICATE_SEGMENT.value, source_ids=[segment.id], module=UploadResourceModuleEnum.SHOT_REPLICATE.value, ) pending_delete_resource_ids: list[str] = list(upload_release.get("released_resource_ids") or []) released_size_bytes += int(upload_release.get("released_size_bytes") or 0) upload_resource_released = int(upload_release.get("released") or 0) deleted_module_project_id: str | None = None if module_project_id: project_delete_out = await _delete_linked_replication_project( db, current_user=current_user, project_id=module_project_id, ) deleted_module_project_id = project_delete_out.project_id released_size_bytes += int(project_delete_out.released_size_bytes or 0) upload_resource_released += int(project_delete_out.upload_resource_released or 0) pending_delete_resource_ids.extend(project_delete_out.pending_delete_resource_ids or []) segment.deleted_at = deleted_at segment.replicate_status = ( ShotSegmentReplicateStatusEnum.NOT_STARTED.value if not deleted_module_project_id else ShotSegmentReplicateStatusEnum.FAILED.value ) await refresh_task_set_split_summary(db, task_set_id) await db.flush() log_module_event_file( module=MODULE, event_type="SHOT_SEGMENT_DELETED", project_id=task_set_id, step_id=segment.id, user_id=segment.user_id, message="软删除拆镜片段并释放用户容量账本记录", detail={ "segment_id": segment.id, "task_set_id": task_set_id, "module_project_id": module_project_id, "deleted_module_project_id": deleted_module_project_id, "released_size_bytes": released_size_bytes, "upload_resource_release": {k: v for k, v in upload_release.items() if k != "released_resource_ids"}, "pending_delete_resource_count": len(pending_delete_resource_ids), "physical_file_delete": "after_commit", "media_refund": False, "llm_charge_refund_on_cancel": segment.source_mode == ShotSegmentSourceModeEnum.CUSTOM.value, }, ) return ShotSegmentDeleteOut( message="拆镜片段已删除", segment_id=segment.id, task_set_id=task_set_id, deleted=True, deleted_module_project_id=deleted_module_project_id, released_size_bytes=int(released_size_bytes or 0), upload_resource_released=upload_resource_released, pending_delete_resource_ids=pending_delete_resource_ids, ) async def delete_task_set( db: AsyncSession, *, current_user: User, task_set_id: str, ) -> ShotTaskSetDeleteOut: """软删除整个拆镜任务集。 对外删除入口以 ShotReplicateTaskSet 为边界；内部 ModuleGenerationProject 只作为片段复刻链路被联动软删。这里不 commit、不 rollback、不删除真实文件。 """ task_set = await get_task_set_for_user(db, task_set_id=task_set_id, user=current_user, for_update=True) task_set_id_snapshot = task_set.id user_id_snapshot = task_set.user_id if task_set.analysis_status == ShotAnalysisStatusEnum.PROCESSING.value: raise HTTPException(status_code=409, detail="原视频分析正在处理中，暂不能删除任务集") if task_set.split_status == ShotSplitStatusEnum.PROCESSING.value: raise HTTPException(status_code=409, detail="拆镜切片正在处理中，暂不能删除任务集") segments_result = await db.execute( select(ShotReplicateSegment) .where( ShotReplicateSegment.task_set_id == task_set_id_snapshot, ShotReplicateSegment.deleted_at.is_(None), ) .with_for_update() ) segments = list(segments_result.scalars().all()) for segment in segments: if segment.split_status == ShotSplitStatusEnum.PROCESSING.value: raise HTTPException(status_code=409, detail=f"片段{segment.segment_index}正在切割处理中，暂不能删除任务集") if segment.analysis_status == ShotSegmentAnalysisStatusEnum.PROCESSING.value: raise HTTPException(status_code=409, detail=f"片段{segment.segment_index}正在分析处理中，暂不能删除任务集") if segment.replicate_status == ShotSegmentReplicateStatusEnum.PROCESSING.value: raise HTTPException(status_code=409, detail=f"片段{segment.segment_index}关联复刻流程正在处理中，暂不能删除任务集") # 删除是对未执行/失败任务的最终取消动作；处理中的任务已在上方拦截。 # 这里仅做账务补偿，不改变现有逐项目删除流程。 await refund_on_final_failure( db, build_task_set_analysis_billing_context(task_set), error="用户删除拆镜任务集，退回原视频分析消费积分", ) for segment in segments: if segment.source_mode == ShotSegmentSourceModeEnum.CUSTOM.value: await refund_on_final_failure( db, build_segment_analysis_billing_context(segment), error="用户删除拆镜任务集，退回片段分析消费积分", ) segment_ids = [segment.id for segment in segments] module_project_ids = [segment.module_project_id for segment in segments if segment.module_project_id] deleted_at = _now() released_size_bytes = 0 upload_resource_released = 0 pending_delete_resource_ids: list[str] = [] task_upload_release = await release_upload_resources_by_source( db, source_model=UploadResourceSourceModelEnum.SHOT_REPLICATE_TASK_SET.value, source_ids=[task_set_id_snapshot], module=UploadResourceModuleEnum.SHOT_REPLICATE.value, ) released_size_bytes += int(task_upload_release.get("released_size_bytes") or 0) upload_resource_released += int(task_upload_release.get("released") or 0) pending_delete_resource_ids.extend(task_upload_release.get("released_resource_ids") or []) segment_upload_release = await release_upload_resources_by_source( db, source_model=UploadResourceSourceModelEnum.SHOT_REPLICATE_SEGMENT.value, source_ids=segment_ids, module=UploadResourceModuleEnum.SHOT_REPLICATE.value, ) released_size_bytes += int(segment_upload_release.get("released_size_bytes") or 0) upload_resource_released += int(segment_upload_release.get("released") or 0) pending_delete_resource_ids.extend(segment_upload_release.get("released_resource_ids") or []) deleted_module_project_count = 0 for module_project_id in dict.fromkeys(module_project_ids): project_delete_out = await _delete_linked_replication_project( db, current_user=current_user, project_id=module_project_id, ) deleted_module_project_count += 1 released_size_bytes += int(project_delete_out.released_size_bytes or 0) upload_resource_released += int(project_delete_out.upload_resource_released or 0) pending_delete_resource_ids.extend(project_delete_out.pending_delete_resource_ids or []) task_set.deleted_at = deleted_at task_set.status = ShotTaskSetStatusEnum.DELETED.value for segment in segments: segment.deleted_at = deleted_at segment.replicate_status = ShotSegmentReplicateStatusEnum.FAILED.value if segment.module_project_id else segment.replicate_status await db.flush() log_module_event_file( module=MODULE, event_type="SHOT_TASK_SET_DELETED", project_id=task_set_id_snapshot, user_id=user_id_snapshot, message="软删除拆镜任务集并标记上传资源待物理删除", detail={ "task_set_id": task_set_id_snapshot, "segment_count": len(segment_ids), "module_project_count": deleted_module_project_count, "released_size_bytes": released_size_bytes, "upload_resource_released": upload_resource_released, "pending_delete_resource_count": len(pending_delete_resource_ids), "task_upload_release": {k: v for k, v in task_upload_release.items() if k != "released_resource_ids"}, "segment_upload_release": {k: v for k, v in segment_upload_release.items() if k != "released_resource_ids"}, "physical_file_delete": "after_commit", "media_refund": False, "llm_charge_refund_on_cancel": True, }, ) return ShotTaskSetDeleteOut( message="拆镜任务集已删除", task_set_id=task_set_id_snapshot, deleted=True, deleted_segment_count=len(segment_ids), deleted_module_project_count=deleted_module_project_count, released_size_bytes=int(released_size_bytes or 0), upload_resource_released=upload_resource_released, pending_delete_resource_ids=pending_delete_resource_ids, ) async def prepare_reanalyze_task_set( db: AsyncSession, *, current_user: User, task_set_id: str, reason: str | None = None, ) -> ShotReanalyzeOut: """仅对已失败且旧账务已关闭的原视频分析创建新 attempt。""" task_set = await get_task_set_for_user(db, task_set_id=task_set_id, user=current_user, for_update=True) previous_attempt_no = max(1, int(task_set.analysis_attempt_no or 1)) if task_set.analysis_status != ShotAnalysisStatusEnum.FAILED.value: log_module_event_file( module=MODULE, event_type=ShotReplicateLogEventEnum.TASK_SET_REANALYZE_REJECTED.value, project_id=task_set.id, user_id=task_set.user_id, message="原视频分析不是失败终态，拒绝再次分析", detail={ "task_set_id": task_set.id, "analysis_status": task_set.analysis_status, "analysis_attempt_no": previous_attempt_no, "reason": reason, }, event_status="rejected", ) raise HTTPException( status_code=409, detail=f"只有分析失败的原视频任务才能重新分析，当前状态：{task_set.analysis_status}", ) previous_context = build_task_set_analysis_billing_context(task_set) previous_validation = await validate_retryable_previous_attempt(db, previous_context) if not previous_validation.can_execute: log_module_event_file( module=MODULE, event_type=ShotReplicateLogEventEnum.TASK_SET_REANALYZE_REJECTED.value, project_id=task_set.id, user_id=task_set.user_id, message="原视频旧分析 attempt 账务未关闭，拒绝再次分析", detail={ "task_set_id": task_set.id, "analysis_attempt_no": previous_attempt_no, "ledger_state": previous_validation.state.value, "ledger_reason": previous_validation.reason, "reason": reason, }, event_status="rejected", ) raise HTTPException( status_code=409, detail=( "上一次原视频分析的冻结积分尚未完成释放或账务状态异常，" f"当前账务状态：{previous_validation.state.value}" ), ) task_set.status = ShotTaskSetStatusEnum.PENDING_ANALYSIS.value task_set.analysis_status = ShotAnalysisStatusEnum.PENDING.value task_set.analysis_attempt_no = previous_attempt_no + 1 task_set.analysis_claim_token = None task_set.analysis_started_at = None task_set.analysis_lease_until = None task_set.analysis_error_message = None task_set.original_video_content = None task_set.original_video_category = None task_set.original_video_audience = None task_set.ai_suggestion_json = None task_set.analysis_raw_json = None task_set.analysis_result_json = None await db.flush() await charge_llm_credits(db, build_task_set_analysis_billing_context(task_set)) log_module_event_file( module=MODULE, event_type=ShotReplicateLogEventEnum.TASK_SET_REANALYZE_RECEIVED.value, project_id=task_set.id, user_id=task_set.user_id, message="原视频再次分析已重置状态", detail={ "task_set_id": task_set.id, "previous_analysis_attempt_no": previous_attempt_no, "analysis_attempt_no": int(task_set.analysis_attempt_no), "previous_ledger_state": previous_validation.state.value, "reason": reason, "video_url": task_set.video_url, }, ) return ShotReanalyzeOut( message="原视频再次分析任务已准备投递", task_set_id=task_set.id, segment_id=None, analysis_attempt_no=max(1, int(task_set.analysis_attempt_no or 1)), analysis_status=task_set.analysis_status, celery_task_name="shot_replicate.analyze_original_video", ) async def prepare_reanalyze_segment( db: AsyncSession, *, current_user: User, segment_id: str, reason: str | None = None, ) -> ShotReanalyzeOut: """仅对已失败且旧账务已关闭的自定义切片分析创建新 attempt。""" segment = await get_segment_for_user(db, segment_id=segment_id, user=current_user, for_update=True) if segment.split_status != ShotSplitStatusEnum.COMPLETED.value: raise HTTPException(status_code=409, detail="当前片段还未切割完成，不能再次分析") if not segment.segment_video_url: raise HTTPException(status_code=409, detail="当前片段缺少 segment_video_url，不能再次分析") if segment.source_mode != ShotSegmentSourceModeEnum.CUSTOM.value: raise HTTPException(status_code=409, detail="只有自定义切片视频支持重新分析") previous_attempt_no = max(1, int(segment.analysis_attempt_no or 1)) if segment.analysis_status != ShotSegmentAnalysisStatusEnum.FAILED.value: log_module_event_file( module=MODULE, event_type=ShotReplicateLogEventEnum.SEGMENT_REANALYZE_REJECTED.value, project_id=segment.task_set_id, step_id=segment.id, user_id=segment.user_id, message="切片视频分析不是失败终态，拒绝再次分析", detail={ "segment_id": segment.id, "analysis_status": segment.analysis_status, "analysis_attempt_no": previous_attempt_no, "reason": reason, }, event_status="rejected", ) raise HTTPException( status_code=409, detail=f"只有分析失败的切片视频才能重新分析，当前状态：{segment.analysis_status}", ) previous_context = build_segment_analysis_billing_context(segment) previous_validation = await validate_retryable_previous_attempt(db, previous_context) if not previous_validation.can_execute: log_module_event_file( module=MODULE, event_type=ShotReplicateLogEventEnum.SEGMENT_REANALYZE_REJECTED.value, project_id=segment.task_set_id, step_id=segment.id, user_id=segment.user_id, message="切片视频旧分析 attempt 账务未关闭，拒绝再次分析", detail={ "segment_id": segment.id, "task_set_id": segment.task_set_id, "analysis_attempt_no": previous_attempt_no, "ledger_state": previous_validation.state.value, "ledger_reason": previous_validation.reason, "reason": reason, }, event_status="rejected", ) raise HTTPException( status_code=409, detail=( "上一次切片视频分析的冻结积分尚未完成释放或账务状态异常，" f"当前账务状态：{previous_validation.state.value}" ), ) segment.analysis_status = ShotSegmentAnalysisStatusEnum.PENDING.value segment.analysis_attempt_no = previous_attempt_no + 1 segment.analysis_claim_token = None segment.analysis_started_at = None segment.analysis_lease_until = None segment.analysis_error_message = None segment.analysis_json = None segment.original_video_content = None segment.original_video_category = None segment.original_video_audience = None segment.segment_content = None segment.segment_category = None segment.segment_audience = None await db.flush() await charge_llm_credits(db, build_segment_analysis_billing_context(segment)) log_module_event_file( module=MODULE, event_type=ShotReplicateLogEventEnum.SEGMENT_REANALYZE_RECEIVED.value, project_id=segment.task_set_id, step_id=segment.id, user_id=segment.user_id, message="切片视频再次分析已重置状态", detail={ "segment_id": segment.id, "task_set_id": segment.task_set_id, "previous_analysis_attempt_no": previous_attempt_no, "analysis_attempt_no": int(segment.analysis_attempt_no), "previous_ledger_state": previous_validation.state.value, "reason": reason, "video_url": segment.segment_video_url, }, ) return ShotReanalyzeOut( message="切片视频再次分析任务已准备投递", task_set_id=segment.task_set_id, segment_id=segment.id, analysis_attempt_no=max(1, int(segment.analysis_attempt_no or 1)), analysis_status=segment.analysis_status, celery_task_name="shot_replicate.analyze_custom_segment_video", ) async def mark_task_set_analysis_dispatch_failed( db: AsyncSession, *, current_user: User, task_set_id: str, expected_attempt_no: int, error_message: str, ) -> bool: task_set = await get_task_set_for_user(db, task_set_id=task_set_id, user=current_user, for_update=True) if int(task_set.analysis_attempt_no or 1) != int(expected_attempt_no): return False if ( task_set.analysis_status == ShotAnalysisStatusEnum.PROCESSING.value and task_set.analysis_claim_token and task_set.analysis_lease_until and task_set.analysis_lease_until > _now() ): return False if task_set.analysis_status in (ShotAnalysisStatusEnum.COMPLETED.value, ShotAnalysisStatusEnum.FAILED.value): return False task_set.status = ShotTaskSetStatusEnum.ANALYSIS_FAILED.value task_set.analysis_status = ShotAnalysisStatusEnum.FAILED.value task_set.analysis_claim_token = None task_set.analysis_lease_until = None task_set.analysis_error_message = error_message await refund_on_final_failure(db, build_task_set_analysis_billing_context(task_set), error=error_message) log_module_event_file( module=MODULE, event_type=ShotReplicateLogEventEnum.CELERY_DISPATCH_FAILED.value, project_id=task_set.id, user_id=task_set.user_id, message=error_message, detail={"task_set_id": task_set.id, "reason": "analysis_dispatch_failed"}, error=error_message, ) return True async def mark_segment_analysis_dispatch_failed( db: AsyncSession, *, current_user: User, segment_id: str, expected_attempt_no: int, error_message: str, ) -> bool: segment = await get_segment_for_user(db, segment_id=segment_id, user=current_user, for_update=True) if int(segment.analysis_attempt_no or 1) != int(expected_attempt_no): return False if ( segment.analysis_status == ShotSegmentAnalysisStatusEnum.PROCESSING.value and segment.analysis_claim_token and segment.analysis_lease_until and segment.analysis_lease_until > _now() ): return False if segment.analysis_status in (ShotSegmentAnalysisStatusEnum.COMPLETED.value, ShotSegmentAnalysisStatusEnum.FAILED.value): return False segment.analysis_status = ShotSegmentAnalysisStatusEnum.FAILED.value segment.analysis_claim_token = None segment.analysis_lease_until = None segment.analysis_error_message = error_message await refund_on_final_failure(db, build_segment_analysis_billing_context(segment), error=error_message) log_module_event_file( module=MODULE, event_type=ShotReplicateLogEventEnum.CELERY_DISPATCH_FAILED.value, project_id=segment.task_set_id, step_id=segment.id, user_id=segment.user_id, message=error_message, detail={"segment_id": segment.id, "reason": "analysis_dispatch_failed"}, error=error_message, ) return True async def mark_custom_segment_split_dispatch_failed( db: AsyncSession, *, current_user: User, segment_id: str, error_message: str, ) -> None: segment = await get_segment_for_user(db, segment_id=segment_id, user=current_user, for_update=True) if segment.split_status not in (ShotSplitStatusEnum.COMPLETED.value, ShotSplitStatusEnum.FAILED.value): segment.split_status = ShotSplitStatusEnum.FAILED.value segment.split_claim_token = None segment.split_lease_until = None segment.split_next_retry_at = None segment.split_last_error = error_message # 切片投递失败不改变 LLM attempt 的冻结状态。用户重试切片时继续沿用 # 原场景积分消费；只有片段分析最终失败或用户删除片段时才按原来源退款。 log_module_event_file( module=MODULE, event_type=ShotReplicateLogEventEnum.CELERY_DISPATCH_FAILED.value, project_id=segment.task_set_id, step_id=segment.id, user_id=segment.user_id, message=error_message, detail={"segment_id": segment.id, "reason": "split_dispatch_failed"}, error=error_message, )