style: auto-format with black + isort + prettier [skip ci-format-check]

apps/worker/video_processing/dedup.py

@@ -26,28 +26,29 @@ from packages.shared.storage import get_storage_service
 logger = logging.getLogger(__name__)
 
 # ── 关键帧检测常量 ──────────────────────────────────────────────
-SCENE_CHANGE_THRESHOLD = 30       # 灰度差异阈值
-MIN_KEYFRAME_INTERVAL_SEC = 1.0   # 最小关键帧间隔（秒）
-MAX_KEYFRAMES = 30                # 最大关键帧数
-MIN_KEYFRAMES = 5                 # 最小关键帧数
-LONG_VIDEO_SEGMENT_SEC = 30       # 长视频每段秒数
+SCENE_CHANGE_THRESHOLD = 30  # 灰度差异阈值
+MIN_KEYFRAME_INTERVAL_SEC = 1.0  # 最小关键帧间隔（秒）
+MAX_KEYFRAMES = 30  # 最大关键帧数
+MIN_KEYFRAMES = 5  # 最小关键帧数
+LONG_VIDEO_SEGMENT_SEC = 30  # 长视频每段秒数
 LONG_VIDEO_DURATION_THRESHOLD_SEC = 180  # 3 分钟阈值
-MIN_FRAMES_PER_SEGMENT = 2        # 长视频每段最少帧数
+MIN_FRAMES_PER_SEGMENT = 2  # 长视频每段最少帧数
 
 # ── 滑动窗口匹配常量 ────────────────────────────────────────────
-SEGMENT_MATCH_THRESHOLD = 8       # 帧匹配汉明距离阈值
-MIN_CONSECUTIVE_MATCHES = 5       # 最少连续匹配帧数
-MAX_GAP = 2                       # 允许的最大间隙帧数
+SEGMENT_MATCH_THRESHOLD = 8  # 帧匹配汉明距离阈值
+MIN_CONSECUTIVE_MATCHES = 5  # 最少连续匹配帧数
+MAX_GAP = 2  # 允许的最大间隙帧数
 
 # ── 融合判定常量 ────────────────────────────────────────────────
-PHASH_WEIGHT = 0.7              # pHash 权重
-HISTOGRAM_WEIGHT = 0.3          # 直方图权重
-MATCH_RATIO_THRESHOLD = 0.7     # 至少 70% 帧匹配
-DUPLICATE_THRESHOLD = 0.70      # 融合后相似度阈值
+PHASH_WEIGHT = 0.7  # pHash 权重
+HISTOGRAM_WEIGHT = 0.3  # 直方图权重
+MATCH_RATIO_THRESHOLD = 0.7  # 至少 70% 帧匹配
+DUPLICATE_THRESHOLD = 0.70  # 融合后相似度阈值
 
 
 # ── 感知哈希 & 颜色直方图工具函数 ────────────────────────────────
 
