apps/worker/video_processing/dedup.py

@@ -37,14 +37,19 @@ LONG_VIDEO_SEGMENT_SEC = 30  # 长视频每段秒数
 LONG_VIDEO_DURATION_THRESHOLD_SEC = 180  # 3 分钟阈值
 MIN_FRAMES_PER_SEGMENT = 2  # 长视频每段最少帧数
 
-# ── 滑动窗口匹配常量（Issue #1702 重新校准） ─────────────────────
-# 阈值经 staging 真实数据回归校准（2026-09-05，worker 容器内离线实验）：
-#   - 同源成片对（20s/11s，各自 2-5% 随机边缘裁剪降重，1s 密集采样）：
-#     全部帧对最小汉明距离 min=8，<=12 命中 10/31 帧（B->A 4/11）
-#   - 异源成片对（4 个不同项目真实视频）：最小距离 24，<=16 命中 0 帧
-# 8（#1658 旧值）会漏掉同源裁剪（自对照实验：同帧两次 2-5% 随机裁剪距离 4~10），
-# 12 能检出同源/局部复用且与异源分布（>=24）间隔 12bit，无误报空间。
-PHASH_THRESHOLD = 12
+# ── 滑动窗口匹配常量（Issue #1702 二次校准） ─────────────────────
+# 阈值经 staging 真实数据两轮回归校准（worker 容器内离线实验）：
+#   第一轮（2026-09-05）：同源对 <=12 命中 4/11，异源最小距离 24 → 定 12；
+#   第二轮（2026-09-05，证据视频 B->A 仍漏检）：扩大样本到该用户全部
+#   15 个真实成片（13 个异源候选）实测：
+#   - 同源成片对（A 20s / B、C 各 11.75s，1s 密集采样）：
+#       B->A 中位数距离 14，<=16 命中 8/11=0.73；C->A 8/11=0.73
+#   - 异源成片对（13 个真实视频）：每帧全局最近邻最小距离 18，
+#       <=16 命中帧数全部为 0（最近邻 18 仅个别帧，中位数 22~28）
+# 12 漏掉同源降重对（降重滤镜/字幕/画面扰动把距离从 ~8 推到 14~16）；
+# 16 对同源命中 0.73+ 且与异源分布（最近邻 >=18）仍有 >=2bit 安全裕度，
+# 异源 <=16 命中 0 帧，无误报空间。
+PHASH_THRESHOLD = 16
 SEGMENT_MATCH_THRESHOLD = PHASH_THRESHOLD  # 片段匹配阈值与帧匹配统一（#1702：阈值常量统一来源）
 MIN_CONSECUTIVE_MATCHES = 5  # 连续匹配默认门槛；短视频自适应 min(5, max(2, 分片数//2))
 MAX_GAP = 2  # 允许的最大间隙帧数
