From decbc107fb765dd3d5fc90924a07e30c1b9648d1 Mon Sep 17 00:00:00 2001 From: saas-backend-agent Date: Tue, 29 Sep 2026 04:13:43 +0800 Subject: [PATCH] =?UTF-8?q?fix(worker+api):=20P1=20=E5=B0=81=E9=9D=A2?= =?UTF-8?q?=E8=AF=84=E5=88=86=E6=97=B6=E5=BA=8Fbug=20+=20direct/complete?= =?UTF-8?q?=E5=90=9Eingest=E5=8D=A0=E4=BD=8Dbug=20(#2092)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bug1 (P1) 封面评分在 TemporaryDirectory 退出后执行，帧文件已删除： - score_frames 移入 with TemporaryDirectory 块内，帧文件存在时评分 - 评分在上传之前完成，按评分顺序并发上传，best 帧 is_best=True 标为 frame_0 - MediaKit 路径帧在 NamedTemporaryFile 持久存在，改在 MediaKit 循环内就地评分 - 评分失败打 warning 保持原顺序，不静默、不阻塞上传 - 本地帧 by TemporaryDirectory 自动清理，不再入 _temp_paths 重复 unlink - 补 3 个单测：帧文件存在性、scorer 异常兜底、best 排序 Bug2 (P1) /direct/complete 命中 prepare 占位 asset 直接返 duplicated 吞 ingest： - 新增 IngestJobRepository.find_by_asset_id（port + SQLAlchemy + InMemory 实现） - 新增 _is_true_duplicate 区分真重复 vs 占位： - READY → 真重复（返 duplicated） - PROCESSING/UPLOADING 且有在跑/已完成 job → 幂等重试返 duplicated + 已有 job_id - PROCESSING/UPLOADING 且无 job → 占位/孤儿，继续补提 ingest - ERROR/DELETED → 允许重新 ingest - complete_direct_upload / upload_asset 两处早返逻辑替换为 _is_true_duplicate - _submit_ingest_job 前加幂等守卫：find_by_asset_id 命中已有 job 直接复用，避免竞态重复提交 - 补 5 个单测覆盖：占位补 ingest、READY 真重复、PROCESSING+已有job幂等、ERROR 重入、multipart 占位补 ingest --- apps/api/app/api/routes/upload.py | 140 ++++++++++--- .../video_processing/thumbnail_generator.py | 91 +++++++-- .../in_memory/ingest_job_repository.py | 18 ++ .../sqlalchemy_impl/ingest_job_repository.py | 37 ++++ packages/ports/ingest_job_repository.py | 3 + tests/unit/test_thumbnail_generator.py | 127 ++++++++++++ .../test_upload_complete_idempotency_1714.py | 188 ++++++++++++++++++ 7 files changed, 555 insertions(+), 49 deletions(-) diff --git a/apps/api/app/api/routes/upload.py b/apps/api/app/api/routes/upload.py index dcf4f9927..4bfdef025 100644 --- a/apps/api/app/api/routes/upload.py +++ b/apps/api/app/api/routes/upload.py @@ -290,6 +290,43 @@ def _submit_ingest_job( return job +def _find_active_ingest_job(ingest_job_repository: Any, asset_id: str) -> Any | None: + """查询 asset 上是否存在"仍在跑或已成功"的 ingest job（FAILED 视为不存在，需重提）。""" + if not asset_id: + return None + find = getattr(ingest_job_repository, "find_by_asset_id", None) + if not callable(find): + # 旧仓储未实现 find_by_asset_id，无法判断 → 保守返回 None（走正常流程， + # _submit_ingest_job 自身有数据库唯一约束/幂等兜底，不会重复建 job） + return None + try: + return find(asset_id) + except Exception: # noqa: BLE001 + logger.warning("[upload] find_by_asset_id 查询失败，按无 job 处理: asset=%s", asset_id, exc_info=True) + return None + + +def _is_true_duplicate(existing_asset: Asset, ingest_job_repository: Any) -> tuple[bool, Any | None]: + """判断 `existing_asset` 是真重复（应短路返 duplicated）还是占位（应补提 ingest）。 + + 返回 (is_duplicate, existing_job): + - READY 素材：真重复，job 可能为 None（已就绪不需要 job_id） + - PROCESSING/UPLOADING 且已有在跑/已完成 ingest job：幂等重试，真重复，job 返回给前端轮询 + - PROCESSING/UPLOADING 且无 job：prepare 建的占位 / 之前 ingest 创建失败 → 非重复，需补提 ingest + - ERROR/DELETED：非重复（允许重新上传覆盖） + """ + status = getattr(existing_asset, "status", None) + if status == AssetStatus.READY: + return True, None + if status in (AssetStatus.PROCESSING, AssetStatus.UPLOADING): + job = _find_active_ingest_job(ingest_job_repository, existing_asset.id) + if job is not None: + return True, job + return False, None + # ERROR / DELETED / 其它：走正常流程重新 ingest + return False, None + + @router.post("/direct/prepare", response_model=DirectUploadPrepareResponse) async def prepare_direct_upload( request: DirectUploadPrepareRequest, @@ -444,12 +481,25 @@ async def complete_direct_upload( file_size=request.file_size, ) if existing is not None: - return DirectUploadCompleteResponse( - storage_key=existing.storage_key, - ingest_job_id="", - duplicated=True, - asset_id=existing.id, - url=storage_service.get_url(existing.storage_key), + is_dup, existing_job = _is_true_duplicate(existing, ingest_job_repository) + if is_dup: + logger.info( + "[upload] complete 幂等命中真重复: asset=%s status=%s job=%s", + existing.id, + getattr(existing, "status", None), + getattr(existing_job, "id", None), + ) + return DirectUploadCompleteResponse( + storage_key=existing.storage_key, + ingest_job_id=getattr(existing_job, "id", "") or "", + duplicated=True, + asset_id=existing.id, + url=storage_service.get_url(existing.storage_key), + ) + logger.info( + "[upload] complete 命中占位 asset（status=%s 无 ingest job），继续补提 ingest: asset=%s", + getattr(existing, "status", None), + existing.id, ) try: @@ -479,14 +529,24 @@ async def complete_direct_upload( ) # Issue #1776: 计数由 asset_repository.create() 自动维护 - job = _submit_ingest_job( - project_id=request.project_id, - library_id=request.library_id, - storage_key=normalized_key, - ingest_job_repository=ingest_job_repository, - file_hash=request.file_hash, - asset_id=pending_asset.id, - ) + # 幂等保护：补提占位场景下可能已有 job（极端竞态），先查一次 + existing_job = _find_active_ingest_job(ingest_job_repository, pending_asset.id) + if existing_job is not None: + logger.info( + "[upload] complete 补提时发现 job 已存在（竞态/并发重试），复用: asset=%s job=%s", + pending_asset.id, + existing_job.id, + ) + job = existing_job + else: + job = _submit_ingest_job( + project_id=request.project_id, + library_id=request.library_id, + storage_key=normalized_key, + ingest_job_repository=ingest_job_repository, + file_hash=request.file_hash, + asset_id=pending_asset.id, + ) return DirectUploadCompleteResponse( storage_key=normalized_key, ingest_job_id=job.id, @@ -533,12 +593,27 @@ async def upload_asset( file_size=0, ) if existing is not None: - return UploadAssetResponse( - storage_key=existing.storage_key, - ingest_job_id="", - url="", - duplicated=True, - asset_id=existing.id, + is_dup, existing_job = _is_true_duplicate(existing, ingest_job_repository) + if is_dup: + logger.info( + "[upload] multipart 幂等命中真重复: asset=%s status=%s job=%s", + existing.id, + getattr(existing, "status", None), + getattr(existing_job, "id", None), + ) + return UploadAssetResponse( + storage_key=existing.storage_key, + ingest_job_id=getattr(existing_job, "id", "") or "", + url=storage_service.get_url(existing.storage_key) + if getattr(existing, "status", None) == AssetStatus.READY + else "", + duplicated=True, + asset_id=existing.id, + ) + logger.info( + "[upload] multipart 命中占位 asset（status=%s 无 ingest job），继续补提 ingest: asset=%s", + getattr(existing, "status", None), + existing.id, ) file_id = uuid4().hex[:8] @@ -574,14 +649,23 @@ async def upload_asset( ) # Issue #1776: 计数由 asset_repository.create() 自动维护 - job = _submit_ingest_job( - project_id=project_id, - library_id=library_id, - storage_key=storage_key, - ingest_job_repository=ingest_job_repository, - file_hash=file_hash, - asset_id=pending_asset.id, - ) + existing_job = _find_active_ingest_job(ingest_job_repository, pending_asset.id) + if existing_job is not None: + logger.info( + "[upload] multipart 补提时发现 job 已存在（竞态/并发重试），复用: asset=%s job=%s", + pending_asset.id, + existing_job.id, + ) + job = existing_job + else: + job = _submit_ingest_job( + project_id=project_id, + library_id=library_id, + storage_key=storage_key, + ingest_job_repository=ingest_job_repository, + file_hash=file_hash, + asset_id=pending_asset.id, + ) return UploadAssetResponse( storage_key=storage_key, diff --git a/apps/worker/video_processing/thumbnail_generator.py b/apps/worker/video_processing/thumbnail_generator.py index 375aa9f9c..0d295fc86 100755 --- a/apps/worker/video_processing/thumbnail_generator.py +++ b/apps/worker/video_processing/thumbnail_generator.py @@ -606,6 +606,20 @@ def extract_and_upload_cover_frames( if len(candidates) >= num_frames: logger.info("[thumbnail] MediaKit 抽帧完成: %d 帧", len(candidates)) + # MediaKit 路径帧在 NamedTemporaryFile 中持久存在（finally 清理），在进入本地 ffmpeg 前评分 + if len(candidates) > 1: + try: + from packages.shared.cover_frame_scorer import score_frames + + candidates = score_frames(candidates) + logger.info( + "[thumbnail] MediaKit 封面帧评分完成: count=%d best_score=%.1f", + len(candidates), + candidates[0].get("score", 0.0) if candidates else 0.0, + ) + except Exception: + logger.warning("[thumbnail] MediaKit 封面帧质量评分失败，保持原始顺序", exc_info=True) + # ── 默认路径：本地 ffmpeg 单次 select 抽帧 + 并发上传 ────────────── if len(candidates) < num_frames: if candidates: @@ -651,54 +665,89 @@ def extract_and_upload_cover_frames( except Exception as e: logger.warning("[thumbnail] 标题叠加失败 %s: %s", fp, e) - # 5) 并发上传 OSS（线程池并发） + # 5) 质量评分（必须在 TemporaryDirectory 内，帧文件还在磁盘上） + t_score = time.monotonic() + local_candidates: list[dict] = [{"position": st, "image_path": fp} for (st, fp) in frame_results] + scored: list[dict] = local_candidates + if len(local_candidates) > 1: + try: + from packages.shared.cover_frame_scorer import score_frames + + scored = score_frames(local_candidates) + logger.info( + "[thumbnail] 封面评分耗时: %.2fs (best_score=%.1f, count=%d)", + time.monotonic() - t_score, + scored[0].get("score", 0.0) if scored else 0.0, + len(scored), + ) + except Exception: + logger.warning( + "[thumbnail] 封面帧质量评分失败，保持 seek 点原始顺序", + exc_info=True, + ) + scored = local_candidates + + # 6) 按评分顺序并发上传 OSS（best 帧先上传；best 已是 scored[0]） t2 = time.monotonic() - def _upload_one(idx: int, st: float, fp: str) -> dict | None: + def _upload_one(rank: int, st: float, fp: str, score: float) -> dict | None: try: - storage_key = f"covers/{plan_id}/{task_id}/frame_{idx}.jpg" + storage_key = f"covers/{plan_id}/{task_id}/frame_{rank}.jpg" url = upload_to_oss(fp, storage_key) if url: - return {"url": url, "position": st, "image_path": fp} - logger.warning("[thumbnail] 上传失败 idx=%d", idx) + return { + "url": url, + "position": st, + "image_path": fp, + "score": score, + "is_best": rank == 0, + } + logger.warning("[thumbnail] 上传失败 rank=%d t=%.2f", rank, st) except Exception as e: - logger.warning("[thumbnail] 上传异常 idx=%d t=%.2f: %s", idx, st, e) + logger.warning("[thumbnail] 上传异常 rank=%d t=%.2f: %s", rank, st, e) return None - upload_results: list[dict | None] = [None] * len(frame_results) - max_workers = min(8, max(2, len(frame_results))) + upload_results: list[dict | None] = [None] * len(scored) + max_workers = min(8, max(2, len(scored))) with ThreadPoolExecutor(max_workers=max_workers) as pool: - future_map = {pool.submit(_upload_one, i, st, fp): i for i, (st, fp) in enumerate(frame_results)} + future_map = { + pool.submit( + _upload_one, + i, + float(c.get("position", 0.0)), + str(c["image_path"]), + float(c.get("score", 0.0)), + ): i + for i, c in enumerate(scored) + } for fut in as_completed(future_map): i = future_map[fut] try: upload_results[i] = fut.result() except Exception as e: - logger.warning("[thumbnail] 上传 feature 异常 idx=%d: %s", i, e) + logger.warning("[thumbnail] 上传 future 异常 rank=%d: %s", i, e) logger.info("[thumbnail] 并发上传耗时: %.2fs", time.monotonic() - t2) for r in upload_results: if r is not None: - _temp_paths.append(r["image_path"]) + # 本地帧在 TemporaryDirectory 内，with 退出自动删除，无需进 _temp_paths candidates.append(r) - # ── 阶段 2：质量评分 ──────────────────────────────────────────── + # 如果本地 ffmpeg 路径产生了候选（已评分）但未经过 MediaKit 路径，candidates 已按评分顺序排好。 + # 混合场景下（MediaKit + 本地 ffmpeg 都产出），统一按 score 降序排列；缺失 score 的（理论上不应出现）排末尾。 if len(candidates) > 1: - try: - from packages.shared.cover_frame_scorer import score_frames - - candidates = score_frames(candidates) + candidates.sort(key=lambda c: c.get("score", -1.0), reverse=True) + if candidates: + candidates[0]["is_best"] = True elapsed = time.monotonic() - t0 logger.info( - "[thumbnail] 封面帧评分完成: plan_id=%s count=%d best_score=%.1f elapsed=%.2fs path=%s", + "[thumbnail] 封面完成: plan_id=%s count=%d best=t%.2fs score=%.1f elapsed=%.2fs", plan_id, len(candidates), - candidates[0].get("score", 0.0) if candidates else 0.0, + candidates[0].get("position", 0.0), + candidates[0].get("score", 0.0), elapsed, - "mediakit" if use_mediakit else "local", ) - except Exception: - logger.warning("[thumbnail] 封面帧质量评分失败，保持原始顺序", exc_info=True) for c in candidates: c.pop("image_path", None) diff --git a/packages/adapters/in_memory/ingest_job_repository.py b/packages/adapters/in_memory/ingest_job_repository.py index 13c530bdb..41a089797 100644 --- a/packages/adapters/in_memory/ingest_job_repository.py +++ b/packages/adapters/in_memory/ingest_job_repository.py @@ -14,6 +14,24 @@ class InMemoryIngestJobRepository: def get(self, job_id: str) -> IngestJob | None: return self._items.get(job_id) + def find_by_asset_id(self, asset_id: str) -> IngestJob | None: + if not asset_id: + return None + from packages.domain.classification import IngestJobStatus + + running: IngestJob | None = None + completed: IngestJob | None = None + for job in self._items.values(): + if getattr(job, "asset_id", "") != asset_id: + continue + if job.status in (IngestJobStatus.PENDING, IngestJobStatus.PROCESSING): + if running is None or job.created_at > running.created_at: + running = job + elif job.status == IngestJobStatus.COMPLETED: + if completed is None or job.created_at > completed.created_at: + completed = job + return running or completed + def update(self, job: IngestJob) -> IngestJob: self._items[job.id] = job return job diff --git a/packages/adapters/sqlalchemy_impl/ingest_job_repository.py b/packages/adapters/sqlalchemy_impl/ingest_job_repository.py index 7c42fdeee..280915242 100644 --- a/packages/adapters/sqlalchemy_impl/ingest_job_repository.py +++ b/packages/adapters/sqlalchemy_impl/ingest_job_repository.py @@ -46,6 +46,43 @@ class SQLAlchemyIngestJobRepository: updated_at=model.updated_at, ) + def find_by_asset_id(self, asset_id: str) -> IngestJob | None: + """返回 asset 最近一条未失败的 ingest job（PENDING/PROCESSING/COMPLETED 均算存在，用于幂等判断）。""" + if not asset_id: + return None + # 优先返回仍在跑的 (PENDING/PROCESSING)，否则返回最新一条 COMPLETED + model = ( + self.session.query(IngestJobModel) + .filter(IngestJobModel.asset_id == asset_id) + .filter(IngestJobModel.status.in_([IngestJobStatus.PENDING.value, IngestJobStatus.PROCESSING.value])) + .order_by(IngestJobModel.created_at.desc()) + .first() + ) + if model is None: + model = ( + self.session.query(IngestJobModel) + .filter(IngestJobModel.asset_id == asset_id) + .filter(IngestJobModel.status == IngestJobStatus.COMPLETED.value) + .order_by(IngestJobModel.created_at.desc()) + .first() + ) + if model is None: + return None + return IngestJob( + id=model.id, + project_id=model.project_id, + library_id=model.library_id, + storage_key=model.storage_key, + status=IngestJobStatus(model.status), + error_message=model.error_message, + result_asset_id=model.result_asset_id, + file_hash=model.file_hash or "", + asset_id=getattr(model, "asset_id", "") or "", + celery_task_id=getattr(model, "celery_task_id", "") or "", + created_at=model.created_at, + updated_at=model.updated_at, + ) + def list_by_project(self, project_id: str) -> list[IngestJob]: models = self.session.query(IngestJobModel).filter(IngestJobModel.project_id == project_id).all() return [self.get(model.id) for model in models if self.get(model.id) is not None] diff --git a/packages/ports/ingest_job_repository.py b/packages/ports/ingest_job_repository.py index d9184fe79..3badc0b5f 100644 --- a/packages/ports/ingest_job_repository.py +++ b/packages/ports/ingest_job_repository.py @@ -12,5 +12,8 @@ class IngestJobRepository(Protocol): def get(self, job_id: str) -> IngestJob | None: """Retrieve an ingest job by ID.""" + def find_by_asset_id(self, asset_id: str) -> IngestJob | None: + """Retrieve the most recent ingest job for a given asset (幂等判断).""" + def update(self, job: IngestJob) -> IngestJob: """Update an ingest job and return it.""" diff --git a/tests/unit/test_thumbnail_generator.py b/tests/unit/test_thumbnail_generator.py index e27863634..01fa7813b 100755 --- a/tests/unit/test_thumbnail_generator.py +++ b/tests/unit/test_thumbnail_generator.py @@ -282,3 +282,130 @@ class TestSeekPointBlackAvoidance: assert out[2] == 8.0 # 中点必须不在黑屏内 assert not (4.5 <= out[1] <= 5.5) + + +class TestScorerRunsInsideTempDir: + """P1 修复：scorer 必须在 TemporaryDirectory 块内调用（帧文件还在时）。""" + + def _setup_mocks(self, monkeypatch, tmp_path, *, scorer_should_read=True): + from video_processing.thumbnail_generator import extract_and_upload_cover_frames + + class FakeClient: + is_available = False + + monkeypatch.setattr( + "packages.shared.mediakit_client.get_mediakit_client", + lambda: FakeClient(), + ) + + self._frames_on_disk_when_called = [] + + def _fake_single_pass(video_path, seek_points, out_dir, prefix="frame", **kw): + results = [] + for i, st in enumerate(seek_points): + fp = Path(out_dir) / f"{prefix}_{i + 1:02d}.jpg" + fp.write_bytes(b"\xff\xd8\xff\xe0" + b"X" * 200) + results.append((st, str(fp))) + return results + + monkeypatch.setattr( + "video_processing.thumbnail_generator._extract_frames_single_pass", + _fake_single_pass, + ) + monkeypatch.setattr( + "video_processing.thumbnail_generator._detect_black_intervals", + lambda *a, **kw: [], + ) + monkeypatch.setattr( + "video_processing.ffmpeg_utils.probe_duration", + lambda path: 60.0, + ) + monkeypatch.setattr( + "video_processing.oss_helpers.upload_to_oss", + lambda path, key: f"https://oss.example.com/{key}", + ) + # 标题叠加 no-op + monkeypatch.setattr( + "video_processing.thumbnail_generator.apply_title_overlay", + lambda *a, **kw: None, + ) + + # 记录 scorer 被调用时各 image_path 是否存在 + def _fake_scorer(candidates): + for c in candidates: + self._frames_on_disk_when_called.append(Path(c["image_path"]).exists()) + # 给个假评分：倒序排，验证顺序被应用 + scored = list(candidates) + for i, c in enumerate(scored): + c["score"] = float(len(scored) - i) + scored.sort(key=lambda c: c["score"], reverse=True) + return scored + + monkeypatch.setattr( + "packages.shared.cover_frame_scorer.score_frames", + _fake_scorer, + ) + return extract_and_upload_cover_frames + + def test_scorer_reads_files_while_they_exist(self, tmp_path, monkeypatch): + """核心 P1：评分时帧文件必须还在磁盘上（在 TemporaryDirectory 内调用）。""" + extract = self._setup_mocks(monkeypatch, tmp_path) + video_file = tmp_path / "t.mp4" + video_file.write_bytes(b"fake") + result = extract(str(video_file), "plan1", num_frames=3) + # scorer 看到的 3 个文件都必须存在 + assert len(self._frames_on_disk_when_called) == 3 + assert all(self._frames_on_disk_when_called), f"scorer 调用时有文件已被删除: {self._frames_on_disk_when_called}" + # 结果按评分降序排列（is_best 在第一个） + assert len(result) == 3 + assert result[0].get("is_best") is True + # 结果中不应该再暴露 image_path + assert all("image_path" not in c for c in result) + + def test_scorer_failure_falls_back_gracefully(self, tmp_path, monkeypatch): + """评分抛异常时不应中断上传，仍返回所有候选帧。""" + from video_processing.thumbnail_generator import extract_and_upload_cover_frames + + class FakeClient: + is_available = False + + monkeypatch.setattr("packages.shared.mediakit_client.get_mediakit_client", lambda: FakeClient()) + + def _fake_single_pass(video_path, seek_points, out_dir, prefix="frame", **kw): + results = [] + for i, st in enumerate(seek_points): + fp = Path(out_dir) / f"{prefix}_{i + 1:02d}.jpg" + fp.write_bytes(b"\xff\xd8\xff\xe0" + b"X" * 100) + results.append((st, str(fp))) + return results + + monkeypatch.setattr("video_processing.thumbnail_generator._extract_frames_single_pass", _fake_single_pass) + monkeypatch.setattr("video_processing.thumbnail_generator._detect_black_intervals", lambda *a, **kw: []) + monkeypatch.setattr("video_processing.ffmpeg_utils.probe_duration", lambda p: 60.0) + monkeypatch.setattr( + "video_processing.oss_helpers.upload_to_oss", + lambda path, key: f"https://oss/{key}", + ) + monkeypatch.setattr("video_processing.thumbnail_generator.apply_title_overlay", lambda *a, **kw: None) + + def _boom(candidates): + raise RuntimeError("cv2 crashed") + + monkeypatch.setattr("packages.shared.cover_frame_scorer.score_frames", _boom) + + video_file = tmp_path / "t.mp4" + video_file.write_bytes(b"fake") + # 不应抛出 + result = extract_and_upload_cover_frames(str(video_file), "plan1", num_frames=3) + assert len(result) == 3 + assert all("url" in c for c in result) + + def test_best_frame_is_first_after_scoring(self, tmp_path, monkeypatch): + """评分后 best 帧（score 最高）在 candidates[0]，is_best=True。""" + extract = self._setup_mocks(monkeypatch, tmp_path) + video_file = tmp_path / "t.mp4" + video_file.write_bytes(b"fake") + result = extract(str(video_file), "plan1", num_frames=5) + assert result[0]["is_best"] is True + scores = [c.get("score", 0.0) for c in result] + assert scores == sorted(scores, reverse=True) diff --git a/tests/unit/test_upload_complete_idempotency_1714.py b/tests/unit/test_upload_complete_idempotency_1714.py index 544a36887..783c3a3bc 100644 --- a/tests/unit/test_upload_complete_idempotency_1714.py +++ b/tests/unit/test_upload_complete_idempotency_1714.py @@ -128,6 +128,25 @@ class StubIngestJobRepository: def get(self, job_id: str) -> IngestJob | None: return self._jobs.get(job_id) + def find_by_asset_id(self, asset_id: str) -> IngestJob | None: + """返回 asset 最近一条在跑/已完成 job（FAILED 不返回，允许重提）。""" + from packages.domain.classification import IngestJobStatus as S + + if not asset_id: + return None + running = None + completed = None + for job in self._jobs.values(): + if getattr(job, "asset_id", "") != asset_id: + continue + if job.status in (S.PENDING, S.PROCESSING): + if running is None or job.created_at > running.created_at: + running = job + elif job.status == S.COMPLETED: + if completed is None or job.created_at > completed.created_at: + completed = job + return running or completed + def update(self, job: IngestJob) -> IngestJob: self._jobs[job.id] = job return job @@ -420,3 +439,172 @@ class TestMultipartUploadIdempotency: assert ingest_repo.created_count == 1 # OSS 上传只发生一次（第二次在幂等检查处直接返回） assert storage.upload_file.call_count == 1 + + +# ── P1 修复（#2092）：占位 asset 被误判为 duplicate → 素材永久 processing ── + + +class TestPlaceholderAssetNotTreatedAsDuplicate: + """prepare 建了 PROCESSING 占位但还没 ingest，complete 必须补提 ingest 而不是短路返 duplicated。""" + + def test_complete_hits_prepare_placeholder_without_job_submits_ingest(self): + """场景：/direct/prepare 建了 PROCESSING 占位（同 client_upload_id），complete 命中后应补提 ingest。""" + from packages.domain.classification import IngestJobStatus + + # 预建占位 asset（prepare 建的，PROCESSING，无 ingest job） + placeholder = Asset( + id="asset-placeholder", + project_id="proj-1", + library_id="lib-1", + name="IMG_9999.MOV", + storage_key="uploads/prepare/IMG_9999.MOV", + mime_type="video/quicktime", + status=AssetStatus.PROCESSING, + client_upload_id="tok-prep", + ) + # 注意：占位的 storage_key 是 prepare 生成的 key，complete 传入的是用户实际上传的 key + client, asset_repo, ingest_repo, storage = _client( + asset_repo=StubAssetRepository([placeholder]), + ) + storage._normalize_storage_key = lambda k: k # complete 用自己的 key + storage.get_url = lambda k: f"https://oss/{k}" + + body = { + "project_id": "proj-1", + "library_id": "lib-1", + "storage_key": "uploads/actual/IMG_9999.MOV", # complete 用真实上传 key + "client_upload_id": "tok-prep", # 命中占位 + "file_size": 1024, + } + r = client.post("/api/v1/direct/complete", json=body) + assert r.status_code == 200, r.text + b = r.json() + # 关键：不是 duplicate；返回 job_id；占位被复用（不新建 asset） + assert b["duplicated"] is False, f"占位被误判为 duplicate: {b}" + assert b["ingest_job_id"], "应补提 ingest job" + assert b["asset_id"] == "asset-placeholder" + # 不新建 asset（占位复用） + assert len(asset_repo.created) == 0 + # job 被提交 + assert ingest_repo.created_count == 1 + + def test_ready_asset_treated_as_true_duplicate(self): + """READY 素材命中 → 真重复，返 duplicated 且不提交新 job。""" + ready = Asset( + id="asset-ready", + project_id="proj-1", + library_id="lib-1", + name="done.MOV", + storage_key="uploads/done/done.MOV", + mime_type="video/quicktime", + status=AssetStatus.READY, + client_upload_id="tok-done", + ) + client, asset_repo, ingest_repo, _storage = _client(asset_repo=StubAssetRepository([ready])) + body = { + "project_id": "proj-1", + "library_id": "lib-1", + "storage_key": "uploads/done/done.MOV", + "client_upload_id": "tok-done", + } + r = client.post("/api/v1/direct/complete", json=body) + assert r.status_code == 200 + b = r.json() + assert b["duplicated"] is True + assert b["asset_id"] == "asset-ready" + assert ingest_repo.created_count == 0 + + def test_processing_asset_with_existing_job_is_idempotent_duplicate(self): + """PROCESSING 但已有在跑 job → 幂等重试，返 duplicated + 已有 job_id，不重复提交。""" + processing = Asset( + id="asset-running", + project_id="proj-1", + library_id="lib-1", + name="running.MOV", + storage_key="uploads/run/running.MOV", + mime_type="video/quicktime", + status=AssetStatus.PROCESSING, + client_upload_id="tok-run", + ) + client, asset_repo, ingest_repo, _ = _client(asset_repo=StubAssetRepository([processing])) + # 预置一个在跑 job + existing_job = IngestJob( + id="job-existing", + project_id="proj-1", + library_id="lib-1", + storage_key="uploads/run/running.MOV", + asset_id="asset-running", + ) + ingest_repo.create(existing_job) + before = ingest_repo.created_count + + body = { + "project_id": "proj-1", + "library_id": "lib-1", + "storage_key": "uploads/run/running.MOV", + "client_upload_id": "tok-run", + } + r = client.post("/api/v1/direct/complete", json=body) + assert r.status_code == 200 + b = r.json() + assert b["duplicated"] is True + assert b["ingest_job_id"] == "job-existing" + # 没有新建 job + assert ingest_repo.created_count == before + assert len(asset_repo.created) == 0 + + def test_error_asset_allows_reingest(self): + """ERROR 状态素材命中 → 不视为 duplicate，重新走 ingest。""" + errored = Asset( + id="asset-err", + project_id="proj-1", + library_id="lib-1", + name="err.MOV", + storage_key="uploads/err/err.MOV", + mime_type="video/quicktime", + status=AssetStatus.ERROR, + client_upload_id="tok-err", + ) + client, asset_repo, ingest_repo, _ = _client(asset_repo=StubAssetRepository([errored])) + body = { + "project_id": "proj-1", + "library_id": "lib-1", + "storage_key": "uploads/err/err.MOV", + "client_upload_id": "tok-err", + } + r = client.post("/api/v1/direct/complete", json=body) + assert r.status_code == 200 + b = r.json() + assert b["duplicated"] is False + assert b["ingest_job_id"] + assert ingest_repo.created_count == 1 + + def test_multipart_placeholder_without_job_submits_ingest(self): + """multipart 上传命中 PROCESSING 占位且无 job → 补提 ingest（不短路返 duplicated）。""" + placeholder = Asset( + id="asset-mp-placeholder", + project_id="proj-1", + library_id="lib-1", + name="mp.MOV", + storage_key="uploads/mp/mp.MOV", + mime_type="video/quicktime", + status=AssetStatus.PROCESSING, + client_upload_id="tok-mp", + ) + client, asset_repo, ingest_repo, storage = _client(asset_repo=StubAssetRepository([placeholder])) + storage.upload_file = MagicMock(return_value="https://oss/mp.MOV") + storage.get_url = MagicMock(return_value="https://oss/mp.MOV") + r = client.post( + "/api/v1", + data={ + "project_id": "proj-1", + "library_id": "lib-1", + "client_upload_id": "tok-mp", + }, + files={"file": ("mp.MOV", b"data", "video/quicktime")}, + ) + assert r.status_code == 200, r.text + b = r.json() + assert b["duplicated"] is False, f"multipart 占位被误判为 duplicate: {b}" + assert b["ingest_job_id"] + assert ingest_repo.created_count == 1