apps/api/app/api/routes/videos.py

@@ -15,6 +15,7 @@ from app.schemas.video_center import (
     VideoItemResponse,
 )
 from fastapi import APIRouter, Depends, HTTPException, Query, Response
+from pydantic import BaseModel, Field
 
 from packages.application import (
     GetGeneratedVideoUseCase,
@@ -239,3 +240,70 @@ def get_batch_download_status(
         status=api_status,
         download_url=download_url,
     )
+
+
+# ── 重新计算查重率 ─────────────────────────────────────────────────
+
+
+class RecomputeDedupRequest(BaseModel):
+    """重新计算查重率请求。"""
+
+    video_ids: list[str] | None = Field(
+        None,
+        description="指定视频 ID 列表。为空则对当前用户所有缺少查重数据的视频重新计算。",
+    )
+
+
+class RecomputeDedupResponse(BaseModel):
+    """重新计算查重率响应。"""
+
+    enqueued: int = Field(..., description="已入队的任务数量")
+    total_scanned: int = Field(..., description="扫描的视频总数")
+    skipped: int = Field(..., description="已有查重数据跳过的数量")
+    message: str = ""
+
+
+@router.post("/videos/recompute-dedup", response_model=RecomputeDedupResponse)
+def recompute_dedup(
+    request: RecomputeDedupRequest = RecomputeDedupRequest(),
+    repo=Depends(get_generated_video_repository),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+):
+    """重新计算视频的查重率/视觉相似度。
+
+    对于已存在但缺少 duplicate_rate / video_fingerprint 的视频，
+    触发异步 Celery 任务重新下载并计算指纹 + 查重率。
+
+    不传 video_ids 时，对当前用户所有视频进行检查。
+    """
+    user_id = current_user.user.id
+
+    # 获取目标视频列表
+    if request.video_ids:
+        all_videos = repo.get_by_ids(request.video_ids)
+        # 安全校验：只处理当前用户的视频
+        target_videos = [v for v in all_videos if v.user_id == user_id]
+    else:
+        target_videos = repo.list_by_user(user_id)
+
+    total_scanned = len(target_videos)
+    enqueued = 0
+    skipped = 0
+
+    for video in target_videos:
+        # 已有完整查重数据的跳过
+        if video.duplicate_rate is not None and video.video_fingerprint:
+            skipped += 1
+            continue
+
+        # 触发异步查重任务
+        celery_app.send_task("worker.check_duplicate", args=[video.id])
+        enqueued += 1
+        logger.info("Enqueued re-dedup for video %s (user=%s)", video.id, user_id)
+
+    return RecomputeDedupResponse(
+        enqueued=enqueued,
+        total_scanned=total_scanned,
+        skipped=skipped,
+        message=f"已入队 {enqueued} 个查重任务" if enqueued > 0 else "所有视频查重数据已完整",
+    )

apps/worker/video_processing/dedup_helpers.py

@@ -2,6 +2,9 @@
 
 供 generate_video 共同复用，
 创建 GeneratedVideo 记录后计算指纹并执行项目级 + 批次内查重。
+
+v2: 两阶段持久化 — 先计算所有查重数据，再一次性 commit，
+避免中间异常导致 duplicate_rate 等字段缺失。
 """
 
 from __future__ import annotations
@@ -34,24 +37,14 @@ def create_video_record_and_dedup(
 ) -> int:
     """创建 GeneratedVideo 记录，计算指纹并执行查重（历史 + 批次）。
 
-    Args:
-        generation_task_id: 生成任务 ID
-        project_id: 项目 ID
-        batch_id: 批次 ID（可为空字符串）
-        file_url: 视频文件 URL
-        file_size: 文件大小（字节）
-        duration: 视频时长（秒）
-        video_path: 视频本地路径（用于计算指纹）
-        mode: 剪辑模式名称
-        session: 数据库会话
-        width: 视频宽度
-        height: 视频高度
-        fps: 视频帧率
+    采用两阶段持久化：先计算所有指纹/查重数据（内存），
+    再一次性写入数据库并 commit。若指纹计算失败，
+    视频记录仍会创建（无查重数据），但保证不会出现"写了记录却没 commit"的中间态。
 
     Returns:
         创建的视频记录数量（1 表示成功，0 表示失败）
     """
-    from video_processing.dedup import VideoDeduplicator
+    from video_processing.dedup import VideoDeduplicator, _save_fingerprint_chunks
 
     from packages.adapters.sqlalchemy_impl.generated_video_repository import (
         SQLAlchemyGeneratedVideoRepository,
@@ -60,8 +53,9 @@ def create_video_record_and_dedup(
 
     try:
         video_id = uuid4().hex
-        # 使用传入的名称，没有则 fallback 到默认命名
         video_name = name.strip() if name else f"generated-{generation_task_id[:8]}.mp4"
+
+        # ── Phase 1: 构建视频记录（内存，不 commit） ────────────────
         generated_video = GeneratedVideo(
             id=video_id,
             project_id=project_id,
@@ -76,98 +70,94 @@ def create_video_record_and_dedup(
             fps=fps,
             status="completed",
             generation_params={"mode": mode},
+            thumbnail_url=thumbnail_url or None,
         )
 
-        video_repo = SQLAlchemyGeneratedVideoRepository(session)
-        video_repo.create(generated_video)
-
-        # 生成封面缩略图
-        if thumbnail_url:
-            generated_video.thumbnail_url = thumbnail_url
-            video_repo.update_thumbnail(video_id, thumbnail_url)
-            logger.info("Thumbnail set for video %s: %s", video_id, thumbnail_url[:80] if thumbnail_url else "")
-        else:
-            logger.debug("No thumbnail_url provided for video %s, skipping", video_id)
-
-        # 计算视频指纹
+        # ── Phase 2: 计算指纹 & 查重（全部在内存） ────────────────
         deduplicator = VideoDeduplicator()
+        fingerprint = 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)
-            session.commit()
-            return 1
 
-        generated_video.video_fingerprint = fingerprint.to_dict()
+        if fingerprint is not None:
+            generated_video.video_fingerprint = fingerprint.to_dict()
 
-        # 写入分片指纹表
-        from video_processing.dedup import _save_fingerprint_chunks
+            # 写入分片指纹表（失败不阻塞）
+            try:
+                _save_fingerprint_chunks(fingerprint, video_id, project_id, user_id, session)
+            except Exception as chunk_err:
+                logger.warning("Failed to save fingerprint chunks for %s: %s", video_id, chunk_err)
 
-        try:
-            _save_fingerprint_chunks(fingerprint, video_id, project_id, user_id, session)
-        except Exception as chunk_err:
-            logger.warning("Failed to save fingerprint chunks for %s: %s", video_id, chunk_err)
-
-        # (a) 历史成片查重（跨项目全局 + 时长预过滤）
-        duration_sec = fingerprint.duration / 1000 if fingerprint.duration else 0
-        duplicate_result = deduplicator.check_duplicate(
-            fingerprint,
-            project_id,
-            session,
-            scope="user",
-            user_id=user_id,
-            duration_sec=duration_sec,
-        )
-
-        # (b) 批次内查重（仅当有 batch_id 时）
-        if not duplicate_result and batch_id:
-            duplicate_result = deduplicator.check_batch_duplicate(fingerprint, batch_id, video_id, session)
-
-        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(
+            # (a) 历史成片查重（跨项目全局 + 时长预过滤）
+            duration_sec = fingerprint.duration / 1000 if fingerprint.duration else 0
+            duplicate_result = deduplicator.check_duplicate(
                 fingerprint,
                 project_id,
-                video_id,
                 session,
                 scope="user",
                 user_id=user_id,
+                duration_sec=duration_sec,
             )
-            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
 
-        video_repo.update(generated_video)
+            # (b) 批次内查重（仅当有 batch_id 时）
+            if not duplicate_result and batch_id:
+                duplicate_result = deduplicator.check_batch_duplicate(fingerprint, batch_id, video_id, session)
+
+            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])
+
         session.commit()
         logger.info(
-            "GeneratedVideo record created: %s (task=%s, dup=%s)",
+            "GeneratedVideo record created: %s (task=%s, dup=%s, rate=%s)",
             video_id,
             generation_task_id,
             generated_video.is_duplicate,
+            generated_video.duplicate_rate,
         )
         return 1
     except Exception as e:

tests/unit/test_dedup_helpers_user_id.py

@@ -159,6 +159,6 @@ class TestDedupHelpersUserIdPassthrough:
             )
 
         # 验证 update 被调用（包含 duplicate_rate 的记录）
-        mock_video_repo.update.assert_called_once()
-        updated_video = mock_video_repo.update.call_args[0][0]
+        mock_video_repo.create.assert_called_once()
+        updated_video = mock_video_repo.create.call_args[0][0]
         assert updated_video.duplicate_rate == 78.5

tests/unit/test_dedup_two_phase_commit.py

@@ -0,0 +1,234 @@
+"""Tests for two-phase commit pattern in dedup_helpers (#1664 follow-up).
+
+Verifies that the new dedup_helpers.py:
+1. Creates video with all dedup fields in a single commit
+2. Still creates video when fingerprint computation fails
+3. Creates video with fingerprint but no rate when rate computation fails
+4. Never does a partial commit (no create + separate update)
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+# Mock cv2/numpy before imports
+sys.modules.setdefault("cv2", MagicMock())
+sys.modules.setdefault("numpy", MagicMock())
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT / "apps" / "api"))
+sys.path.insert(0, str(ROOT / "packages"))
+sys.path.insert(0, str(ROOT / "apps" / "worker"))
+
+import os
+
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret")
+os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
+
+import pytest
+from video_processing.dedup_helpers import create_video_record_and_dedup
+
+
+@pytest.fixture
+def session():
+    s = MagicMock()
+    return s
+
+
+@pytest.fixture
+def mock_fingerprint():
+    fp = MagicMock()
+    fp.duration = 15000  # 15 seconds in ms
+    fp.to_dict.return_value = {"md5": "abc123", "keyframe_phashes": ["aabb"], "color_histograms": []}
+    fp.chunks = []
+    fp.keyframe_phashes = ["aabb"]
+    fp.color_histograms = []
+    fp.md5 = "abc123"
+    return fp
+
+
+class TestTwoPhaseCommit:
+    """Verify that dedup data is computed before commit."""
+
+    def test_video_created_with_all_dedup_fields(self, session, mock_fingerprint):
+        """When all computations succeed, video is created with all fields in one commit."""
+        mock_repo = MagicMock()
+        mock_deduplicator = MagicMock()
+        mock_deduplicator.compute_fingerprint.return_value = mock_fingerprint
+        mock_deduplicator.check_duplicate.return_value = None
+        mock_deduplicator.compute_duplicate_rate.return_value = {
+            "duplicate_rate": 42.5,
+            "visual_similarity": 0.75,
+            "match_count": 2,
+        }
+
+        with (
+            patch(
+                "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository",
+                return_value=mock_repo,
+            ),
+            patch("video_processing.dedup.VideoDeduplicator", return_value=mock_deduplicator),
+            patch("video_processing.dedup._save_fingerprint_chunks"),
+        ):
+            result = create_video_record_and_dedup(
+                generation_task_id="task-001",
+                project_id="proj-001",
+                user_id="user-001",
+                batch_id="",
+                file_url="https://example.com/v.mp4",
+                file_size=1024,
+                duration=15.0,
+                video_path="/tmp/fake.mp4",
+                mode="smart",
+                session=session,
+            )
+
+        assert result == 1
+        # create() should be called exactly once with the complete video object
+        mock_repo.create.assert_called_once()
+        created_video = mock_repo.create.call_args[0][0]
+        assert created_video.duplicate_rate == 42.5
+        assert created_video.visual_similarity == 0.75
+        assert created_video.match_count == 2
+        assert created_video.video_fingerprint is not None
+        # session.commit should be called exactly once (at the end)
+        session.commit.assert_called_once()
+
+    def test_video_created_even_when_fingerprint_fails(self, session):
+        """When fingerprint computation fails, video is still created (without dedup data)."""
+        mock_repo = MagicMock()
+        mock_deduplicator = MagicMock()
+        mock_deduplicator.compute_fingerprint.side_effect = RuntimeError("cv2 not available")
+
+        with (
+            patch(
+                "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository",
+                return_value=mock_repo,
+            ),
+            patch("video_processing.dedup.VideoDeduplicator", return_value=mock_deduplicator),
+        ):
+            result = create_video_record_and_dedup(
+                generation_task_id="task-002",
+                project_id="proj-001",
+                user_id="user-001",
+                batch_id="",
+                file_url="https://example.com/v.mp4",
+                file_size=1024,
+                duration=15.0,
+                video_path="/tmp/fake.mp4",
+                mode="smart",
+                session=session,
+            )
+
+        assert result == 1
+        mock_repo.create.assert_called_once()
+        created_video = mock_repo.create.call_args[0][0]
+        assert created_video.duplicate_rate is None
+        assert created_video.video_fingerprint is None
+        session.commit.assert_called_once()
+        # No dedup methods should have been called
+        mock_deduplicator.check_duplicate.assert_not_called()
+        mock_deduplicator.compute_duplicate_rate.assert_not_called()
+
+    def test_video_created_with_fingerprint_but_no_rate(self, session, mock_fingerprint):
+        """When rate computation fails, video is created with fingerprint but no rate."""
+        mock_repo = MagicMock()
+        mock_deduplicator = MagicMock()
+        mock_deduplicator.compute_fingerprint.return_value = mock_fingerprint
+        mock_deduplicator.check_duplicate.return_value = None
+        mock_deduplicator.compute_duplicate_rate.side_effect = RuntimeError("DB error")
+
+        with (
+            patch(
+                "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository",
+                return_value=mock_repo,
+            ),
+            patch("video_processing.dedup.VideoDeduplicator", return_value=mock_deduplicator),
+            patch("video_processing.dedup._save_fingerprint_chunks"),
+        ):
+            result = create_video_record_and_dedup(
+                generation_task_id="task-003",
+                project_id="proj-001",
+                user_id="user-001",
+                batch_id="",
+                file_url="https://example.com/v.mp4",
+                file_size=1024,
+                duration=15.0,
+                video_path="/tmp/fake.mp4",
+                mode="smart",
+                session=session,
+            )
+
+        assert result == 1
+        mock_repo.create.assert_called_once()
+        created_video = mock_repo.create.call_args[0][0]
+        # Fingerprint should be set
+        assert created_video.video_fingerprint is not None
+        # But duplicate_rate should be None
+        assert created_video.duplicate_rate is None
+        session.commit.assert_called_once()
+
+    def test_no_separate_update_call(self, session, mock_fingerprint):
+        """Verify the new pattern uses create() only, not create() + update()."""
+        mock_repo = MagicMock()
+        mock_deduplicator = MagicMock()
+        mock_deduplicator.compute_fingerprint.return_value = mock_fingerprint
+        mock_deduplicator.check_duplicate.return_value = None
+        mock_deduplicator.compute_duplicate_rate.return_value = {
+            "duplicate_rate": 10.0,
+            "visual_similarity": 0.5,
+            "match_count": 1,
+        }
+
+        with (
+            patch(
+                "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository",
+                return_value=mock_repo,
+            ),
+            patch("video_processing.dedup.VideoDeduplicator", return_value=mock_deduplicator),
+            patch("video_processing.dedup._save_fingerprint_chunks"),
+        ):
+            create_video_record_and_dedup(
+                generation_task_id="task-004",
+                project_id="proj-001",
+                user_id="user-001",
+                batch_id="",
+                file_url="https://example.com/v.mp4",
+                file_size=1024,
+                duration=15.0,
+                video_path="/tmp/fake.mp4",
+                mode="smart",
+                session=session,
+            )
+
+        # Only create() should be called, not update()
+        mock_repo.create.assert_called_once()
+        mock_repo.update.assert_not_called()
+
+    def test_commit_not_called_on_total_failure(self, session):
+        """When the entire function fails, session.rollback is called instead of commit."""
+        mock_repo = MagicMock()
+        mock_repo.create.side_effect = RuntimeError("DB connection lost")
+
+        with patch(
+            "packages.adapters.sqlalchemy_impl.generated_video_repository.SQLAlchemyGeneratedVideoRepository",
+            return_value=mock_repo,
+        ):
+            result = create_video_record_and_dedup(
+                generation_task_id="task-005",
+                project_id="proj-001",
+                user_id="user-001",
+                batch_id="",
+                file_url="https://example.com/v.mp4",
+                file_size=1024,
+                duration=15.0,
+                video_path="/tmp/fake.mp4",
+                mode="smart",
+                session=session,
+            )
+
+        assert result == 0
+        session.commit.assert_not_called()
+        session.rollback.assert_called_once()