@@ -372,9 +377,10 @@ def find_duplicate_segments(
     1. 构建 query×target 全量汉明距离矩阵；每个 query chunk 保留所有
        距离 <= match_threshold 的候选 target 分片（与帧匹配判定同一阈值）。
     2. 时序一致贪心对齐：沿 query 时序推进，run 内优先选择与上一匹配帧
-       目标序号连贯（0 <= delta <= neighbor_window+1，允许 ±1 邻接窗口 /
-       时序偏移对齐，缓解场景切割导致的切点、取帧错位）的候选；同距时
-       偏好大索引，避免重复 hash 塌缩到 target 首帧。
+       目标序号连贯（|delta| <= neighbor_window+1，允许 ±1 邻接/时序偏移
+       对齐——1s 密集采样下相邻帧 pHash 接近，最近邻在目标相邻帧间
+       正/反向跳变均属正常，缓解场景切割切点、取帧错位、局部倒退）的
+       候选；同距时偏好小索引（最早对齐位置）。
     3. 连贯匹配中允许 <= max_gap 帧间隙桥接；断裂后另起新 run——天然
        支持局部片段复用（复用片段可出现在任意时序位置，各成独立片段）。
     4. 连续匹配帧数 >= min_consecutive 的 run 报为重复片段。短视频自适应：
@@ -387,7 +393,7 @@ def find_duplicate_segments(
         match_threshold: 汉明距离匹配阈值（统一常量 PHASH_THRESHOLD）
         min_consecutive: 最少连续匹配帧数；None 时按短视频自适应
         max_gap: 允许的最大间隙帧数
-        neighbor_window: 时序对齐允许的目标分片序号邻接窗口
+        neighbor_window: 时序对齐允许的目标分片序号邻接窗口（正/反向均允许）
 
     Returns:
         DuplicateSegment 列表
@@ -422,8 +428,9 @@ def find_duplicate_segments(
         min_consecutive = min(MIN_CONSECUTIVE_MATCHES, max(2, n // 2))
 
     # Step 2: 时序一致贪心对齐。
-    # run 内偏好与上一匹配帧目标序号连贯（0 <= delta <= neighbor_window+1，
-    # 支持 ±1 邻接窗口/时序偏移对齐）的候选；无连贯候选时关闭旧 run。
+    # run 内偏好与上一匹配帧目标序号连贯（|delta| <= neighbor_window+1，
+    # 支持 ±1 邻接窗口/时序偏移对齐，正反向抖动均允许）的候选；
+    # 无连贯候选时关闭旧 run。
     # 这天然支持局部片段复用：同一 query 视频中多个复用片段各自形成独立 run。
     frame_matches: list[tuple[bool, int, int]] = []
     runs: list[tuple[int, int]] = []
@@ -444,7 +451,7 @@ def find_duplicate_segments(
             chosen = cand[0] if cand else None
         else:
             chosen = next(
-                (c for c in cand if 0 <= c[0] - run_last_t <= neighbor_window + 1),
+                (c for c in cand if abs(c[0] - run_last_t) <= neighbor_window + 1),
                 None,
             )
 

tests/unit/test_bad_fingerprint_filter.py

@@ -287,7 +287,7 @@ class TestComputeDuplicateRateBadFingerprint:
 
         videos = [
             _make_existing_video("vid-b1", "md5_b1", ["aaaaaaaaaaaaaaaa"] * 10),
-            _make_existing_video("vid-b2", "md5_b2", ["bbbbbbbbbbbbbbbb"] * 5),
+            _make_existing_video("vid-b2", "md5_b2", ["cccccccccccccccc"] * 5),  # hamming(a,c)=32 > PHASH_THRESHOLD
         ]
         mock_repo = MagicMock()
         mock_repo.list_by_user.return_value = videos

tests/unit/test_dedup_1702_zero_rate_fix.py

@@ -131,7 +131,7 @@ class TestSameSourceDifferentCrop:
     def test_same_source_distance_at_threshold_still_detected(self):
         """距离正好等于阈值（<=）也要算匹配——阈值比较统一为 <=。"""
 
-        assert PHASH_THRESHOLD <= 12, "阈值应经校准保持在能检出同源裁剪的范围"
+        assert PHASH_THRESHOLD <= 16, "阈值应经真实数据校准保持在能检出同源裁剪/降重对的范围（#1702 二次校准为 16）"
         ddp = VideoDeduplicator()
         base = [_h(0) for _ in range(6)]
         new = [_h(PHASH_THRESHOLD) for _ in range(6)]
@@ -430,3 +430,119 @@ class TestCheckDuplicateExcludesSelf:
             MockRepo.return_value.list_by_project.return_value = [self_video, real_dup]
             result = ddp.check_duplicate(fp, "proj1", session, exclude_video_id="v-self")
         assert result is not None and result["duplicate_of"] == "v-real"
+
+
+# ── 阈值 16 二次校准 + 时序抖动对齐（#1702 第二轮真实数据校准） ──────
+
+
+class TestThreshold16Calibration:
+    """二次校准：staging 15 个真实成片实测——同源降重对中位数距离 14、
+    <=16 命中 8/11=0.73；异源 13 个候选每帧全局最近邻最小距离 18、<=16
+    命中全 0。阈值 16 检出同源且异源零误报（>=2bit 安全裕度）。"""
+
+    def test_threshold_calibrated_to_16(self):
+        assert PHASH_THRESHOLD == 16
+
+    @staticmethod
+    def _variant(phash: str, d: int) -> str:
+        """在 phash 基础上翻转恰好 d 个低位 bit → 与原哈希汉明距离恰为 d。"""
+        v = int(phash, 16)
+        for b in range(d):
+            v ^= 1 << b
+        return f"{v:016x}"
+
+    def test_distance_18_unrelated_not_matched(self):
+        """距离 18（异源实测最小最近邻距离）不判匹配，距离 16 判匹配。"""
+        ddp = VideoDeduplicator()
+        # 多样化 base（相邻帧各不相同，避免黑屏过滤器）
+        base = [_h(i + 4) for i in range(8)]
+        near = [self._variant(h, 16) for h in base]  # 同源降重：每帧距离恰 16
+        far = [self._variant(h, 18) for h in base]  # 异源边界：每帧距离恰 18
+
+        fp_near = _fingerprint(near, 8.0, chunks=[_chunk(h, i, i + 1) for i, h in enumerate(near)])
+        fp_far = _fingerprint(far, 8.0, chunks=[_chunk(h, i, i + 1) for i, h in enumerate(far)])
+
+        r_near = _rate(ddp, fp_near, [_video("v-base", base, duration=8.0)])
+        r_far = _rate(ddp, fp_far, [_video("v-base", base, duration=8.0)])
+
+        assert r_near["duplicate_rate"] > 0, "距离16的同源降重对必须检出"
+        assert r_far["duplicate_rate"] == 0.0, "距离18的异源对不得误报"
+        assert r_far["match_count"] == 0
+
+    def test_deduped_pair_frame_match_rate_over_threshold(self):
+        """真实场景比例：11 帧中 8 帧距离 <=16（0.73 >= 0.7），
+        其余 3 帧异源距离（>=18）——frame_match_rate 必须过 0.7 门槛。"""
+        ddp = VideoDeduplicator()
+        base = [_h(i + 4) for i in range(11)]
+        near = [self._variant(h, 14) for h in base[:8]]  # 中位数 14 的同源降重帧
+        # 异源帧用完全不同前缀（与 base 距离 >=30）
+        far = [_h(52 + i) for i in range(3)]
+        query = near + far
+
+        fp = _fingerprint(query, 11.0, chunks=[_chunk(h, i, i + 1) for i, h in enumerate(query)])
+        r = _rate(ddp, fp, [_video("v-base", base, duration=11.0)])
+        # frame_match_rate=8/11=0.73、时序片段覆盖 ~0.73
+        # → duplicate_rate = 0.4*0.73+0.6*0.73 ≈ 73%（空直方图回退下 fusion=0.6965
+        # 略低于 is_duplicate 的 0.70 判定阈值，故此处断言查重率而非 match_count；
+        # 真实视频带颜色直方图时 fusion≈0.80，staging A-C 实测 is_duplicate=True）
+        assert r["duplicate_rate"] >= 70.0
+
+
+class TestTemporalJitterAlignment:
+    """时序对齐允许目标索引正/反向 ±(neighbor_window+1) 抖动。
+
+    密集 1s 采样下相邻帧 pHash 接近，全局最近邻会在目标相邻帧间
+    正负 1 跳变（场景切割/取帧错位/局部倒退）；旧逻辑只允许正向
+    delta，把同源连续匹配拆碎，min_consecutive 门槛够不上而漏检。
+    """
+
+    def test_backward_jitter_keeps_run_continuous(self):
+        """匹配目标索引序列 0,1,2,1,2,3（含一次 -1 倒退）应保持同一 run。"""
+        from video_processing.dedup import find_duplicate_segments
+
+        # 构造 target 相邻帧 pHash 相同（距离0），query 帧的最近邻在
+        # target[1]/target[2] 之间抖动；全部 <= 阈值
+        t_hash = _h(0)
+        other = _h(40)
+        # target: 帧0-3 相同场景，帧4+ 异源
+        t_chunks = [_chunk(t_hash, i, i + 1) for i in range(4)] + [_chunk(other, i, i + 1) for i in range(4, 8)]
+        # query 6 帧同场景（最近邻会落到 target 0~3，索引可正可负）
+        q_chunks = [_chunk(t_hash, i, i + 1) for i in range(6)]
+
+        segments = find_duplicate_segments(q_chunks, t_chunks)
+        assert segments, "含 ±1 时序抖动的连续匹配必须形成片段"
+        # 6 帧匹配 >= min_consecutive(min(5,max(2,6//2))=5)，报为一个片段
+        assert len(segments) == 1
+        seg = segments[0]
+        assert seg.query_end_ms - seg.query_start_ms >= 5000
+
+    def test_large_backward_jump_breaks_run(self):
+        """目标索引倒退 > neighbor_window+1（如从 5 跳回 0）不属于抖动，
+        不桥接为同一片段；孤立短匹配 < min_consecutive 不报片段。"""
+        from video_processing.dedup import find_duplicate_segments
+
+        # 异源段：9-bit 不重叠段（相邻段隔 3 bit），跨段距离 18~24 > 阈值 16
+        def _bit_seg(start):
+            bits = ["0"] * 64
+            for b in range(9):
+                bits[start + b] = "1"
+            return f"{int(''.join(bits), 2):016x}"
+
+        t_hash = _bit_seg(0)  # 复用场景：bit 0-8
+        t_other = [_bit_seg(22 + 4 * i) for i in range(4)]  # target 异源段
+        q_other = [_bit_seg(40 + 4 * i) for i in range(3)]  # query 异源段
+        # target: 帧0 同场景；帧1-4 异源；帧5-6 同场景
+        t_chunks = (
+            [_chunk(t_hash, 0, 1)]
+            + [_chunk(t_other[i - 1], i, i + 1) for i in range(1, 5)]
+            + [_chunk(t_hash, i, i + 1) for i in range(5, 7)]
+        )
+        # query: 帧0 匹配 target[0]；帧1-3 异源（与 target 任何帧距离 >16）；帧4-5 匹配 target[5,6]
+        q_chunks = (
+            [_chunk(t_hash, 0, 1)]
+            + [_chunk(q_other[i - 1], i, i + 1) for i in range(1, 4)]
+            + [_chunk(t_hash, i, i + 1) for i in range(4, 6)]
+        )
+        segments = find_duplicate_segments(q_chunks, t_chunks)
+        # 两段各 1、2 帧 < min_consecutive=5 → 不报片段（大跳跃不桥接）
+        assert segments == []

tests/unit/test_dedup_v2.py

@@ -103,6 +103,7 @@ from video_processing.dedup import (  # noqa: E402
     MIN_CONSECUTIVE_MATCHES,
     MIN_KEYFRAME_INTERVAL_SEC,
     MIN_KEYFRAMES,
+    PHASH_THRESHOLD,
     PHASH_WEIGHT,
     SCENE_CHANGE_THRESHOLD,
     SEGMENT_MATCH_THRESHOLD,
@@ -269,22 +270,17 @@ class TestFindDuplicateSegments:
         注意：使用不同的 hash 对，确保后半部分帧距离 > 阈值。
         """
         same_hash = "aaaaaaaaaaaaaaaa"
-        # 4 帧匹配，后面 6 帧各自不同（在 query 和 target 中使用不同 hash）
+        # 4 帧匹配，后面 6 帧用与匹配哈希距离 32 的不匹配哈希（> PHASH_THRESHOLD=16）
+        nomatch_hash = "cccccccccccccccc"  # hamming(aaaa, cccc)=32
         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)
+            _make_chunk(i * 1000, (i + 1) * 1000, nomatch_hash) 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)
+            _make_chunk(i * 1000, (i + 1) * 1000, nomatch_hash) 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...
-        # Actually: hamming_distance("bbbbbbbbbbbbbbbb", "cccccccccccccccc")
-        # b=0xb=1011, c=0xc=1100 → XOR=0111=0x7 → 3 bits per digit × 16 = 48
-        # That's > 8 so won't match
-
+        # hamming(aaaa..., cccc...) = 32 > PHASH_THRESHOLD(16)，后半段不匹配；
+        # 前 4 帧匹配 < min_consecutive=5，不形成片段
         segments = find_duplicate_segments(chunks_a, chunks_b)
-        # 只有 4 帧匹配（< min_consecutive=5），所以不报告
         assert segments == []
 
     def test_max_gap_behavior(self):
@@ -293,10 +289,12 @@ class TestFindDuplicateSegments:
         关键：间隙帧必须在 query 和 target 中使用不同 hash，使其真正不匹配。
         """
         match_hash = "aaaaaaaaaaaaaaaa"
-        gap_hash_a = "bbbbbbbbbbbbbbbb"  # query 端
-        gap_hash_b = "cccccccccccccccc"  # target 端（与 query 端距离 > 8）
-        tail_hash_a = "dddddddddddddddd"
-        tail_hash_b = "eeeeeeeeeeeeeeee"
+        # 间隙/尾部哈希与 match_hash 及彼此之间汉明距离均 >64 (> PHASH_THRESHOLD=16)，
+        # 确保在 ±(neighbor_window+1) 时序抖动对齐窗口内也不会误匹配
+        gap_hash_a = "ffffffffffffffff"  # hamming(a,f)=128
+        gap_hash_b = "9999999999999999"  # hamming(a,9)=128, hamming(f,9)=128
+        tail_hash_a = "7777777777777777"  # hamming(a,7)=192
+        tail_hash_b = "1111111111111111"  # hamming(a,1)=192, hamming(7,1)=128
 
         # 5 帧匹配, 1 帧间隙, 3 帧匹配, 5 帧不匹配
         hashes_a = [match_hash] * 5 + [gap_hash_a] + [match_hash] * 3 + [tail_hash_a] * 5
@@ -318,10 +316,10 @@ class TestFindDuplicateSegments:
     def test_max_gap_exceeded(self):
         """间隙超过 max_gap → 分成两段."""
         match_hash = "aaaaaaaaaaaaaaaa"
-        gap_hash_a = "bbbbbbbbbbbbbbbb"
-        gap_hash_b = "cccccccccccccccc"
-        tail_hash_a = "dddddddddddddddd"
-        tail_hash_b = "eeeeeeeeeeeeeeee"
+        gap_hash_a = "ffffffffffffffff"  # hamming(a,f)=128
+        gap_hash_b = "9999999999999999"  # hamming(a,9)=128
+        tail_hash_a = "7777777777777777"  # hamming(a,7)=192
+        tail_hash_b = "1111111111111111"  # hamming(a,1)=192
 
         # 5 帧匹配, 3 帧间隙 (> max_gap=2), 5 帧匹配, 5 帧不匹配
         hashes_a = [match_hash] * 5 + [gap_hash_a] * 3 + [match_hash] * 5 + [tail_hash_a] * 5
@@ -497,9 +495,11 @@ class TestConstants:
     """常量值验证 — 使用已在模块顶部导入的常量，避免重新 import."""
 
     def test_segment_match_threshold(self):
-        # Issue #1702: pHash 阈值经 staging 真实同源/异源指纹回归校准
-        # （同源密集采样 min=8、异源 min=24），统一为模块常量 PHASH_THRESHOLD=12。
-        assert SEGMENT_MATCH_THRESHOLD == 12
+        # Issue #1702 二次校准：阈值经 staging 真实数据两轮回归——
+        # 第一轮同源 4/11、异源 min=24 定 12；第二轮扩样本（15 个真实成片）
+        # 同源降重对中位数距离 14、<=16 命中 8/11=0.73，异源 13 个候选
+        # <=16 命中全 0、最近邻最小距离 18 → 校准为 16。
+        assert SEGMENT_MATCH_THRESHOLD == PHASH_THRESHOLD == 16
 
     def test_min_consecutive_matches(self):
         assert MIN_CONSECUTIVE_MATCHES == 5

tests/unit/test_phash_threshold_calibration_1658.py

@@ -129,15 +129,17 @@ _ZERO_HIST = [0.0] * 96  # 全黑视频的全零直方图（有效数据）
 
 
 class TestThresholdCalibration:
-    """pHash 阈值校准（Issue #1658 收紧到 8，Issue #1702 经真实指纹分布重校准为 12）。
+    """pHash 阈值校准（#1658 收紧到 8，#1702 两轮真实数据重校准 12→16）。
 
-    #1702 staging 离线实验：同帧两次 2-5% 随机裁剪距离 4~10；同源成片（密集 1s
-    采样）最小距离 8、<=12 命中 10/31；异源成片最小距离 24。8 会漏检同源裁剪，
-    12 检出同源且与异源分布（>=24）间隔充足。
+    #1702 第一轮 staging 离线实验：同帧两次 2-5% 随机裁剪距离 4~10；同源成片
+    （密集 1s 采样）<=12 命中 4/11、异源成片最小距离 24 → 初定 12。
+    #1702 第二轮（证据视频 B->A 仍漏检）扩样本到该用户 15 个真实成片实测：
+    同源降重对中位数距离 14、<=16 命中 8/11=0.73；异源 13 个候选 <=16 命中
+    全 0、每帧全局最近邻最小距离 18 → 校准为 16（与异源仍有 >=2bit 裕度）。
     """
 
     def test_phash_threshold_is_calibrated(self):
-        assert VideoDeduplicator.PHASH_THRESHOLD == PHASH_THRESHOLD == 12
+        assert VideoDeduplicator.PHASH_THRESHOLD == PHASH_THRESHOLD == 16
 
     def test_match_ratio_threshold_constant(self):
         assert MATCH_RATIO_THRESHOLD == 0.7
@@ -152,18 +154,19 @@ class TestThresholdCalibration:
     def test_threshold_matching_semantics(self):
         """阈值比较统一为 <=（帧匹配与片段匹配同一口径）。
 
-        场景：5 个关键帧距离为 [10, 12, 12, 24, 26]。
-        - <=12（#1702 校准阈值）：3 帧匹配 → 0.6 < 0.7 被帧比例门槛拦截异源
-        - 距离 12 的同源裁剪帧应算匹配（< 与 <= 口径统一）
+        场景：5 个关键帧距离为 [10, 14, 16, 18, 26]。
+        - <=16（#1702 二次校准阈值）：3 帧匹配 → 0.6 < 0.7，被帧比例门槛
+          拦截（异源安全边界：真实数据异源最近邻最小距离 18，<=16 命中 0）
+        - 距离正好 16 的同源降重帧应算匹配（< 与 <= 口径统一）
         """
-        distances = [10, 12, 12, 24, 26]
+        distances = [10, 14, 16, 18, 26]
         matched = sum(1 for d in distances if d <= VideoDeduplicator.PHASH_THRESHOLD)
         assert matched == 3
         assert matched / len(distances) == 0.6
         assert matched / len(distances) < MATCH_RATIO_THRESHOLD
 
-        # 异源典型距离（>=24）绝不匹配
-        assert not any(d <= VideoDeduplicator.PHASH_THRESHOLD for d in (24, 26, 30))
+        # 异源安全边界（实测最小距离 18）及以上绝不匹配
+        assert not any(d <= VideoDeduplicator.PHASH_THRESHOLD for d in (18, 24, 26, 30))
 
 
 # ── TestComputeFusionScore：统一融合得分方法 ────────────────────