+
 def compute_phash(image: np.ndarray, hash_size: int = 8) -> str:
     """计算图像的感知哈希（pHash），基于 DCT（离散余弦变换）。
 
@@ -110,6 +111,7 @@ def compute_color_histogram(image: np.ndarray, bins: int = 32) -> list[float]:
 
 # ── 关键帧检测 ──────────────────────────────────────────────────
 
+
 def detect_keyframe_timestamps(
     video_path: str,
     *,
@@ -209,6 +211,7 @@ def detect_keyframe_timestamps(
 
 # ── 数据类 ──────────────────────────────────────────────────────
 
+
 @dataclass
 class FingerprintChunk:
     """单个分片指纹数据。"""
@@ -287,6 +290,7 @@ class VideoFingerprint:
 
 # ── 滑动窗口时序匹配 ────────────────────────────────────────────
 
+
 def find_duplicate_segments(
     query_chunks: list,
     target_chunks: list,
@@ -414,6 +418,7 @@ def find_duplicate_segments(
 
 # ── VideoDeduplicator ───────────────────────────────────────────
 
+
 class VideoDeduplicator:
     """Video deduplication using multiple fingerprint methods."""
 
@@ -625,10 +630,11 @@ class VideoDeduplicator:
                 continue
 
             # 滑动窗口时序匹配：获取具体重复片段
-            existing_chunk_objects = chunk_data if chunk_data else [
-                {"phash_binary": p, "start_time_ms": 0, "end_time_ms": 0}
-                for p in existing_phashes
-            ]
+            existing_chunk_objects = (
+                chunk_data
+                if chunk_data
+                else [{"phash_binary": p, "start_time_ms": 0, "end_time_ms": 0} for p in existing_phashes]
+            )
             segments = find_duplicate_segments(fingerprint.chunks, existing_chunk_objects)
 
             return {
@@ -736,10 +742,11 @@ class VideoDeduplicator:
                 continue
 
             # 滑动窗口时序匹配
-            existing_chunk_objects = chunk_data if chunk_data else [
-                {"phash_binary": p, "start_time_ms": 0, "end_time_ms": 0}
-                for p in existing_phashes
-            ]
+            existing_chunk_objects = (
+                chunk_data
+                if chunk_data
+                else [{"phash_binary": p, "start_time_ms": 0, "end_time_ms": 0} for p in existing_phashes]
+            )
             segments = find_duplicate_segments(fingerprint.chunks, existing_chunk_objects)
 
             return {

tests/unit/test_dedup_engine.py

@@ -285,7 +285,9 @@ class TestVideoDeduplicatorCheckDuplicate:
             result = deduplicator.check_duplicate(fingerprint, "proj-1", mock_session)
             assert result is not None
             assert result["duplicate"] is True
-            assert result["similarity"] == pytest.approx(0.85, abs=0.01)  # combined: 0.7*1.0 + 0.3*0.5 (no hist fallback)
+            assert result["similarity"] == pytest.approx(
+                0.85, abs=0.01
+            )  # combined: 0.7*1.0 + 0.3*0.5 (no hist fallback)
             assert result["reason"] == "phash_histogram_fusion"
         finally:
             self._restore_repo(mod, orig)

tests/unit/test_dedup_v2.py

@@ -95,19 +95,19 @@ sys.modules["packages.shared.storage"] = _mock_module()
 import video_processing.dedup as _dedup_mod
 from video_processing.dedup import (  # noqa: E402
     DUPLICATE_THRESHOLD,
-    DuplicateSegment,
-    FingerprintChunk,
     HISTOGRAM_WEIGHT,
     LONG_VIDEO_DURATION_THRESHOLD_SEC,
     MATCH_RATIO_THRESHOLD,
     MAX_GAP,
     MAX_KEYFRAMES,
     MIN_CONSECUTIVE_MATCHES,
-    MIN_KEYFRAMES,
     MIN_KEYFRAME_INTERVAL_SEC,
+    MIN_KEYFRAMES,
     PHASH_WEIGHT,
     SCENE_CHANGE_THRESHOLD,
     SEGMENT_MATCH_THRESHOLD,
+    DuplicateSegment,
+    FingerprintChunk,
     VideoDeduplicator,
     VideoFingerprint,
     detect_keyframe_timestamps,
@@ -129,6 +129,7 @@ del _SAVED_MODULES_KEYS, _SAVED_MODULES_VALUES, _key, _value
 
 # ── Helper ──────────────────────────────────────────────────────
 
+
 def _make_chunk(start_ms: int, end_ms: int, phash: str, hist: list[float] | None = None) -> FingerprintChunk:
     """创建测试用 FingerprintChunk."""
     return FingerprintChunk(
@@ -142,6 +143,7 @@ def _make_chunk(start_ms: int, end_ms: int, phash: str, hist: list[float] | None
 
 # ── TestDuplicateSegment ────────────────────────────────────────
 
+
 class TestDuplicateSegment:
     """DuplicateSegment 数据类测试."""
 
@@ -167,6 +169,7 @@ class TestDuplicateSegment:
 
 # ── TestDetectKeyframeTimestamps ────────────────────────────────
 
+
 class TestDetectKeyframeTimestamps:
     """detect_keyframe_timestamps 关键帧检测测试.
 
