fix: 查重流程全面核实修复 — 两阶段持久化 + 重新计算查重API (#1664) (#1684)

Co-authored-by: xiaoxia <dev@xiaoxiajianji.com>
Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>

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()