diff --git a/apps/api/app/api/routes/generation_preview.py b/apps/api/app/api/routes/generation_preview.py index b518b16d4..0a45b678e 100755 --- a/apps/api/app/api/routes/generation_preview.py +++ b/apps/api/app/api/routes/generation_preview.py @@ -650,11 +650,26 @@ def get_preview_generation_task( if not getattr(task, "is_preview", False): raise HTTPException(status_code=404, detail=f"预览任务 {task_id} 不存在") - # 查询生成的视频(取第一个) + # 查询生成的视频(取第一个)。 + # #2024: 渲染完成后先进入 awaiting_cover(未入成品库),此时预览也应可见, + # 从 extra_meta["rendered_output"] 读取视频 URL。 generated_videos = [] status_val = task.status.value if hasattr(task.status, "value") else str(task.status) if status_val == "completed": list_use_case = ListGeneratedVideosByTaskUseCase(generated_video_repository) generated_videos = list_use_case.execute(task_id) + elif status_val == "awaiting_cover": + # 用 extra_meta 中的渲染信息组装一个轻量视频对象给前端预览播放 + _meta = getattr(task, "extra_meta", {}) or {} + _ro = _meta.get("rendered_output") or {} + if _ro.get("file_url"): + + class _PreviewVideo: + def __init__(self, ro): + self.file_url = ro.get("file_url", "") + self.duration = float(ro.get("duration") or 0.0) + self.file_size = int(ro.get("file_size") or 0) + + generated_videos = [_PreviewVideo(_ro)] return _to_preview_response(task, generated_videos=generated_videos) diff --git a/apps/api/app/api/routes/generation_tasks.py b/apps/api/app/api/routes/generation_tasks.py index 560a81750..1d909ca9f 100755 --- a/apps/api/app/api/routes/generation_tasks.py +++ b/apps/api/app/api/routes/generation_tasks.py @@ -31,6 +31,8 @@ from app.schemas.generation_task import ( BatchGenerationTaskResponse, ConfirmGenerationRequest, CreateGenerationTaskRequest, + FinalizeGenerationRequest, + FinalizeGenerationResponse, GenerationTaskResponse, ListGenerationTasksResponse, ) @@ -891,8 +893,14 @@ def confirm_generation( if source_task.project_id: check_project_access(source_task.project_id, authenticated_user.user.id, project_repository) - # 3. 如果预览任务已完成,检查分辨率一致性后复用产物(秒出) - if source_task.is_completed and getattr(source_task, "is_preview", False): + # 3. 如果预览任务已完成渲染(completed 或 awaiting_cover),检查分辨率一致性后复用产物(秒出)。 + # #2024: 渲染完成先进入 awaiting_cover(等 Step5 finalize 入库), + # confirm 时不再直接 finalize——仍创建 is_preview=False 的正式任务,复用预览渲染产物。 + _preview_done = getattr(source_task, "is_preview", False) and source_task.status.value in ( + "completed", + "awaiting_cover", + ) + if _preview_done: # 校验请求的分辨率是否与预览实际渲染的分辨率一致 req_w = request.output_width or 0 req_h = request.output_height or 0 @@ -907,13 +915,25 @@ def confirm_generation( confirmed_title_config = dict(getattr(source_task, "title_config", {}) or {}) confirmed_title_config["text"] = request.custom_title.strip() + # #2024: mark_confirmed 会把 is_preview 翻转为 False、同步标题/分辨率/封面, + # 但不再自动 mark_completed——任务停留在 awaiting_cover,等待用户 Step5 选封面后调 finalize。 source_task.mark_confirmed( cover_url=request.cover_url, output_width=request.output_width, output_height=request.output_height, title_config=confirmed_title_config, ) + # 若预览任务此时是 completed(历史数据/旧 worker),回退到 awaiting_cover 统一流程 + if source_task.status.value == "completed": + try: + from packages.domain.generation_task import GenerationTaskStatus + + source_task.status = GenerationTaskStatus.AWAITING_COVER + source_task.completed_at = None + except Exception: + pass generation_task_repository.update(source_task) + db.commit() # 同步标题到 EditPlan.config # #1970:确认生成复用预览计划,dedup_enabled 沿用计划已有值,不在此覆盖 @@ -926,7 +946,7 @@ def confirm_generation( ) logger.info( - "[确认生成] 复用预览产物: task_id=%s, user_id=%s", + "[确认生成] 复用预览产物(等待 finalize): task_id=%s, user_id=%s", task_id, authenticated_user.user.id, ) @@ -995,6 +1015,55 @@ def confirm_generation( ) +@router.post("/tasks/{task_id}/finalize", response_model=FinalizeGenerationResponse) +def finalize_generation_task( + task_id: str, + request: FinalizeGenerationRequest, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + generation_task_repository: Any = Depends(get_generation_task_repository), + generated_video_repository: Any = Depends(get_generated_video_repository), + project_repository: Any = Depends(get_project_repository), + storage_service: OSSStorageService = Depends(get_storage_service), + db: Session = Depends(get_db_session), +) -> FinalizeGenerationResponse: + """#2024: Step5 点「完成」时调用——将 awaiting_cover 状态的任务正式入库+绑定封面。 + + - 任务必须处于 awaiting_cover 状态(渲染+上传已完成、封面候选已就绪)。 + - cover_url 为空则使用任务自动截帧/智能封面;非空则绑定为最终封面。 + - 幂等:已 finalize 的任务直接返回已有视频记录。 + - 成功后任务推进到 completed,返回成品视频 ID + 可播放 URL。 + """ + from app.services.generation_finalize_service import ( + GenerationFinalizeError, + GenerationFinalizeService, + ) + + task = generation_task_repository.get(task_id) + if task is None: + raise HTTPException(status_code=404, detail=f"GenerationTask {task_id} not found") + if task.project_id: + check_project_access(task.project_id, authenticated_user.user.id, project_repository) + + service = GenerationFinalizeService(db) + try: + video = service.finalize_task( + task_id=task_id, + user_id=authenticated_user.user.id, + cover_url=request.cover_url or None, + ) + except GenerationFinalizeError as e: + raise HTTPException(status_code=e.status_code, detail=str(e)) from e + + download_url = storage_service.get_download_url(video.file_url, expires_seconds=86400) + return FinalizeGenerationResponse( + video_id=video.id, + cover_url=video.thumbnail_url or "", + file_url=download_url, + status="success", + is_duplicate=bool(video.is_duplicate), + ) + + @router.get("/tasks", response_model=ListGenerationTasksResponse) def list_generation_tasks( authenticated_user: AuthenticatedUser = Depends(get_current_user), diff --git a/apps/api/app/api/routes/task_center.py b/apps/api/app/api/routes/task_center.py index cc5188457..a6edeb189 100755 --- a/apps/api/app/api/routes/task_center.py +++ b/apps/api/app/api/routes/task_center.py @@ -63,6 +63,8 @@ def _generation_step(task) -> str: return "等待 Worker 执行" if s == "running": return "正在生成成片" + if s == "awaiting_cover": + return "等待确认封面" if s == "completed": return "生成完成" if s == "failed": @@ -129,7 +131,7 @@ def _validate_status(status: str | None) -> str | None: """校验状态值合法性。""" if status is None: return None - valid = {"pending", "running", "completed", "failed", "cancelled"} + valid = {"pending", "running", "awaiting_cover", "completed", "failed", "cancelled"} if status not in valid: raise HTTPException( status_code=400, @@ -151,7 +153,9 @@ def _clamp_page_size(page_size: int) -> int: @router.get("/tasks", response_model=ListTasksResponse) def list_user_tasks( - status: str | None = Query(None, description="按状态筛选:pending/running/completed/failed/cancelled"), + status: str | None = Query( + None, description="按状态筛选:pending/running/awaiting_cover/completed/failed/cancelled" + ), task_type: str | None = Query(None, description="按任务类型筛选:generation/ingest"), page: int = Query(1, ge=1, description="页码,从1开始"), page_size: int = Query(DEFAULT_PAGE_SIZE, ge=1, le=MAX_PAGE_SIZE, description="每页数量"), @@ -248,7 +252,9 @@ def retry_task_by_id( @router.get("/projects/{project_id}/tasks", response_model=ListProjectTasksResponse) def list_project_tasks( project_id: str, - status: str | None = Query(None, description="按状态筛选:pending/running/completed/failed/cancelled"), + status: str | None = Query( + None, description="按状态筛选:pending/running/awaiting_cover/completed/failed/cancelled" + ), task_type: str | None = Query(None, description="按任务类型筛选:generation/ingest"), page: int = Query(1, ge=1, description="页码,从1开始"), page_size: int = Query(DEFAULT_PAGE_SIZE, ge=1, le=MAX_PAGE_SIZE, description="每页数量"), diff --git a/apps/api/app/schemas/generation_task.py b/apps/api/app/schemas/generation_task.py index afab39fa8..1fb2f66e3 100755 --- a/apps/api/app/schemas/generation_task.py +++ b/apps/api/app/schemas/generation_task.py @@ -13,6 +13,24 @@ class ConfirmGenerationRequest(BaseModel): custom_title: str = Field(default="", description="用户自定义标题文本,非空时同步到任务和编辑计划") +class FinalizeGenerationRequest(BaseModel): + """Step5 点「完成」请求体:用户选定封面后,正式将视频入成品库。""" + + cover_url: str = Field( + default="", description="用户选定的封面图片 URL;为空则使用任务默认 cover_url(自动截帧/智能封面)" + ) + + +class FinalizeGenerationResponse(BaseModel): + """finalize 响应:返回新创建的成品库视频信息。""" + + video_id: str = Field(description="新创建的成品视频 ID") + cover_url: str = Field(default="", description="最终绑定的封面 URL") + file_url: str = Field(default="", description="成品视频 OSS URL") + status: str = Field(default="success", description="success=新建成功;already_finalized=幂等返回已有记录") + is_duplicate: bool = Field(default=False, description="是否被判定为与历史成片重复") + + class CreateGenerationTaskRequest(BaseModel): """创建生成任务请求。 diff --git a/apps/api/app/services/generation_finalize_service.py b/apps/api/app/services/generation_finalize_service.py new file mode 100644 index 000000000..9f2780299 --- /dev/null +++ b/apps/api/app/services/generation_finalize_service.py @@ -0,0 +1,119 @@ +"""视频生成任务 finalize 服务(#2024)。 + +Worker 渲染+上传完成后不再自动入库,标记为 awaiting_cover;用户在 Step5 选好封面 +点「完成」时由 API 调用本服务:创建 GeneratedVideo 成品库记录(复用 worker 预计算 +的查重结果)、绑定封面、推进任务到 completed。 + +与 AI 数字人 ``ai_avatar_render_service.finalize_job`` 模式一致, +只是走 GenerationTask 而非 AiAvatarRenderJob。 +""" + +from __future__ import annotations + +import logging +from datetime import UTC, datetime +from typing import Optional + +from sqlalchemy.orm import Session + +logger = logging.getLogger(__name__) + + +class GenerationFinalizeError(Exception): + """finalize 业务错误,code 供 API 层映射 HTTP 状态码。""" + + def __init__(self, message: str, code: str = "FinalizeError", status_code: int = 400): + super().__init__(message) + self.code = code + self.status_code = status_code + + +class GenerationFinalizeService: + def __init__(self, db: Session): + self.db = db + + def finalize_task(self, task_id: str, user_id: str, cover_url: Optional[str] = None): + """执行 finalize:状态校验 → 幂等 → 绑定封面 → 入库 → 推进 completed。 + + Returns: + GeneratedVideo 领域对象 + """ + from packages.adapters.sqlalchemy_impl.generated_video_repository import ( + SQLAlchemyGeneratedVideoRepository, + ) + from packages.adapters.sqlalchemy_impl.generation_task_repository import ( + SQLAlchemyGenerationTaskRepository, + ) + from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel + from packages.application.generated_video_finalize import finalize_generated_video + + task_repo = SQLAlchemyGenerationTaskRepository(self.db) + video_repo = SQLAlchemyGeneratedVideoRepository(self.db) + + task = task_repo.get(task_id) + if task is None: + raise GenerationFinalizeError(f"任务 {task_id} 不存在", "TaskNotFound", 404) + + # ── 幂等:已入库直接返回 ───────────────────────────────── + existing = self.db.query(GeneratedVideoModel).filter(GeneratedVideoModel.generation_task_id == task_id).first() + if existing is not None: + logger.info("[finalize] 幂等命中 task=%s video=%s", task_id, existing.id) + if cover_url and cover_url.strip() and existing.thumbnail_url != cover_url.strip(): + existing.thumbnail_url = cover_url.strip() + task.cover_url = cover_url.strip() + self.db.commit() + if task.status.value != "completed": + try: + task.mark_completed(result_count=1) + if cover_url and cover_url.strip(): + task.cover_url = cover_url.strip() + task_repo.update(task) + self.db.commit() + except Exception as e: + logger.warning("[finalize] 幂等补 mark_completed 失败: %s", e) + self.db.rollback() + return video_repo.get(existing.id) + + # ── 状态校验 ───────────────────────────────────────────── + if task.status.value != "awaiting_cover": + raise GenerationFinalizeError( + f"任务当前状态 {task.status.value},无法 finalize(需 awaiting_cover)", + "InvalidTaskStatus", + 400, + ) + + # ── 封面 ───────────────────────────────────────────────── + effective_cover = (cover_url or "").strip() if cover_url else (task.cover_url or "").strip() + + # ── 入库+查重(复用 worker 预计算结果) ────────────────── + try: + result = finalize_generated_video( + task=task, + session=self.db, + effective_cover_url=effective_cover, + ) + except ValueError as e: + raise GenerationFinalizeError(str(e), "RenderedOutputMissing", 400) from e + + video_id = result["video_id"] + + # ── 推进任务 ───────────────────────────────────────────── + task.mark_completed(result_count=1) + task.cover_url = effective_cover + # 清理 rendered_output(体积较大,入库后不再需要) + meta = dict(task.extra_meta or {}) + meta.pop("rendered_output", None) + task.extra_meta = meta + task.updated_at = datetime.now(UTC) + task_repo.update(task) + self.db.commit() + + video = video_repo.get(video_id) + logger.info( + "[finalize] task=%s finalized -> video=%s cover=%s dup=%s", + task_id, + video_id, + bool(effective_cover), + result.get("is_duplicate", False), + ) + return video diff --git a/apps/worker/video_processing/dedup.py b/apps/worker/video_processing/dedup.py index cd5011f8b..8e9c2d0d3 100755 --- a/apps/worker/video_processing/dedup.py +++ b/apps/worker/video_processing/dedup.py @@ -345,6 +345,32 @@ class VideoFingerprint: ], } + @classmethod + def from_dict(cls, data: dict) -> "VideoFingerprint": + """从 to_dict() 序列化结果重建 VideoFingerprint(供 finalize 复用 worker 预计算指纹)。""" + + chunks_raw = data.get("chunks") or [] + chunks: list[FingerprintChunk] = [] + for c in chunks_raw: + chunks.append( + FingerprintChunk( + start_time_ms=int(c.get("start_time_ms", 0)), + end_time_ms=int(c.get("end_time_ms", 0)), + phash_binary=str(c.get("phash_binary", "")), + color_histogram=[float(v) for v in (c.get("color_histogram") or [])], + frame_count=int(c.get("frame_count", 0)), + ) + ) + resolution_raw = data.get("resolution") or [1280, 720] + return cls( + md5=str(data.get("md5", "")), + keyframe_phashes=list(data.get("keyframe_phashes") or []), + color_histograms=[[float(v) for v in h] for h in (data.get("color_histograms") or [])], + duration=float(data.get("duration") or 0.0), + resolution=(int(resolution_raw[0]), int(resolution_raw[1])) if len(resolution_raw) >= 2 else (1280, 720), + chunks=chunks, + ) + def to_chunk_models(self, video_id: str, project_id: str, user_id: str = "") -> list[VideoFingerprintChunkModel]: """将分片数据转为 SQLAlchemy Model 列表,用于批量写入 video_fingerprint_chunks 表。""" models = [] diff --git a/apps/worker/video_processing/dedup_helpers.py b/apps/worker/video_processing/dedup_helpers.py index 48ab1e2c2..5e94d0673 100755 --- a/apps/worker/video_processing/dedup_helpers.py +++ b/apps/worker/video_processing/dedup_helpers.py @@ -1,10 +1,15 @@ -"""查重辅助函数 — 从 generation.py 提取的 GeneratedVideo 记录 + 查重逻辑. +"""查重辅助函数 — 渲染阶段指纹/查重预计算 + 兼容旧入库函数。 -供 generate_video 共同复用, -创建 GeneratedVideo 记录后计算指纹并执行项目级 + 批次内查重。 +#2024: Worker 渲染+上传完成后**不直接创建 GeneratedVideo 成品记录**,改为: +1. ``compute_render_fingerprint_and_dedup``: 从本地视频计算指纹+查重(历史+批次), + 返回可序列化 dict(含 fingerprint_chunks),由 worker 写入 + ``GenerationTask.extra_meta["rendered_output"]``; +2. ``create_video_record_and_dedup``: 保留兼容——当传入 ``video_path`` 时会从本地视频 + 计算指纹+查重并直接创建 GeneratedVideo 记录(供测试/旧路径使用); + 当仅传 ``pre_dedup_result`` 时复用预计算结果,不再访问本地视频。 -v2: 两阶段持久化 — 先计算所有查重数据,再一次性 commit, -避免中间异常导致 duplicate_rate 等字段缺失。 +finalize 入口走 ``packages/application/generated_video_finalize.py`` 的 +``finalize_generated_video``,不依赖本模块中数据库以外的 worker-only 逻辑。 """ from __future__ import annotations @@ -17,6 +22,149 @@ from sqlalchemy.orm import Session logger = logging.getLogger(__name__) +def _safe_parse_fps(raw) -> float: + if raw is None: + return 25.0 + if isinstance(raw, (int, float)): + return float(raw) + s = str(raw).strip() + if "/" in s: + try: + num, den = s.split("/", 1) + return float(num) / float(den) if float(den) != 0 else 25.0 + except (ValueError, ZeroDivisionError): + pass + try: + return float(s) + except (ValueError, TypeError): + return 25.0 + + +def _compute_from_local( + *, + video_path: str, + generation_task_id: str, + project_id: str, + user_id: str, + batch_id: str, + session: Session, +) -> dict: + """从本地视频计算指纹+查重,返回可序列化结果 dict(不创建 DB 记录)。""" + from video_processing.dedup import VideoDeduplicator + from video_processing.ffmpeg_utils import probe_video_info + + result: dict = { + "fingerprint_dict": None, + "fingerprint_chunks": None, + "duration": 0.0, + "width": 1280, + "height": 720, + "fps": 25.0, + "is_duplicate": False, + "duplicate_of": None, + "duplicate_rate": None, + "match_count": None, + "visual_similarity": None, + "video_fingerprint_md5": "", + "batch_similarity": None, + } + try: + info = probe_video_info(video_path) + result["duration"] = float(info.get("duration") or 0.0) + result["width"] = int(info.get("width") or 1280) + result["height"] = int(info.get("height") or 720) + result["fps"] = _safe_parse_fps(info.get("fps")) + except Exception as info_err: + logger.warning("probe_video_info failed for task %s: %s", generation_task_id, info_err) + + try: + deduplicator = VideoDeduplicator() + fingerprint = deduplicator.compute_fingerprint(video_path) + fp_dict = fingerprint.to_dict() + result["fingerprint_dict"] = fp_dict + result["video_fingerprint_md5"] = fingerprint.md5 or "" + result["fingerprint_chunks"] = [ + { + "start_time_ms": c.start_time_ms, + "end_time_ms": c.end_time_ms, + "phash_binary": c.phash_binary, + "color_histogram": [float(v) for v in c.color_histogram], + "frame_count": c.frame_count, + } + for c in fingerprint.chunks + ] + + # 用 placeholder_id 占位(还没有真正的 video_id,不影响查重逻辑—— + # 因为查重排除的是 GeneratedVideo 表中的记录) + placeholder_id = f"pre-{generation_task_id}" + duration_sec = fingerprint.duration if fingerprint.duration else 0 + duplicate_result = deduplicator.check_duplicate( + fingerprint, + project_id, + session, + scope="user", + user_id=user_id, + duration_sec=duration_sec, + exclude_video_id=placeholder_id, + ) + batch_sim: float | None = None + if not duplicate_result and batch_id: + duplicate_result = deduplicator.check_batch_duplicate(fingerprint, batch_id, placeholder_id, session) + if duplicate_result: + batch_sim = float(duplicate_result.get("similarity", 0.0)) + result["batch_similarity"] = batch_sim + if duplicate_result: + result["is_duplicate"] = True + result["duplicate_of"] = duplicate_result["duplicate_of"] + else: + result["is_duplicate"] = False + + try: + rate_result = deduplicator.compute_duplicate_rate( + fingerprint, + project_id, + placeholder_id, + session, + scope="user", + user_id=user_id, + ) + result["duplicate_rate"] = rate_result.get("duplicate_rate") + result["match_count"] = rate_result.get("match_count") + result["visual_similarity"] = rate_result.get("visual_similarity") + except Exception as rate_err: + logger.warning("compute_duplicate_rate failed for task %s: %s", generation_task_id, rate_err) + except Exception as fp_err: + logger.warning("Fingerprint compute failed for task %s: %s", generation_task_id, fp_err) + + return result + + +def compute_render_fingerprint_and_dedup( + *, + video_path: str, + generation_task_id: str, + project_id: str, + user_id: str, + batch_id: str, + mode: str, + session: Session, +) -> dict: + """渲染+上传完成后的预计算:计算指纹+历史/批次查重,返回可序列化 dict。 + + **不创建 GeneratedVideo 记录**。结果由调用方写入 extra_meta["rendered_output"], + finalize 时复用。mode 参数保留签名一致性(查重结果中不直接使用)。 + """ + _ = mode # 保留在签名里便于调用方对齐;查重结果不含 mode + return _compute_from_local( + video_path=video_path, + generation_task_id=generation_task_id, + project_id=project_id, + user_id=user_id, + batch_id=batch_id, + session=session, + ) + + def create_video_record_and_dedup( *, generation_task_id: str, @@ -25,8 +173,8 @@ def create_video_record_and_dedup( batch_id: str, file_url: str, file_size: int, - duration: float, - video_path: str, + duration: float | None = None, + video_path: str | None, mode: str, session: Session, width: int = 1280, @@ -34,30 +182,55 @@ def create_video_record_and_dedup( fps: float = 25.0, name: str = "", thumbnail_url: str = "", + pre_fingerprint_dict: dict | None = None, + pre_fingerprint_chunks: list[dict] | None = None, + pre_dedup_result: dict | None = None, ) -> dict: - """Returns: {"video_count": int, "is_duplicate": bool, "batch_similarity": float|None, - "duplicate_of": str|None} —— batch_similarity 为批次内最高相似度(无批次查重时 None)。""" - """创建 GeneratedVideo 记录,计算指纹并执行查重(历史 + 批次)。 + """创建 GeneratedVideo 记录 + 可选查重。 - 采用两阶段持久化:先计算所有指纹/查重数据(内存), - 再一次性写入数据库并 commit。若指纹计算失败, - 视频记录仍会创建(无查重数据),但保证不会出现"写了记录却没 commit"的中间态。 + 两种用法: + - 传入 ``video_path``(非 None):从本地视频计算指纹+查重,直接创建记录(旧路径/测试)。 + - 仅传入 ``pre_*``:复用 worker 预计算结果,不访问本地视频(finalize 用)。 Returns: - 创建的视频记录数量(1 表示成功,0 表示失败) + {"video_id", "video_count", "is_duplicate", "batch_similarity", "duplicate_of"} """ - from video_processing.dedup import VideoDeduplicator, _save_fingerprint_chunks - from packages.adapters.sqlalchemy_impl.generated_video_repository import ( SQLAlchemyGeneratedVideoRepository, ) - from packages.domain import GeneratedVideo + from packages.adapters.sqlalchemy_impl.models import VideoFingerprintChunkModel + from packages.domain.generated_video import GeneratedVideo try: video_id = uuid4().hex video_name = name.strip() if name else f"generated-{generation_task_id[:8]}.mp4" - # ── Phase 1: 构建视频记录(内存,不 commit) ──────────────── + # 决定查重/元信息来源 + if video_path: + pre = _compute_from_local( + video_path=video_path, + generation_task_id=generation_task_id, + project_id=project_id, + user_id=user_id, + batch_id=batch_id, + session=session, + ) + else: + pre = dict(pre_dedup_result or {}) + pre.setdefault("fingerprint_dict", pre_fingerprint_dict) + pre.setdefault("fingerprint_chunks", pre_fingerprint_chunks) + pre.setdefault("is_duplicate", False) + pre.setdefault("duplicate_of", None) + pre.setdefault("duplicate_rate", None) + pre.setdefault("match_count", None) + pre.setdefault("visual_similarity", None) + pre.setdefault("batch_similarity", None) + + used_duration = float(duration if duration is not None else pre.get("duration", 0.0)) + used_width = int(pre.get("width", width) or width) + used_height = int(pre.get("height", height) or height) + used_fps = float(pre.get("fps", fps) or fps) + generated_video = GeneratedVideo( id=video_id, project_id=project_id, @@ -66,117 +239,66 @@ def create_video_record_and_dedup( name=video_name, file_url=file_url, file_size=file_size, - duration=duration, - width=width, - height=height, - fps=fps, + duration=used_duration, + width=used_width, + height=used_height, + fps=used_fps, status="completed", generation_params={"mode": mode}, thumbnail_url=thumbnail_url or None, + video_fingerprint=pre.get("fingerprint_dict"), + is_duplicate=bool(pre.get("is_duplicate", False)), + duplicate_of=pre.get("duplicate_of"), + duplicate_rate=pre.get("duplicate_rate"), + match_count=pre.get("match_count"), + visual_similarity=pre.get("visual_similarity"), ) - # ── Phase 2: 计算指纹 & 查重(全部在内存) ──────────────── - deduplicator = VideoDeduplicator() - fingerprint = None - batch_similarity: float | None = None - - try: - fingerprint = deduplicator.compute_fingerprint(video_path) - except Exception as fp_err: - logger.warning("Fingerprint computation failed for %s: %s", video_id, fp_err) - - if fingerprint is not None: - generated_video.video_fingerprint = fingerprint.to_dict() - - # 写入分片指纹表(失败不阻塞) + # 写分片指纹表 + chunks = pre.get("fingerprint_chunks") + if chunks: try: - _save_fingerprint_chunks(fingerprint, video_id, project_id, user_id, session) + chunk_models = [ + VideoFingerprintChunkModel( + id=uuid4().hex, + video_id=video_id, + project_id=project_id, + user_id=user_id, + start_time_ms=int(c.get("start_time_ms", 0)), + end_time_ms=int(c.get("end_time_ms", 0)), + phash_binary=str(c.get("phash_binary", "")), + color_histogram=[float(v) for v in (c.get("color_histogram") or [])], + frame_count=int(c.get("frame_count", 0)), + ) + for c in chunks + if isinstance(c, dict) + ] + if chunk_models: + # 幂等:先清理旧分片 + session.query(VideoFingerprintChunkModel).filter( + VideoFingerprintChunkModel.video_id == video_id + ).delete(synchronize_session=False) + session.bulk_save_objects(chunk_models) except Exception as chunk_err: logger.warning("Failed to save fingerprint chunks for %s: %s", video_id, chunk_err) - # (a) 历史成片查重(跨项目全局 + 时长预过滤) - # Issue #1702: fingerprint.duration 单位是秒,旧代码 /1000 让时长预过滤失效 - duration_sec = fingerprint.duration if fingerprint.duration else 0 - duplicate_result = deduplicator.check_duplicate( - fingerprint, - project_id, - session, - scope="user", - user_id=user_id, - duration_sec=duration_sec, - exclude_video_id=video_id, - ) - - # (b) 批次内查重(仅当有 batch_id 时) - batch_similarity: float | None = None - if not duplicate_result and batch_id: - duplicate_result = deduplicator.check_batch_duplicate(fingerprint, batch_id, video_id, session) - if duplicate_result: - batch_similarity = float(duplicate_result.get("similarity", 0.0)) - if duplicate_result: - generated_video.is_duplicate = True - generated_video.duplicate_of = duplicate_result["duplicate_of"] - logger.info( - "Duplicate detected: %s -> %s (reason=%s, similarity=%.3f)", - video_id, - duplicate_result["duplicate_of"], - duplicate_result["reason"], - duplicate_result["similarity"], - ) - else: - generated_video.is_duplicate = False - generated_video.duplicate_of = None - - # 计算重复率百分比(跨项目全局) - try: - rate_result = deduplicator.compute_duplicate_rate( - fingerprint, - project_id, - video_id, - session, - scope="user", - user_id=user_id, - ) - generated_video.duplicate_rate = rate_result["duplicate_rate"] - generated_video.match_count = rate_result["match_count"] - generated_video.visual_similarity = rate_result["visual_similarity"] - logger.info( - "Duplicate rate for %s: %.2f%% (visual_sim=%.3f, matches=%d)", - video_id, - rate_result["duplicate_rate"], - rate_result["visual_similarity"], - rate_result["match_count"], - ) - except Exception as rate_err: - logger.warning("Failed to compute duplicate_rate for %s: %s", video_id, rate_err) - generated_video.duplicate_rate = None - - # ── Phase 3: 一次性持久化 ───────────────────────────────── - video_repo = SQLAlchemyGeneratedVideoRepository(session) - video_repo.create(generated_video) - - if thumbnail_url: - logger.info("Thumbnail set for video %s: %s", video_id, thumbnail_url[:80]) - + repo = SQLAlchemyGeneratedVideoRepository(session) + repo.create(generated_video) session.commit() - logger.info( - "GeneratedVideo record created: %s (task=%s, dup=%s, rate=%s)", - video_id, - generation_task_id, - generated_video.is_duplicate, - generated_video.duplicate_rate, - ) return { + "video_id": video_id, "video_count": 1, - "is_duplicate": bool(generated_video.is_duplicate), - "batch_similarity": batch_similarity, - "duplicate_of": generated_video.duplicate_of, + "is_duplicate": bool(pre.get("is_duplicate", False)), + "batch_similarity": pre.get("batch_similarity"), + "duplicate_of": pre.get("duplicate_of"), } except Exception as e: - logger.error( - "Failed to create video record / dedup for task %s: %s", - generation_task_id, - e, - ) + logger.error("Failed to create video record for task %s: %s", generation_task_id, e) session.rollback() - return {"video_count": 0, "is_duplicate": False, "batch_similarity": None, "duplicate_of": None} + return { + "video_id": "", + "video_count": 0, + "is_duplicate": False, + "batch_similarity": None, + "duplicate_of": None, + } diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index cf6ec8ccf..1ba072f71 100644 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -17,7 +17,6 @@ import logging import time from pathlib import Path -from video_processing.ffmpeg_utils import probe_duration from worker_app.celery_app import celery_app from worker_app.db import SessionLocal from worker_app.tasks.generation_plan_builder import build_error_info as _build_error_info @@ -142,7 +141,7 @@ def _flush_logs(task_id: str, gen_task) -> None: # ── 共享工具模块导入 ────────────────────────────────────────────────────────── -from video_processing.dedup_helpers import create_video_record_and_dedup +from video_processing.dedup_helpers import compute_render_fingerprint_and_dedup from video_processing.oss_helpers import ( download_asset, get_signed_download_url, @@ -555,7 +554,7 @@ def _reselect_plan_for_batch_retry(task_id: str, plan_id: str, task_info: dict) return None -def _record_video_and_dedup( +def _precompute_render_metadata( *, task_id: str, project_id: str, @@ -568,28 +567,48 @@ def _record_video_and_dedup( video_name: str = "", thumbnail_url: str = "", ) -> dict: - """成片落库 + 指纹查重(含批次内)。返回查重信息 dict。""" - duration = probe_duration(Path(video_path)) - dedup_session = SessionLocal() + """渲染+上传完成后的预处理:计算指纹/查重(不落 GeneratedVideo 库)。 + + #2024: 视频生成后不再自动入成品库。本函数计算视频元信息、指纹、历史+批次查重, + 结果以 dict 返回,由调用方写入 GenerationTask.extra_meta["rendered_output"], + 等用户 Step5 调 finalize 时复用,避免 finalize 时从 OSS 下载视频重算。 + """ + pre_session = SessionLocal() try: - result = create_video_record_and_dedup( + fp_result = compute_render_fingerprint_and_dedup( + video_path=video_path, generation_task_id=task_id, project_id=project_id, user_id=user_id, batch_id=batch_id, - file_url=file_url, - file_size=file_size, - duration=duration, - video_path=video_path, mode=editing_mode.value, - session=dedup_session, - name=video_name, - thumbnail_url=thumbnail_url, + session=pre_session, ) finally: - dedup_session.close() - result["duration"] = duration - return result + pre_session.close() + return { + "file_url": file_url, + "file_size": file_size, + "duration": fp_result.get("duration", 0.0), + "width": fp_result.get("width", 1280), + "height": fp_result.get("height", 720), + "fps": fp_result.get("fps", 25.0), + "name": video_name, + "thumbnail_url": thumbnail_url, + "mode": editing_mode.value, + "batch_id": batch_id, + "project_id": project_id, + "user_id": user_id, + # 查重结果(finalize 时直接写入 GeneratedVideo 字段,无需重算) + "fingerprint_dict": fp_result.get("fingerprint_dict"), + "fingerprint_chunks": fp_result.get("fingerprint_chunks"), + "is_duplicate": bool(fp_result.get("is_duplicate", False)), + "duplicate_of": fp_result.get("duplicate_of"), + "duplicate_rate": fp_result.get("duplicate_rate"), + "match_count": fp_result.get("match_count"), + "visual_similarity": fp_result.get("visual_similarity"), + "video_fingerprint_md5": fp_result.get("video_fingerprint_md5", ""), + } # ── Celery Task ────────────────────────────────────────────────────────────── @@ -945,8 +964,8 @@ def generate_video(self, task_id: str) -> dict: ) file_size = output_path.stat().st_size - # ── 4.5 落库 + 查重(批次任务检查批次内相似度) ─────────── - dedup_info = _record_video_and_dedup( + # ── 4.5 预计算指纹/元信息(#2024: 不自动入成品库,finalize 时再落库+查重) ── + rendered_output = _precompute_render_metadata( task_id=task_id, project_id=project_id, batch_id=batch_id, @@ -958,68 +977,23 @@ def generate_video(self, task_id: str) -> dict: video_name=task_info.get("video_title", ""), thumbnail_url=thumbnail_url, ) - duration = dedup_info.get("duration", render_duration) - video_count = dedup_info.get("video_count", 1) - batch_sim = dedup_info.get("batch_similarity") + duration = rendered_output.get("duration", render_duration) + # #2024: 批次内重渲依赖已 finalize 的同批次视频。渲染阶段暂不做批次查重决策, + # 统一在 finalize 阶段查重;首版即视为最终渲染结果。 + file_size_final = file_size if gen_task: gen_task.append_log( "OSS上传", - f"第{render_attempt + 1}版上传成功, 大小={file_size}" - + (f", 批次相似度={batch_sim:.0%}" if batch_sim is not None else ""), + f"第{render_attempt + 1}版上传成功, 大小={file_size},等待用户确认封面", file_size=file_size, file_url=file_url, ) _flush_logs(task_id, gen_task) - # 非批次 / 相似度达标 / 已是最后一次 → 结束循环 - if not should_rerender_for_batch_dedup( - batch_id=batch_id, - render_attempt=render_attempt, - batch_similarity=batch_sim, - ): - file_size_final = file_size - break - - # 批次内相似度过高:重选独立 plan 后重渲一次 - logger.warning( - "[task_id=%s] 批次内查重相似度 %.2f 超阈值 %.2f,重选 plan 重渲", - task_id, - batch_sim, - BATCH_RENDER_SIMILARITY_LIMIT, - ) - if gen_task: - gen_task.append_log("批次查重", f"与批次内成片相似度过高({batch_sim:.0%}),重新选片渲染") - _flush_logs(task_id, gen_task) - new_plan_id = _reselect_plan_for_batch_retry(task_id, current_plan_id, task_info) - if not new_plan_id: - logger.warning("[task_id=%s] 重选 plan 失败,保留首版", task_id) - file_size_final = file_size - break - # 回写任务关联的 plan(重渲版以新 plan 渲染) - try: - _ps = SessionLocal() - try: - from packages.adapters.sqlalchemy_impl.generation_task_repository import ( - SQLAlchemyGenerationTaskRepository, - ) - - _pr = SQLAlchemyGenerationTaskRepository(_ps) - _gt = _pr.get(task_id) - if _gt: - _gt.source_edit_plan_id = new_plan_id - _pr.update(_gt) - finally: - _ps.close() - except Exception: - logger.warning("[task_id=%s] 回写重渲 plan_id 失败", task_id, exc_info=True) - current_plan_id = new_plan_id - # 清理本轮临时目录,下一轮重新渲染 - if render_temp_dir: - import shutil - - shutil.rmtree(render_temp_dir, ignore_errors=True) - render_temp_dir = None + # #2024: 不再因批次内相似度过高而重渲(finalize 阶段统一查重), + # 首版即视为最终渲染结果,直接结束循环。 + break file_size = file_size_final or file_size _update_task_progress(task_id, 95, "上传完成") @@ -1062,8 +1036,47 @@ def generate_video(self, task_id: str) -> dict: except Exception: logger.warning("[task_id=%s] 封面帧持久化失败", task_id, exc_info=True) - # ── 5. 标记完成 ────────────────────────────────────────────────── - _update_task_status(task_id, "mark_completed", result_count=video_count) + # ── 5. 保存渲染产物到 extra_meta 并标记为等待封面确认(#2024: 不自动入成品库) ── + try: + _finalize_meta_session = SessionLocal() + try: + from packages.adapters.sqlalchemy_impl.models import ( + GenerationTaskModel, + ) + + _meta_model = ( + _finalize_meta_session.query(GenerationTaskModel) + .filter(GenerationTaskModel.id == task_id) + .first() + ) + if _meta_model: + meta = dict(_meta_model.extra_meta or {}) + meta["rendered_output"] = { + "file_url": file_url, + "file_size": file_size, + "duration": duration, + "width": rendered_output.get("width", 1280), + "height": rendered_output.get("height", 720), + "fps": rendered_output.get("fps", 25.0), + "name": rendered_output.get("name", ""), + "thumbnail_url": rendered_output.get("thumbnail_url", ""), + "mode": rendered_output.get("mode", editing_mode.value), + "fingerprint_dict": rendered_output.get("fingerprint_dict"), + "batch_id": batch_id, + "project_id": project_id, + "user_id": user_id, + } + _meta_model.extra_meta = meta + _finalize_meta_session.commit() + finally: + _finalize_meta_session.close() + except Exception as meta_err: + logger.warning( + "[task_id=%s] 保存 rendered_output 到 extra_meta 失败: %s", task_id, meta_err, exc_info=True + ) + + # #2024: 标记为「等待用户确认封面」,不自动入成品库;等用户调 finalize 接口才真正 mark_completed + _update_task_status(task_id, "mark_awaiting_cover") # 5.1 更新标题使用次数 try: diff --git a/packages/application/generated_video_finalize.py b/packages/application/generated_video_finalize.py new file mode 100644 index 000000000..82cbd12eb --- /dev/null +++ b/packages/application/generated_video_finalize.py @@ -0,0 +1,191 @@ +"""#2024: 视频生成 finalize 入库用例。 + +Worker 渲染+上传完成后不自动入库,只把渲染产物与查重结果保存到 +GenerationTask.extra_meta["rendered_output"],并标记为 awaiting_cover。 +用户点「完成」时由 API 调用本用例:创建 GeneratedVideo 记录(复用预计算查重结果)、 +推进任务到 completed,返回新记录 id。 + +设计原则:finalize 必须快速(仅 DB 写入,不下载视频、不重算指纹)—— +所有耗时操作(指纹计算、历史/批次查重)都在 worker 渲染阶段预完成。 +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from datetime import UTC, datetime +from typing import Any +from uuid import uuid4 + +from sqlalchemy.orm import Session + +logger = logging.getLogger(__name__) + + +@dataclass(slots=True) +class RenderedOutput: + """Worker 预计算并写入 extra_meta 的渲染产物+查重结果。""" + + file_url: str + file_size: int = 0 + duration: float = 0.0 + width: int = 1280 + height: int = 720 + fps: float = 25.0 + name: str = "" + thumbnail_url: str = "" + mode: str = "narrative" + batch_id: str = "" + project_id: str = "" + user_id: str = "" + # 查重结果(worker 预计算) + fingerprint_dict: dict[str, Any] | None = None + fingerprint_chunks: list[dict[str, Any]] | None = None + is_duplicate: bool = False + duplicate_of: str | None = None + duplicate_rate: float | None = None + match_count: int | None = None + visual_similarity: float | None = None + video_fingerprint_md5: str = "" + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "RenderedOutput": + if not isinstance(data, dict): + raise ValueError("rendered_output must be a dict") + return cls( + file_url=str(data.get("file_url") or ""), + file_size=int(data.get("file_size") or 0), + duration=float(data.get("duration") or 0.0), + width=int(data.get("width") or 1280), + height=int(data.get("height") or 720), + fps=float(data.get("fps") or 25.0), + name=str(data.get("name") or ""), + thumbnail_url=str(data.get("thumbnail_url") or ""), + mode=str(data.get("mode") or "narrative"), + batch_id=str(data.get("batch_id") or ""), + project_id=str(data.get("project_id") or ""), + user_id=str(data.get("user_id") or ""), + fingerprint_dict=data.get("fingerprint_dict"), + fingerprint_chunks=data.get("fingerprint_chunks"), + is_duplicate=bool(data.get("is_duplicate", False)), + duplicate_of=data.get("duplicate_of"), + duplicate_rate=_safe_float(data.get("duplicate_rate")), + match_count=_safe_int(data.get("match_count")), + visual_similarity=_safe_float(data.get("visual_similarity")), + video_fingerprint_md5=str(data.get("video_fingerprint_md5") or ""), + ) + + +def _safe_float(v) -> float | None: + if v is None: + return None + try: + return float(v) + except (TypeError, ValueError): + return None + + +def _safe_int(v) -> int | None: + if v is None: + return None + try: + return int(v) + except (TypeError, ValueError): + return None + + +def finalize_generated_video( + *, + task, + session: Session, + effective_cover_url: str = "", +) -> dict: + """将 awaiting_cover 的任务正式入库。 + + 从 ``task.extra_meta["rendered_output"]`` 读取 worker 预存的渲染结果与查重数据, + 创建 GeneratedVideo 记录并 commit;调用方负责将 task 推进到 completed 并 update。 + + Returns: + {"video_id": str, "is_duplicate": bool, "duplicate_of": str|None} + """ + from packages.adapters.sqlalchemy_impl.generated_video_repository import ( + SQLAlchemyGeneratedVideoRepository, + ) + from packages.adapters.sqlalchemy_impl.models import VideoFingerprintChunkModel + from packages.domain.generated_video import GeneratedVideo + + meta = dict(task.extra_meta or {}) + rendered_dict = meta.get("rendered_output") or {} + rendered = RenderedOutput.from_dict(rendered_dict) + + if not rendered.file_url.strip(): + raise ValueError(f"task {task.id} rendered_output.file_url 为空,无法 finalize") + + video_id = uuid4().hex + video_name = rendered.name.strip() or f"generated-{task.id[:8]}.mp4" + + generated_video = GeneratedVideo( + id=video_id, + project_id=(rendered.project_id or task.project_id or "").strip(), + user_id=(rendered.user_id or task.created_by_user_id or "").strip(), + generation_task_id=task.id, + name=video_name, + file_url=rendered.file_url.strip(), + file_size=rendered.file_size, + duration=rendered.duration, + width=rendered.width, + height=rendered.height, + fps=rendered.fps, + status="completed", + generation_params={"mode": rendered.mode}, + thumbnail_url=effective_cover_url or rendered.thumbnail_url or None, + video_fingerprint=rendered.fingerprint_dict, + is_duplicate=rendered.is_duplicate, + duplicate_of=rendered.duplicate_of, + duplicate_rate=rendered.duplicate_rate, + match_count=rendered.match_count, + visual_similarity=rendered.visual_similarity, + created_at=datetime.now(UTC), + generated_at=datetime.now(UTC), + ) + + # 写入分片指纹(worker 预序列化的 chunk 列表) + if rendered.fingerprint_chunks: + try: + chunk_models = [] + for c in rendered.fingerprint_chunks: + if not isinstance(c, dict): + continue + chunk_models.append( + VideoFingerprintChunkModel( + id=uuid4().hex, + video_id=video_id, + project_id=generated_video.project_id, + user_id=generated_video.user_id, + start_time_ms=int(c.get("start_time_ms", 0)), + end_time_ms=int(c.get("end_time_ms", 0)), + phash_binary=str(c.get("phash_binary", "")), + color_histogram=[float(v) for v in (c.get("color_histogram") or [])], + frame_count=int(c.get("frame_count", 0)), + ) + ) + if chunk_models: + session.bulk_save_objects(chunk_models) + except Exception as chunk_err: + logger.warning("Failed to persist fingerprint chunks for video %s: %s", video_id, chunk_err) + + video_repo = SQLAlchemyGeneratedVideoRepository(session) + video_repo.create(generated_video) + session.commit() + logger.info( + "[finalize] GeneratedVideo created: %s (task=%s, dup=%s, cover=%s)", + video_id, + task.id, + rendered.is_duplicate, + bool(effective_cover_url), + ) + return { + "video_id": video_id, + "is_duplicate": rendered.is_duplicate, + "duplicate_of": rendered.duplicate_of, + } diff --git a/packages/domain/generation_task.py b/packages/domain/generation_task.py index 30897bb4e..f84ae3a46 100755 --- a/packages/domain/generation_task.py +++ b/packages/domain/generation_task.py @@ -1,9 +1,13 @@ """GenerationTask 领域模型 — 视频生成任务. 状态机: - pending → running → completed + pending → running → awaiting_cover → completed ↘ failed → pending (重试) ↘ cancelled + +``awaiting_cover`` 表示渲染已完成、视频文件已上传、封面候选已就绪, +但用户尚未在 Step5 确认封面并点击「完成」,此时不创建 GeneratedVideo 成品记录。 +用户调用 finalize 接口后才进入 ``completed`` 并正式入库。 """ from __future__ import annotations @@ -34,8 +38,11 @@ class GenerationTaskStatus(StrEnum): RUNNING = "running" """运行中(正在生成视频)""" + AWAITING_COVER = "awaiting_cover" + """视频已渲染上传、封面候选已就绪,等待用户在 Step5 确认封面(finalize 前的中间态)""" + COMPLETED = "completed" - """已完成(视频生成成功)""" + """已完成(用户已确认封面,视频已正式入库)""" FAILED = "failed" """失败(生成失败)""" @@ -61,6 +68,8 @@ class GenerationTaskStatus(StrEnum): return cls.FAILED if normalized in ("process", "processing", "run", "running", "in_progress"): return cls.RUNNING + if normalized in ("awaiting_cover", "waiting_cover", "video_ready", "rendered", "pending_cover"): + return cls.AWAITING_COVER if normalized in ("cancel", "cancelled", "canceled"): return cls.CANCELLED return cls.PENDING @@ -79,6 +88,12 @@ _VALID_TRANSITIONS: dict[GenerationTaskStatus, set[GenerationTaskStatus]] = { GenerationTaskStatus.CANCELLED, }, GenerationTaskStatus.RUNNING: { + GenerationTaskStatus.AWAITING_COVER, + GenerationTaskStatus.COMPLETED, # 兜底/测试兼容:允许直接完成;主路径走 awaiting_cover + GenerationTaskStatus.FAILED, + GenerationTaskStatus.CANCELLED, + }, + GenerationTaskStatus.AWAITING_COVER: { GenerationTaskStatus.COMPLETED, GenerationTaskStatus.FAILED, GenerationTaskStatus.CANCELLED, @@ -210,6 +225,11 @@ class GenerationTask: """是否运行中。""" return self.status == GenerationTaskStatus.RUNNING + @property + def is_awaiting_cover(self) -> bool: + """是否等待用户确认封面(渲染已完成、视频已上传、尚未 finalize 入库)。""" + return self.status == GenerationTaskStatus.AWAITING_COVER + # ── 状态转换 ──────────────────────────────────────────────────────────── def transition_to(self, new_status: GenerationTaskStatus | str) -> None: @@ -248,13 +268,27 @@ class GenerationTask: self.started_at = datetime.now(UTC) self.error_message = "" + def mark_awaiting_cover(self) -> None: + """标记为等待确认封面(running → awaiting_cover)。 + + 渲染与上传已完成、封面候选已就绪,等待用户在 Step5 选封面并点「完成」。 + 此时不创建 GeneratedVideo 成品记录;progress 置 100,completed_at 暂不设置 + (finalize 完成入库时才真正结束任务)。 + + Raises: + ValueError: 当前状态不允许转换到 awaiting_cover + """ + self.transition_to(GenerationTaskStatus.AWAITING_COVER) + self.progress = 100.0 + self.error_message = "" + def mark_completed(self, result_count: int = 1) -> None: - """标记为已完成(running → completed)。 + """标记为已完成(awaiting_cover → completed,由 finalize 调用)。 设置 completed_at、progress=100.0、result_count,清除 error_message。 Args: - result_count: 生成的视频数量,默认为 1 + result_count: 入库的视频数量,默认为 1 Raises: ValueError: 当前状态不允许转换到 completed diff --git a/tests/unit/domain/test_generation_task.py b/tests/unit/domain/test_generation_task.py index cb2c9a831..b9bc82169 100644 --- a/tests/unit/domain/test_generation_task.py +++ b/tests/unit/domain/test_generation_task.py @@ -13,8 +13,9 @@ class TestGenerationTaskStatus: """GenerationTaskStatus 枚举测试.""" def test_five_statuses(self): - """五种状态.""" - assert len(GenerationTaskStatus) == 5 + """六种状态(#2024 新增 awaiting_cover).""" + assert len(GenerationTaskStatus) == 6 + assert GenerationTaskStatus.AWAITING_COVER == "awaiting_cover" def test_pending(self): assert GenerationTaskStatus.PENDING == "pending" diff --git a/tests/unit/test_finalize_generation.py b/tests/unit/test_finalize_generation.py new file mode 100644 index 000000000..087579468 --- /dev/null +++ b/tests/unit/test_finalize_generation.py @@ -0,0 +1,206 @@ +"""#2024: 视频生成 finalize 流程单测。 + +覆盖: +1. GenerationTask 新状态 awaiting_cover 与 mark_awaiting_cover 方法 +2. finalize 用例:幂等 / 状态校验 / 正常入库 +3. Worker 侧预计算函数 signature 兼容 +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +# 使 worker 目录可导入 +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT)) +sys.path.insert(0, str(ROOT / "apps" / "worker")) + +from packages.domain.generation_task import ( + TERMINAL_STATUSES, + GenerationTask, + GenerationTaskStatus, +) + +# ── 1. 状态机 ───────────────────────────────────────────────────────── + + +class TestAwaitingCoverStatus: + def test_enum_value(self): + assert GenerationTaskStatus.AWAITING_COVER == "awaiting_cover" + + def test_not_terminal(self): + assert GenerationTaskStatus.AWAITING_COVER not in TERMINAL_STATUSES + + def test_is_awaiting_cover_property(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1", asset_ids=["a1"]) + task.mark_processing() + task.mark_awaiting_cover() + assert task.is_awaiting_cover + assert not task.is_completed + assert not task.is_failed + assert task.progress == 100.0 + # awaiting_cover 不设置 completed_at + assert task.completed_at is None + + def test_normal_flow_pending_running_awaiting_completed(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1", asset_ids=["a1"]) + task.mark_processing() + task.mark_awaiting_cover() + assert task.status == GenerationTaskStatus.AWAITING_COVER + task.mark_completed(result_count=1) + assert task.is_completed + assert task.completed_at is not None + assert task.result_count == 1 + + def test_awaiting_to_failed_allowed(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1", asset_ids=["a1"]) + task.mark_processing() + task.mark_awaiting_cover() + task.mark_failed("test error") + assert task.is_failed + + def test_awaiting_to_cancelled_allowed(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1", asset_ids=["a1"]) + task.mark_processing() + task.mark_awaiting_cover() + task.mark_cancelled() + assert task.status == GenerationTaskStatus.CANCELLED + + def test_cannot_jump_pending_to_awaiting(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1", asset_ids=["a1"]) + with pytest.raises(ValueError): + task.mark_awaiting_cover() + + def test_mark_completed_resets_error(self): + task = GenerationTask.create(project_id="p1", asset_library_id="lib1", asset_ids=["a1"]) + task.mark_processing() + task.mark_awaiting_cover() + task.mark_completed() + assert task.error_message == "" + + +class TestFinalizeUseCase: + """finalize_generated_video 用例测试(通过 mock session 避免 DB)。""" + + def _make_task(self, extra_meta=None): + task = GenerationTask.create(project_id="proj1", asset_library_id="lib1", asset_ids=["a1"]) + task.id = "task-123" + task.mark_processing() + task.mark_awaiting_cover() + task.project_id = "proj1" + task.created_by_user_id = "user1" + task.extra_meta = extra_meta or { + "rendered_output": { + "file_url": "oss://bucket/v.mp4", + "file_size": 1024, + "duration": 12.5, + "width": 1080, + "height": 1920, + "fps": 30.0, + "name": "demo.mp4", + "mode": "narrative", + "batch_id": "", + "is_duplicate": False, + "fingerprint_dict": {"md5": "abc"}, + } + } + return task + + def test_missing_rendered_output_raises(self): + """rendered_output.file_url 为空应抛 ValueError。""" + from packages.application.generated_video_finalize import finalize_generated_video + + task = self._make_task(extra_meta={"rendered_output": {"file_url": ""}}) + session = MagicMock() + with pytest.raises(ValueError): + finalize_generated_video( + task=task, + session=session, + effective_cover_url="", + ) + + def test_success_creates_generated_video(self): + """正常 finalize 创建一条 GeneratedVideo,返回 video_id。""" + from packages.application.generated_video_finalize import finalize_generated_video + + task = self._make_task() + session = MagicMock() + # mock video repo + with patch( + "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository" + ) as mock_repo_cls: + mock_repo = MagicMock() + mock_repo_cls.return_value = mock_repo + result = finalize_generated_video( + task=task, + session=session, + effective_cover_url="https://cdn/cover.jpg", + ) + assert result["video_id"], "video_id should be non-empty" + assert mock_repo.create.called, "video_repo.create must be called" + created_video = mock_repo.create.call_args[0][0] + assert created_video.generation_task_id == "task-123" + assert created_video.thumbnail_url == "https://cdn/cover.jpg" + assert created_video.width == 1080 + assert created_video.height == 1920 + assert created_video.duration == 12.5 + session.commit.assert_called() + + def test_cover_fallback_to_task_cover_url(self): + """finalize 未传 cover_url 时使用 rendered_output.thumbnail_url。""" + from packages.application.generated_video_finalize import finalize_generated_video + + task = self._make_task() + session = MagicMock() + with patch( + "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository" + ) as mock_repo_cls: + mock_repo = MagicMock() + mock_repo_cls.return_value = mock_repo + result = finalize_generated_video( + task=task, + session=session, + effective_cover_url="", + ) + assert result["video_id"] + created_video = mock_repo.create.call_args[0][0] + # rendered_output.thumbnail_url 为空时 thumbnail 为 None + assert created_video.thumbnail_url is None + + +class TestRenderedOutputDataclass: + def test_from_dict_defaults(self): + from packages.application.generated_video_finalize import RenderedOutput + + ro = RenderedOutput.from_dict({"file_url": "https://x/y.mp4"}) + assert ro.file_url == "https://x/y.mp4" + assert ro.width == 1280 + assert ro.height == 720 + assert ro.fps == 25.0 + assert ro.is_duplicate is False + + def test_from_dict_full(self): + from packages.application.generated_video_finalize import RenderedOutput + + ro = RenderedOutput.from_dict( + { + "file_url": "https://x/y.mp4", + "width": 1080, + "height": 1920, + "is_duplicate": True, + "duplicate_of": "old-id", + "duplicate_rate": 42.5, + } + ) + assert ro.width == 1080 + assert ro.is_duplicate is True + assert ro.duplicate_of == "old-id" + assert ro.duplicate_rate == 42.5 + + +if __name__ == "__main__": + sys.exit(pytest.main([__file__, "-v"])) diff --git a/tests/unit/test_generation_finalize_service.py b/tests/unit/test_generation_finalize_service.py new file mode 100644 index 000000000..a80b971c5 --- /dev/null +++ b/tests/unit/test_generation_finalize_service.py @@ -0,0 +1,429 @@ +"""#2024 GenerationFinalizeService 单元测试。 + +覆盖 service 层:存在性校验、幂等分支、状态门、封面决策、异常映射、成功路径。 +同时为 packages/application/generated_video_finalize.py 的缺失分支补测。 +""" + +from unittest.mock import MagicMock, patch + +import pytest + +# ---------- helpers ---------- + + +def _make_task( + task_id="task-1", + status="awaiting_cover", + project_id="proj-1", + user_id="user-1", + cover_url="", + extra_meta=None, + error_message="", +): + t = MagicMock() + t.id = task_id + t.project_id = project_id + t.created_by_user_id = user_id + t.cover_url = cover_url + t.extra_meta = extra_meta if extra_meta is not None else {} + t.error_message = error_message + s = MagicMock() + s.value = status + t.status = s + + def _mark_completed(result_count=1): + s.value = "completed" + t.completed_at = "now" + + t.mark_completed = MagicMock(side_effect=_mark_completed) + + def _mark_confirmed(): + t.is_preview = False + + t.mark_confirmed = MagicMock(side_effect=_mark_confirmed) + return t + + +def _make_db(): + db = MagicMock() + db.query.return_value.filter.return_value.first.return_value = None + db.commit = MagicMock() + db.rollback = MagicMock() + db.bulk_save_objects = MagicMock() + return db + + +def _make_rendered_dict(**overrides): + base = { + "file_url": "https://oss.example.com/v.mp4", + "file_size": 123456, + "duration": 10.5, + "width": 1280, + "height": 720, + "fps": 25.0, + "name": "demo.mp4", + "thumbnail_url": "https://oss.example.com/thumb.jpg", + "mode": "narrative", + "batch_id": "", + "project_id": "proj-1", + "user_id": "user-1", + "fingerprint_dict": {"phash": "abc"}, + "fingerprint_chunks": [ + { + "start_time_ms": 0, + "end_time_ms": 1000, + "phash_binary": "0101", + "color_histogram": [0.1, 0.2, 0.3], + "frame_count": 25, + }, + ], + "is_duplicate": False, + "duplicate_of": None, + "duplicate_rate": 0.0, + "match_count": 0, + "visual_similarity": 0.0, + "video_fingerprint_md5": "md5-abc", + } + base.update(overrides) + return base + + +_PATCHES = [ + "packages.adapters.sqlalchemy_impl.generation_task_repository.SQLAlchemyGenerationTaskRepository", + "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository", + "packages.adapters.sqlalchemy_impl.models.GeneratedVideoModel", + "packages.application.generated_video_finalize.finalize_generated_video", +] + + +def _svc(db): + from app.services.generation_finalize_service import GenerationFinalizeService + + return GenerationFinalizeService(db) + + +# ---------- service tests ---------- + + +class TestFinalizeService: + def test_task_not_found_raises_404(self): + from app.services.generation_finalize_service import GenerationFinalizeError + + db = _make_db() + with patch(_PATCHES[0]) as TR, patch(_PATCHES[1]), patch(_PATCHES[2]), patch(_PATCHES[3]): + TR.return_value.get.return_value = None + svc = _svc(db) + with pytest.raises(GenerationFinalizeError) as ei: + svc.finalize_task("nope", "user-1") + assert ei.value.status_code == 404 + assert ei.value.code == "TaskNotFound" + + def test_invalid_status_raises(self): + from app.services.generation_finalize_service import GenerationFinalizeError + + db = _make_db() + task = _make_task(status="running") + with patch(_PATCHES[0]) as TR, patch(_PATCHES[1]), patch(_PATCHES[2]) as GVM, patch(_PATCHES[3]): + TR.return_value.get.return_value = task + GVM.query.filter.return_value.first.return_value = None + svc = _svc(db) + with pytest.raises(GenerationFinalizeError) as ei: + svc.finalize_task("task-1", "user-1") + assert ei.value.code == "InvalidTaskStatus" + assert ei.value.status_code == 400 + + def test_idempotent_when_video_already_exists_updates_cover_and_completes(self): + db = _make_db() + task = _make_task(status="awaiting_cover") + existing = MagicMock() + existing.id = "video-exist" + existing.thumbnail_url = "https://old-cover.jpg" + db.query.return_value.filter.return_value.first.return_value = existing + existing_video = MagicMock() + existing_video.id = "video-exist" + + with patch(_PATCHES[0]) as TR, patch(_PATCHES[1]) as VR, patch(_PATCHES[2]), patch(_PATCHES[3]): + TR.return_value.get.return_value = task + TR.return_value.update = MagicMock() + VR.return_value.get.return_value = existing_video + svc = _svc(db) + result = svc.finalize_task("task-1", "user-1", cover_url="https://new-cover.jpg") + assert result.id == "video-exist" + assert existing.thumbnail_url == "https://new-cover.jpg" + assert task.cover_url == "https://new-cover.jpg" + task.mark_completed.assert_called() + TR.return_value.update.assert_called_with(task) + db.commit.assert_called() + + def test_idempotent_already_completed_skips_mark_completed(self): + db = _make_db() + task = _make_task(status="completed") + existing = MagicMock() + existing.id = "v-exist" + existing.thumbnail_url = "https://c.jpg" + db.query.return_value.filter.return_value.first.return_value = existing + existing_video = MagicMock() + + with patch(_PATCHES[0]) as TR, patch(_PATCHES[1]) as VR, patch(_PATCHES[2]), patch(_PATCHES[3]): + TR.return_value.get.return_value = task + VR.return_value.get.return_value = existing_video + svc = _svc(db) + svc.finalize_task("task-1", "user-1") + task.mark_completed.assert_not_called() + + def test_missing_rendered_output_raises(self): + from app.services.generation_finalize_service import GenerationFinalizeError + + db = _make_db() + task = _make_task(status="awaiting_cover", extra_meta={}) + with patch(_PATCHES[0]) as TR, patch(_PATCHES[1]), patch(_PATCHES[2]), patch(_PATCHES[3]) as fu: + TR.return_value.get.return_value = task + TR.return_value.update = MagicMock() + fu.side_effect = ValueError("file_url 为空") + svc = _svc(db) + with pytest.raises(GenerationFinalizeError) as ei: + svc.finalize_task("task-1", "user-1") + assert ei.value.code == "RenderedOutputMissing" + + def test_success_creates_video_and_marks_completed(self): + db = _make_db() + task = _make_task( + status="awaiting_cover", + cover_url="https://task-cover.jpg", + extra_meta={"rendered_output": _make_rendered_dict()}, + ) + created_video = MagicMock() + created_video.id = "video-new" + + with patch(_PATCHES[0]) as TR, patch(_PATCHES[1]) as VR, patch(_PATCHES[2]), patch(_PATCHES[3]) as fu: + TR.return_value.get.return_value = task + TR.return_value.update = MagicMock() + fu.return_value = {"video_id": "video-new", "is_duplicate": False, "duplicate_of": None} + VR.return_value.get.return_value = created_video + svc = _svc(db) + v = svc.finalize_task("task-1", "user-1") + assert v.id == "video-new" + task.mark_completed.assert_called_once_with(result_count=1) + assert "rendered_output" not in task.extra_meta + TR.return_value.update.assert_called_with(task) + db.commit.assert_called() + + def test_cover_fallback_to_task_cover_url(self): + db = _make_db() + task = _make_task( + status="awaiting_cover", + cover_url="https://task-cover.jpg", + extra_meta={"rendered_output": _make_rendered_dict(thumbnail_url="")}, + ) + with patch(_PATCHES[0]) as TR, patch(_PATCHES[1]) as VR, patch(_PATCHES[2]), patch(_PATCHES[3]) as fu: + TR.return_value.get.return_value = task + TR.return_value.update = MagicMock() + fu.return_value = {"video_id": "v1", "is_duplicate": False, "duplicate_of": None} + VR.return_value.get.return_value = MagicMock(id="v1") + svc = _svc(db) + svc.finalize_task("task-1", "user-1") + assert task.cover_url == "https://task-cover.jpg" + kwargs = fu.call_args.kwargs + assert kwargs["effective_cover_url"] == "https://task-cover.jpg" + + def test_explicit_cover_url_overrides_task_cover(self): + db = _make_db() + task = _make_task( + status="awaiting_cover", + cover_url="https://old.jpg", + extra_meta={"rendered_output": _make_rendered_dict()}, + ) + with patch(_PATCHES[0]) as TR, patch(_PATCHES[1]) as VR, patch(_PATCHES[2]), patch(_PATCHES[3]) as fu: + TR.return_value.get.return_value = task + TR.return_value.update = MagicMock() + fu.return_value = {"video_id": "v1", "is_duplicate": False, "duplicate_of": None} + VR.return_value.get.return_value = MagicMock(id="v1") + svc = _svc(db) + svc.finalize_task("task-1", "user-1", cover_url=" https://new.jpg ") + kwargs = fu.call_args.kwargs + assert kwargs["effective_cover_url"] == "https://new.jpg" + + +# ---------- packages/application/generated_video_finalize.py 覆盖补测 ---------- + + +class TestFinalizeUseCaseCoverage: + def test_rendered_output_non_dict_raises(self): + from packages.application.generated_video_finalize import RenderedOutput + + with pytest.raises(ValueError): + RenderedOutput.from_dict("not-a-dict") + + def test_safe_float_handles_invalid(self): + from packages.application.generated_video_finalize import _safe_float, _safe_int + + assert _safe_float(None) is None + assert _safe_float("abc") is None + assert _safe_float("3.14") == pytest.approx(3.14) + assert _safe_int(None) is None + assert _safe_int("xyz") is None + assert _safe_int("42") == 42 + + def test_fingerprint_chunks_non_dict_entry_is_skipped(self): + """非 dict chunk 被 continue 跳过;bulk_save 只处理合法 chunk。""" + from packages.application import generated_video_finalize as mod + + task = _make_task( + extra_meta={ + "rendered_output": _make_rendered_dict( + fingerprint_chunks=[ + "not-a-dict", + { + "start_time_ms": 0, + "end_time_ms": 500, + "phash_binary": "xx", + "color_histogram": [0.1, 0.2], + "frame_count": 10, + }, + ], + ) + } + ) + db = MagicMock() + db.bulk_save_objects = MagicMock() + db.commit = MagicMock() + # 模块内的 SQLAlchemyGeneratedVideoRepository/GeneratedVideo/VideoFingerprintChunkModel + # 都是在函数内部 import 的,直接 patch 到被 patch 模块的属性上 + fake_repo = MagicMock() + fake_repo.create = MagicMock() + with ( + patch( + "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository", + return_value=fake_repo, + ), + patch( + "packages.adapters.sqlalchemy_impl.models.VideoFingerprintChunkModel", + side_effect=lambda **kw: MagicMock(**kw), + ), + patch("packages.domain.generated_video.GeneratedVideo", side_effect=lambda **kw: MagicMock(**kw)), + ): + result = mod.finalize_generated_video( + task=task, + session=db, + effective_cover_url="https://cover.jpg", + ) + assert "video_id" in result + assert db.bulk_save_objects.call_count == 1 + saved_chunks = db.bulk_save_objects.call_args[0][0] + assert len(saved_chunks) == 1 + db.commit.assert_called() + fake_repo.create.assert_called_once() + + def test_missing_file_url_raises(self): + from packages.application import generated_video_finalize as mod + + task = _make_task( + extra_meta={ + "rendered_output": _make_rendered_dict(file_url=""), + } + ) + db = MagicMock() + with pytest.raises(ValueError): + mod.finalize_generated_video(task=task, session=db, effective_cover_url="") + + def test_no_fingerprint_chunks_skips_bulk_save(self): + from packages.application import generated_video_finalize as mod + + task = _make_task( + extra_meta={ + "rendered_output": _make_rendered_dict(fingerprint_chunks=None), + } + ) + db = MagicMock() + db.bulk_save_objects = MagicMock() + db.commit = MagicMock() + fake_repo = MagicMock() + with ( + patch( + "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository", + return_value=fake_repo, + ), + patch("packages.domain.generated_video.GeneratedVideo", side_effect=lambda **kw: MagicMock(**kw)), + ): + mod.finalize_generated_video(task=task, session=db, effective_cover_url="") + db.bulk_save_objects.assert_not_called() + fake_repo.create.assert_called_once() + db.commit.assert_called() + + def test_name_fallback_when_empty(self): + from packages.application import generated_video_finalize as mod + + task = _make_task( + task_id="abcd1234ef567890", + extra_meta={ + "rendered_output": _make_rendered_dict(name=" ", thumbnail_url=""), + }, + ) + db = MagicMock() + db.bulk_save_objects = MagicMock() + db.commit = MagicMock() + fake_repo = MagicMock() + captured = {} + + def _capture(**kw): + captured.update(kw) + return MagicMock(**kw) + + with ( + patch( + "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository", + return_value=fake_repo, + ), + patch("packages.domain.generated_video.GeneratedVideo", side_effect=_capture), + patch( + "packages.adapters.sqlalchemy_impl.models.VideoFingerprintChunkModel", + side_effect=lambda **kw: MagicMock(**kw), + ), + ): + mod.finalize_generated_video(task=task, session=db, effective_cover_url="") + assert captured["name"].startswith("generated-abcd1234") + assert captured["thumbnail_url"] is None + + def test_chunk_exception_is_swallowed(self): + """chunk 构造异常时 logger.warning,不阻塞主流程。""" + from packages.application import generated_video_finalize as mod + + task = _make_task( + extra_meta={ + "rendered_output": _make_rendered_dict( + fingerprint_chunks=[ + { + "start_time_ms": 0, + "end_time_ms": 500, + "phash_binary": "xx", + "color_histogram": ["not-a-number"], + "frame_count": 10, + }, + ], + ) + } + ) + db = MagicMock() + db.bulk_save_objects = MagicMock() + db.commit = MagicMock() + fake_repo = MagicMock() + with ( + patch( + "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository", + return_value=fake_repo, + ), + patch("packages.domain.generated_video.GeneratedVideo", side_effect=lambda **kw: MagicMock(**kw)), + patch( + "packages.adapters.sqlalchemy_impl.models.VideoFingerprintChunkModel", + side_effect=lambda **kw: MagicMock(**kw), + ), + ): + # color_histogram 里 "not-a-number" 触发 float() 异常,被 except chunk_err 吞掉 + # 但此时 chunk_models 中仍有 1 个元素(MagicMock 构造不会因 float() 失败)—— + # 因为我们把 float 列表推导也放在 try 内,float("not-a-number") 抛 ValueError + # 所以要让 float 真的抛。但 MagicMock side_effect 不触发 float(),这里直接构造: + # 通过真实验证路径 + result = mod.finalize_generated_video(task=task, session=db, effective_cover_url="") + assert "video_id" in result + db.commit.assert_called() + fake_repo.create.assert_called_once()