From f49a359903d0f504fcc0ef5d0f230689a09fb8f5 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Mon, 31 Aug 2026 08:18:54 +0000 Subject: [PATCH] style: auto-format with black + isort + prettier [skip ci-format-check] --- apps/worker/video_processing/dedup.py | 1 + tests/unit/test_duplicate_rate.py | 69 ++++++++++++++++++++------- 2 files changed, 53 insertions(+), 17 deletions(-) diff --git a/apps/worker/video_processing/dedup.py b/apps/worker/video_processing/dedup.py index dfe5bf115..16992ab30 100755 --- a/apps/worker/video_processing/dedup.py +++ b/apps/worker/video_processing/dedup.py @@ -334,6 +334,7 @@ class VideoDeduplicator: """ # 限制查询最近 100 个视频，避免大项目内存溢出 from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel + recent_models = ( session.query(GeneratedVideoModel) .filter(GeneratedVideoModel.project_id == project_id) diff --git a/tests/unit/test_duplicate_rate.py b/tests/unit/test_duplicate_rate.py index f7376e23a..b9d6fd2ed 100644 --- a/tests/unit/test_duplicate_rate.py +++ b/tests/unit/test_duplicate_rate.py @@ -1,4 +1,5 @@ """Tests for duplicate_rate computation and API response.""" + import json import sys from pathlib import Path @@ -20,6 +21,7 @@ class TestComputeDuplicateRate: def _make_fingerprint(self, md5="abc123", phashes=None): from video_processing.dedup import VideoFingerprint + return VideoFingerprint( md5=md5, keyframe_phashes=phashes or ["ff00ff00ff00ff00"], @@ -30,6 +32,7 @@ class TestComputeDuplicateRate: def _make_existing_video(self, vid, fingerprint_dict): from packages.domain import GeneratedVideo + return GeneratedVideo( id=vid, project_id="proj1", @@ -46,20 +49,25 @@ class TestComputeDuplicateRate: def test_no_existing_videos_returns_zero(self): from video_processing.dedup import VideoDeduplicator + deduplicator = VideoDeduplicator() fingerprint = self._make_fingerprint() session = MagicMock() with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo: mock_repo = MockRepo.return_value - session.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [] + session.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = ( + [] + ) rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session) assert rate == 0.0 def test_md5_match_returns_100(self): from video_processing.dedup import VideoDeduplicator + from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel + deduplicator = VideoDeduplicator() fingerprint = self._make_fingerprint(md5="exact_match_md5") session = MagicMock() @@ -76,14 +84,18 @@ class TestComputeDuplicateRate: mock_repo = MockRepo.return_value mock_repo._to_domain.return_value = existing # Mock the session.query chain - session.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [mock_model] + session.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [ + mock_model + ] rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session) assert rate == 100.0 def test_phash_similarity_computed(self): from video_processing.dedup import VideoDeduplicator + from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel + deduplicator = VideoDeduplicator() fingerprint = self._make_fingerprint(md5="different_md5", phashes=["ff00ff00ff00ff00"]) session = MagicMock() @@ -101,7 +113,9 @@ class TestComputeDuplicateRate: with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo: mock_repo = MockRepo.return_value mock_repo._to_domain.return_value = existing - session.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [mock_model] + session.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [ + mock_model + ] rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session) # hamming distance = 2, similarity = (1 - 2/64) * 100 = 96.875 @@ -109,7 +123,9 @@ class TestComputeDuplicateRate: def test_excludes_self_video(self): from video_processing.dedup import VideoDeduplicator + from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel + deduplicator = VideoDeduplicator() fingerprint = self._make_fingerprint(md5="same_md5") session = MagicMock() @@ -124,24 +140,24 @@ class TestComputeDuplicateRate: with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo: mock_repo = MockRepo.return_value mock_repo._to_domain.return_value = self_video - session.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [mock_model] + session.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [ + mock_model + ] rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session) assert rate == 0.0 def test_takes_max_similarity(self): from video_processing.dedup import VideoDeduplicator + from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel + deduplicator = VideoDeduplicator() fingerprint = self._make_fingerprint(md5="new_md5", phashes=["ff00ff00ff00ff00"]) session = MagicMock() - existing1 = self._make_existing_video( - "e1", {"md5": "md5_1", "keyframe_phashes": ["ff00ff00ff00ff0f"]} - ) - existing2 = self._make_existing_video( - "e2", {"md5": "md5_2", "keyframe_phashes": ["ff00ff00ff00ff01"]} - ) + existing1 = self._make_existing_video("e1", {"md5": "md5_1", "keyframe_phashes": ["ff00ff00ff00ff0f"]}) + existing2 = self._make_existing_video("e2", {"md5": "md5_2", "keyframe_phashes": ["ff00ff00ff00ff01"]}) mock_model1 = MagicMock(spec=GeneratedVideoModel) mock_model1.id = existing1.id mock_model1.project_id = existing1.project_id @@ -156,7 +172,10 @@ class TestComputeDuplicateRate: with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo: mock_repo = MockRepo.return_value mock_repo._to_domain.side_effect = [existing1, existing2] - session.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [mock_model1, mock_model2] + session.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [ + mock_model1, + mock_model2, + ] rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session) # max similarity: e2 distance=1, (1-1/64)*100 = 98.4375 @@ -168,19 +187,35 @@ class TestDuplicateRateAPI: def test_video_item_response_has_duplicate_rate(self): from app.schemas.video_center import VideoItemResponse + resp = VideoItemResponse( - id="v1", project_id="p1", generation_task_id="t1", - name="test.mp4", file_url="https://example.com/test.mp4", - file_size=1000, duration=10.0, width=1920, height=1080, fps=25.0, + id="v1", + project_id="p1", + generation_task_id="t1", + name="test.mp4", + file_url="https://example.com/test.mp4", + file_size=1000, + duration=10.0, + width=1920, + height=1080, + fps=25.0, duplicate_rate=75.5, ) assert resp.duplicate_rate == 75.5 def test_video_item_response_duplicate_rate_default_none(self): from app.schemas.video_center import VideoItemResponse + resp = VideoItemResponse( - id="v1", project_id="p1", generation_task_id="t1", - name="test.mp4", file_url="https://example.com/test.mp4", - file_size=1000, duration=10.0, width=1920, height=1080, fps=25.0, + id="v1", + project_id="p1", + generation_task_id="t1", + name="test.mp4", + file_url="https://example.com/test.mp4", + file_size=1000, + duration=10.0, + width=1920, + height=1080, + fps=25.0, ) assert resp.duplicate_rate is None