tests/unit/test_recompute_dedup_api.py

@@ -0,0 +1,167 @@
+"""Tests for POST /videos/recompute-dedup endpoint (#1664 follow-up)."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from fastapi.testclient import TestClient
+
+
+@pytest.fixture
+def mock_video():
+    """Mock video with missing dedup data."""
+    v = MagicMock()
+    v.id = "video-001"
+    v.user_id = "user-abc"
+    v.duplicate_rate = None
+    v.video_fingerprint = None
+    v.project_id = "proj-001"
+    v.generation_task_id = "task-001"
+    v.name = "test.mp4"
+    v.file_url = "https://example.com/test.mp4"
+    v.file_size = 1024
+    v.duration = 10.0
+    v.width = 1920
+    v.height = 1080
+    v.fps = 25.0
+    v.status = "completed"
+    v.review_status = "pending_review"
+    v.generation_params = {}
+    v.thumbnail_url = None
+    v.is_duplicate = False
+    v.duplicate_of = None
+    v.match_count = None
+    v.visual_similarity = None
+    v.generated_at = "2026-09-04T00:00:00"
+    return v
+
+
+@pytest.fixture
+def mock_video_with_dedup(mock_video):
+    """Mock video that already has dedup data."""
+    mock_video.duplicate_rate = 15.5
+    mock_video.video_fingerprint = {"md5": "abc123"}
+    return mock_video
+
+
+class TestRecomputeDedupEndpoint:
+    """POST /videos/recompute-dedup"""
+
+    def test_enqueue_videos_without_dedup(self, mock_video):
+        """Videos missing duplicate_rate should be enqueued."""
+        from app.api.routes.videos import RecomputeDedupRequest
+
+        mock_repo = MagicMock()
+        mock_repo.list_by_user.return_value = [mock_video]
+
+        with (patch("app.api.routes.videos.celery_app") as mock_celery,):
+            mock_celery.send_task.return_value = MagicMock(id="task-xyz")
+            from app.api.routes.videos import recompute_dedup
+
+            auth_user = MagicMock()
+            auth_user.user.id = "user-abc"
+
+            result = recompute_dedup(
+                request=RecomputeDedupRequest(),
+                repo=mock_repo,
+                current_user=auth_user,
+            )
+
+        assert result.enqueued == 1
+        assert result.total_scanned == 1
+        assert result.skipped == 0
+        mock_celery.send_task.assert_called_once_with("worker.check_duplicate", args=["video-001"])
+
+    def test_skip_videos_with_complete_dedup(self, mock_video_with_dedup):
+        """Videos with both duplicate_rate and video_fingerprint should be skipped."""
+        from app.api.routes.videos import RecomputeDedupRequest, recompute_dedup
+
+        mock_repo = MagicMock()
+        mock_repo.list_by_user.return_value = [mock_video_with_dedup]
+
+        with patch("app.api.routes.videos.celery_app") as mock_celery:
+            auth_user = MagicMock()
+            auth_user.user.id = "user-abc"
+
+            result = recompute_dedup(
+                request=RecomputeDedupRequest(),
+                repo=mock_repo,
+                current_user=auth_user,
+            )
+
+        assert result.enqueued == 0
+        assert result.total_scanned == 1
+        assert result.skipped == 1
+        mock_celery.send_task.assert_not_called()
+
+    def test_specific_video_ids(self, mock_video):
+        """When video_ids are provided, only those videos are processed."""
+        from app.api.routes.videos import RecomputeDedupRequest, recompute_dedup
+
+        mock_repo = MagicMock()
+        mock_repo.get_by_ids.return_value = [mock_video]
+
+        with patch("app.api.routes.videos.celery_app") as mock_celery:
+            mock_celery.send_task.return_value = MagicMock(id="task-xyz")
+            auth_user = MagicMock()
+            auth_user.user.id = "user-abc"
+
+            result = recompute_dedup(
+                request=RecomputeDedupRequest(video_ids=["video-001"]),
+                repo=mock_repo,
+                current_user=auth_user,
+            )
+
+        assert result.enqueued == 1
+        mock_repo.get_by_ids.assert_called_once_with(["video-001"])
+
+    def test_security_only_own_videos(self, mock_video):
+        """Videos belonging to other users should be filtered out."""
+        from app.api.routes.videos import RecomputeDedupRequest, recompute_dedup
+
+        mock_video.user_id = "user-OTHER"
+        mock_repo = MagicMock()
+        mock_repo.get_by_ids.return_value = [mock_video]
+
+        with patch("app.api.routes.videos.celery_app") as mock_celery:
+            auth_user = MagicMock()
+            auth_user.user.id = "user-abc"
+
+            result = recompute_dedup(
+                request=RecomputeDedupRequest(video_ids=["video-001"]),
+                repo=mock_repo,
+                current_user=auth_user,
+            )
+
+        assert result.enqueued == 0
+        mock_celery.send_task.assert_not_called()
+
+    def test_mixed_complete_and_incomplete(self, mock_video, mock_video_with_dedup):
+        """Mix of videos with and without dedup data."""
+        import copy
+
+        from app.api.routes.videos import RecomputeDedupRequest, recompute_dedup
+
+        # Create a second video object
+        v2 = MagicMock()
+        v2.id = "video-002"
+        v2.user_id = "user-abc"
+        v2.duplicate_rate = None
+        v2.video_fingerprint = None
+
+        mock_repo = MagicMock()
+        mock_repo.list_by_user.return_value = [mock_video_with_dedup, v2]
+
+        with patch("app.api.routes.videos.celery_app") as mock_celery:
+            mock_celery.send_task.return_value = MagicMock(id="task-xyz")
+            auth_user = MagicMock()
+            auth_user.user.id = "user-abc"
+
+            result = recompute_dedup(
+                request=RecomputeDedupRequest(),
+                repo=mock_repo,
+                current_user=auth_user,
+            )
+
+        assert result.enqueued == 1
+        assert result.total_scanned == 2
+        assert result.skipped == 1