@@ -182,6 +185,7 @@ class TestDetectKeyframeTimestamps:
         cv2_mock.VideoCapture.return_value = mock_cap
 
         import pytest
+
         with pytest.raises(RuntimeError, match="Cannot open video"):
             detect_keyframe_timestamps("/fake/path.mp4")
 
@@ -201,6 +205,7 @@ class TestDetectKeyframeTimestamps:
     def test_function_signature(self):
         """验证函数签名和默认参数."""
         import inspect
+
         sig = inspect.signature(detect_keyframe_timestamps)
         params = sig.parameters
         assert "video_path" in params
@@ -213,9 +218,9 @@ class TestDetectKeyframeTimestamps:
         assert params["min_frames"].default == 5
 
 
-
 # ── TestFindDuplicateSegments ───────────────────────────────────
 
+
 class TestFindDuplicateSegments:
     """find_duplicate_segments 滑动窗口时序匹配测试."""
 
@@ -246,10 +251,12 @@ class TestFindDuplicateSegments:
         diff_hash_a = "0000000000000000"
         diff_hash_b = "ffffffffffffffff"
 
-        chunks_a = [_make_chunk(i * 1000, (i + 1) * 1000, same_hash) for i in range(5)] + \
-                   [_make_chunk(i * 1000, (i + 1) * 1000, diff_hash_a) for i in range(5, 10)]
-        chunks_b = [_make_chunk(i * 1000, (i + 1) * 1000, same_hash) for i in range(5)] + \
-                   [_make_chunk(i * 1000, (i + 1) * 1000, diff_hash_b) for i in range(5, 10)]
+        chunks_a = [_make_chunk(i * 1000, (i + 1) * 1000, same_hash) for i in range(5)] + [
+            _make_chunk(i * 1000, (i + 1) * 1000, diff_hash_a) for i in range(5, 10)
+        ]
+        chunks_b = [_make_chunk(i * 1000, (i + 1) * 1000, same_hash) for i in range(5)] + [
+            _make_chunk(i * 1000, (i + 1) * 1000, diff_hash_b) for i in range(5, 10)
+        ]
 
         segments = find_duplicate_segments(chunks_a, chunks_b)
         # 应该只有前 5 帧的匹配段
@@ -263,10 +270,12 @@ class TestFindDuplicateSegments:
         """
         same_hash = "aaaaaaaaaaaaaaaa"
         # 4 帧匹配，后面 6 帧各自不同（在 query 和 target 中使用不同 hash）
-        chunks_a = [_make_chunk(i * 1000, (i + 1) * 1000, same_hash) for i in range(4)] + \
-                   [_make_chunk(i * 1000, (i + 1) * 1000, "bbbbbbbbbbbbbbbb") for i in range(4, 10)]
-        chunks_b = [_make_chunk(i * 1000, (i + 1) * 1000, same_hash) for i in range(4)] + \
-                   [_make_chunk(i * 1000, (i + 1) * 1000, "cccccccccccccccc") for i in range(4, 10)]
+        chunks_a = [_make_chunk(i * 1000, (i + 1) * 1000, same_hash) for i in range(4)] + [
+            _make_chunk(i * 1000, (i + 1) * 1000, "bbbbbbbbbbbbbbbb") for i in range(4, 10)
+        ]
+        chunks_b = [_make_chunk(i * 1000, (i + 1) * 1000, same_hash) for i in range(4)] + [
+            _make_chunk(i * 1000, (i + 1) * 1000, "cccccccccccccccc") for i in range(4, 10)
+        ]
 
         # hamming("bbbb...", "cccc...") should be > 8 (SEGMENT_MATCH_THRESHOLD)
         # b=1011, c=1100 → 4 bits differ per hex digit × 16 digits = 64 bits total? No...
@@ -333,10 +342,14 @@ class TestFindDuplicateSegments:
 
     def test_dict_chunks_compatibility(self):
         """dict 格式的 chunks 也能正常工作."""
-        chunks_a = [{"phash_binary": "aaaaaaaaaaaaaaaa", "start_time_ms": i * 1000, "end_time_ms": (i + 1) * 1000}
-                    for i in range(10)]
-        chunks_b = [{"phash_binary": "aaaaaaaaaaaaaaaa", "start_time_ms": i * 1000, "end_time_ms": (i + 1) * 1000}
-                    for i in range(10)]
+        chunks_a = [
+            {"phash_binary": "aaaaaaaaaaaaaaaa", "start_time_ms": i * 1000, "end_time_ms": (i + 1) * 1000}
+            for i in range(10)
+        ]
+        chunks_b = [
+            {"phash_binary": "aaaaaaaaaaaaaaaa", "start_time_ms": i * 1000, "end_time_ms": (i + 1) * 1000}
+            for i in range(10)
+        ]
 
         segments = find_duplicate_segments(chunks_a, chunks_b)
         assert len(segments) >= 1
@@ -347,6 +360,7 @@ class TestFindDuplicateSegments:
         每个 query chunk 匹配到 target 中对应的 chunk（相同 hash），
         确保 target 时间范围正确映射。
         """
+
         # 给每个 chunk 唯一的 hash（但保证 query[i] == target[i]）
         def _unique_hash(i: int) -> str:
             return format(i, "016x")
@@ -367,6 +381,7 @@ class TestFindDuplicateSegments:
 
 # ── TestMedianVsMean ────────────────────────────────────────────
 
+
 class TestMedianVsMean:
     """中位数 vs 均值：验证中位数抵抗异常值."""
 
@@ -376,6 +391,7 @@ class TestMedianVsMean:
         但更极端的：[3,3,3,3,60]：均值=14.4，中位数=3.
         """
         import statistics
+
         distances = [3, 3, 3, 3, 60]
         assert statistics.median(distances) == 3
         assert sum(distances) / len(distances) == 14.4
@@ -385,6 +401,7 @@ class TestMedianVsMean:
 
 # ── TestMatchRatioCondition ─────────────────────────────────────
 
+
 class TestMatchRatioCondition:
     """帧匹配比例条件测试."""
 
@@ -409,6 +426,7 @@ class TestMatchRatioCondition:
 
 # ── TestBhattacharyyaFusion ─────────────────────────────────────
 
+
 class TestBhattacharyyaFusion:
     """直方图融合逻辑测试."""
 
@@ -438,6 +456,7 @@ class TestBhattacharyyaFusion:
 
 # ── TestBackwardCompatibility ───────────────────────────────────
 
+
 class TestBackwardCompatibility:
     """向后兼容测试."""
 
@@ -453,8 +472,7 @@ class TestBackwardCompatibility:
         # 实际上我们的实现用 _get_start/_get_end 访问，缺 key 会 KeyError
         # 所以 check_duplicate 传入时会补上默认值
         target_with_defaults = [
-            {"phash_binary": "aaaaaaaaaaaaaaaa", "start_time_ms": 0, "end_time_ms": 0}
-            for _ in range(10)
+            {"phash_binary": "aaaaaaaaaaaaaaaa", "start_time_ms": 0, "end_time_ms": 0} for _ in range(10)
         ]
         segments = find_duplicate_segments(query_chunks, target_with_defaults)
         # 不会崩溃
@@ -472,6 +490,7 @@ class TestBackwardCompatibility:
 
 # ── TestConstants ───────────────────────────────────────────────
 
+
 class TestConstants:
     """常量值验证 — 使用已在模块顶部导入的常量，避免重新 import."""
 

tests/unit/test_fingerprint_chunks.py

@@ -124,7 +124,6 @@ for _key, _value in _SAVED_MODULES_VALUES.items():
 del _SAVED_MODULES_KEYS, _SAVED_MODULES_VALUES, _key, _value
 
 
-
 class TestVideoFingerprintToChunkModels:
     """测试 VideoFingerprint.to_chunk_models() 输出。"""