alembic/versions/063_add_video_fingerprint_chunks.py

@@ -0,0 +1,46 @@
+"""add video_fingerprint_chunks table for per-chunk fingerprint storage
+
+Revision ID: 063_fingerprint_chunks
+Revises: 062_edit_plan_id
+Create Date: 2026-09-03
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "063_fingerprint_chunks"
+down_revision = "062_edit_plan_id"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "video_fingerprint_chunks",
+        sa.Column("id", sa.String(36), primary_key=True),
+        sa.Column("video_id", sa.String(36), nullable=False),
+        sa.Column("project_id", sa.String(36), nullable=False),
+        sa.Column("user_id", sa.String(36), nullable=False, server_default=""),
+        sa.Column("start_time_ms", sa.Integer, nullable=False),
+        sa.Column("end_time_ms", sa.Integer, nullable=False),
+        sa.Column("phash_binary", sa.String(16), nullable=False),
+        sa.Column("color_histogram", sa.JSON, nullable=False),
+        sa.Column("frame_count", sa.Integer, nullable=False, server_default="1"),
+        sa.Column(
+            "created_at",
+            sa.DateTime,
+            nullable=False,
+            server_default=sa.func.now(),
+        ),
+    )
+    op.create_index("ix_vfc_video_id", "video_fingerprint_chunks", ["video_id"])
+    op.create_index("ix_vfc_project_id", "video_fingerprint_chunks", ["project_id"])
+    op.create_index("ix_vfc_user_id", "video_fingerprint_chunks", ["user_id"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_vfc_user_id", table_name="video_fingerprint_chunks")
+    op.drop_index("ix_vfc_project_id", table_name="video_fingerprint_chunks")
+    op.drop_index("ix_vfc_video_id", table_name="video_fingerprint_chunks")
+    op.drop_table("video_fingerprint_chunks")

apps/api/scripts/rebuild_fingerprint_chunks.py

@@ -0,0 +1,174 @@
+#!/usr/bin/env python3
+"""存量指纹重建脚本 — 为已有视频生成 video_fingerprint_chunks 分片数据。
+
+功能：
+- 查询 generated_videos 中 video_fingerprint IS NOT NULL 但尚无分片数据的视频
+- 从 OSS 下载视频 → 用新的分片算法重新计算指纹 → 写入分片表
+- 支持 --dry-run（只打印不写入）和 --batch-size（默认 50）
+- 幂等：已存在分片数据的视频跳过
+
+用法：
+    # 预览（不写入）
+    python rebuild_fingerprint_chunks.py --dry-run
+
+    # 执行重建
+    python rebuild_fingerprint_chunks.py --batch-size 50
+"""
+
+from __future__ import annotations
+
+import argparse
+import logging
+import os
+import sys
+import tempfile
+
+# 确保可以 import worker_app 和 packages
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "worker"))
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", ".."))
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+)
+logger = logging.getLogger("rebuild_fingerprint_chunks")
+
+
+def find_videos_needing_rebuild(session, batch_size: int) -> list[dict]:
+    """查询需要重建分片指纹的视频。"""
+    from sqlalchemy import and_
+
+    from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel, VideoFingerprintChunkModel
+
+    # 有 video_fingerprint 的视频
+    has_fingerprint = GeneratedVideoModel.video_fingerprint.isnot(None)
+    has_fingerprint = and_(has_fingerprint, GeneratedVideoModel.video_fingerprint != "")
+
+    # 排除已有分片数据的视频
+    subq = session.query(VideoFingerprintChunkModel.video_id).distinct().subquery()
+    no_chunks = ~GeneratedVideoModel.id.in_(subq)
+
+    videos = (
+        session.query(GeneratedVideoModel)
+        .filter(and_(has_fingerprint, no_chunks))
+        .order_by(GeneratedVideoModel.generated_at.desc())
+        .limit(batch_size)
+        .all()
+    )
+
+    return [
+        {
+            "id": v.id,
+            "project_id": v.project_id,
+            "user_id": v.user_id or "",
+            "duration": v.duration,
+        }
+        for v in videos
+    ]
+
+
+def rebuild_one(video_info: dict, dry_run: bool = False) -> int:
+    """重建单个视频的分片数据。返回写入的 chunk 数量。"""
+    from video_processing.dedup import VideoDeduplicator, _save_fingerprint_chunks
+    from worker_app.db import SessionLocal
+
+    from packages.adapters.sqlalchemy_impl.models import VideoFingerprintChunkModel
+    from packages.shared.storage import get_storage_service
+
+    video_id = video_info["id"]
+    project_id = video_info["project_id"]
+    user_id = video_info["user_id"]
+
+    if dry_run:
+        logger.info("[DRY-RUN] Would rebuild video %s (project=%s)", video_id, project_id)
+        return 0
+
+    session = SessionLocal()
+    temp_dir = tempfile.mkdtemp()
+
+    try:
+        # 再次检查幂等性
+        existing_count = (
+            session.query(VideoFingerprintChunkModel).filter(VideoFingerprintChunkModel.video_id == video_id).count()
+        )
+        if existing_count > 0:
+            logger.info("Video %s already has %d chunks, skipping", video_id, existing_count)
+            return 0
+
+        # 下载视频
+        storage_service = get_storage_service()
+        local_path = os.path.join(temp_dir, f"{video_id}.mp4")
+        storage_key = f"projects/{project_id}/generated/{video_id}/{video_id}.mp4"
+        storage_service.download_file(storage_key, local_path)
+
+        # 重新计算指纹
+        deduplicator = VideoDeduplicator()
+        fingerprint = deduplicator.compute_fingerprint(local_path)
+
+        # 写入分片表
+        _save_fingerprint_chunks(fingerprint, video_id, project_id, user_id, session)
+        session.commit()
+
+        chunk_count = len(fingerprint.chunks)
+        logger.info("Rebuilt %d chunks for video %s", chunk_count, video_id)
+        return chunk_count
+
+    except Exception as e:
+        logger.error("Failed to rebuild video %s: %s", video_id, e)
+        session.rollback()
+        return -1
+    finally:
+        session.close()
+        import shutil
+
+        shutil.rmtree(temp_dir, ignore_errors=True)
+
+
+def main():
+    parser = argparse.ArgumentParser(description="存量指纹重建脚本")
+    parser.add_argument("--dry-run", action="store_true", help="只打印不写入")
+    parser.add_argument("--batch-size", type=int, default=50, help="每批处理数量（默认 50）")
+    parser.add_argument("--total-limit", type=int, default=0, help="总处理数量限制（0=不限制）")
+    args = parser.parse_args()
+
+    from worker_app.db import SessionLocal
+
+    session = SessionLocal()
+
+    try:
+        videos = find_videos_needing_rebuild(session, args.batch_size)
+        logger.info("Found %d videos needing rebuild", len(videos))
+
+        if args.dry_run:
+            for v in videos:
+                logger.info("[DRY-RUN] Video %s | project=%s | duration=%.1fs", v["id"], v["project_id"], v["duration"])
+            return
+
+        total_chunks = 0
+        processed = 0
+        failed = 0
+
+        for v in videos:
+            if args.total_limit > 0 and processed >= args.total_limit:
+                break
+
+            result = rebuild_one(v, dry_run=False)
+            if result < 0:
+                failed += 1
+            else:
+                total_chunks += result
+                processed += 1
+
+        logger.info(
+            "Rebuild complete: processed=%d, chunks=%d, failed=%d",
+            processed,
+            total_chunks,
+            failed,
+        )
+
+    finally:
+        session.close()
+
+
+if __name__ == "__main__":
+    main()

apps/worker/video_processing/dedup.py

@@ -4,8 +4,9 @@ import hashlib
 import logging
 import os
 import tempfile
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Optional
+from uuid import uuid4
 
 import cv2
 import numpy as np
@@ -15,10 +16,16 @@ from worker_app.celery_app import celery_app
 from worker_app.db import SessionLocal
 
 from packages.adapters.sqlalchemy_impl.generated_video_repository import SQLAlchemyGeneratedVideoRepository
+from packages.adapters.sqlalchemy_impl.models import VideoFingerprintChunkModel
 from packages.shared.storage import get_storage_service
 
 logger = logging.getLogger(__name__)
 
+# 分片策略常量
+SHORT_VIDEO_CHUNK_SEC = 2  # ≤60秒视频，每 2 秒一个分片
+LONG_VIDEO_CHUNK_SEC = 5  # >60秒视频，每 5 秒一个分片
+SHORT_VIDEO_THRESHOLD_SEC = 60
+
 
 def compute_phash(image: np.ndarray, hash_size: int = 8) -> str:
     """计算图像的感知哈希（pHash），基于 DCT（离散余弦变换）。
@@ -80,6 +87,28 @@ def compute_color_histogram(image: np.ndarray, bins: int = 32) -> list[float]:
     return hist
 
 
+def compute_chunk_interval(duration: float) -> float:
+    """根据视频时长返回分片间隔（秒）。
+
+    短视频（≤60秒）：每 2 秒一个分片
+    长视频（>60秒）：每 5 秒一个分片
+    """
+    if duration <= SHORT_VIDEO_THRESHOLD_SEC:
+        return SHORT_VIDEO_CHUNK_SEC
+    return LONG_VIDEO_CHUNK_SEC
+
+
+@dataclass
+class FingerprintChunk:
+    """单个分片指纹数据。"""
+
+    start_time_ms: int
+    end_time_ms: int
+    phash_binary: str
+    color_histogram: list[float]
+    frame_count: int = 1
+
+
 @dataclass
 class VideoFingerprint:
     """Video fingerprint containing multiple similarity metrics."""
@@ -89,6 +118,7 @@ class VideoFingerprint:
     color_histograms: list[list[float]]
     duration: float
     resolution: tuple[int, int]
+    chunks: list[FingerprintChunk] = field(default_factory=list)
 
     def to_dict(self) -> dict:
         # 注意：color_histograms 里的值可能是 np.float32（来自 cv2.normalize），
@@ -101,8 +131,37 @@ class VideoFingerprint:
             "color_histograms": native_histograms,
             "duration": float(self.duration),
             "resolution": [int(self.resolution[0]), int(self.resolution[1])],
+            "chunks": [
+                {
+                    "start_time_ms": c.start_time_ms,
+                    "end_time_ms": c.end_time_ms,
+                    "phash_binary": c.phash_binary,
+                    "color_histogram": [float(v) for v in c.color_histogram],
+                    "frame_count": c.frame_count,
+                }
+                for c in self.chunks
+            ],
         }
 
+    def to_chunk_models(self, video_id: str, project_id: str, user_id: str = "") -> list[VideoFingerprintChunkModel]:
+        """将分片数据转为 SQLAlchemy Model 列表，用于批量写入 video_fingerprint_chunks 表。"""
+        models = []
+        for chunk in self.chunks:
+            models.append(
+                VideoFingerprintChunkModel(
+                    id=uuid4().hex,
+                    video_id=video_id,
+                    project_id=project_id,
+                    user_id=user_id,
+                    start_time_ms=chunk.start_time_ms,
+                    end_time_ms=chunk.end_time_ms,
+                    phash_binary=chunk.phash_binary,
+                    color_histogram=[float(v) for v in chunk.color_histogram],
+                    frame_count=chunk.frame_count,
+                )
+            )
+        return models
+
 
 class VideoDeduplicator:
     """Video deduplication using multiple fingerprint methods."""
@@ -111,7 +170,12 @@ class VideoDeduplicator:
     HISTOGRAM_THRESHOLD = 0.85
 
     def compute_fingerprint(self, video_path: str) -> VideoFingerprint:
-        """Compute video fingerprint using MD5, pHash, and color histogram."""
+        """Compute video fingerprint using MD5, pHash, and color histogram.
+
+        按时间分片抽帧：短视频（≤60s）每 2s 一片，长视频每 5s 一片。
+        每片取 1 帧计算 pHash + color_histogram。
+        同时保留 keyframe_phashes/color_histograms 聚合字段（向后兼容）。
+        """
         cap = cv2.VideoCapture(video_path)
         if not cap.isOpened():
             raise RuntimeError(f"Cannot open video: {video_path}")
@@ -123,42 +187,82 @@ class VideoDeduplicator:
         height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
 
         md5_hash = hashlib.md5(usedforsecurity=False)
-        keyframe_phashes = []
-        color_histograms = []
+        chunks: list[FingerprintChunk] = []
 
-        frame_interval = max(1, frame_count // 10)
-        for i in range(0, frame_count, frame_interval):
-            cap.set(cv2.CAP_PROP_POS_FRAMES, i)
+        # 分片间隔（秒）
+        chunk_interval_sec = compute_chunk_interval(duration)
+        chunk_interval_ms = int(chunk_interval_sec * 1000)
+        duration_ms = int(duration * 1000)
+
+        # 遍历每个分片时间窗口，取 1 帧
+        start_ms = 0
+        while start_ms < duration_ms:
+            end_ms = min(start_ms + chunk_interval_ms, duration_ms)
+            # 定位到分片中点
+            seek_ms = (start_ms + end_ms) / 2
+            cap.set(cv2.CAP_PROP_POS_MSEC, seek_ms)
             ret, frame = cap.read()
-            if not ret:
-                continue
+            if ret:
+                # MD5 计算
+                _, buffer = cv2.imencode(".jpg", frame)
+                md5_hash.update(buffer)
 
-            _, buffer = cv2.imencode(".jpg", frame)
-            md5_hash.update(buffer)
+                phash = compute_phash(frame)
+                hist = compute_color_histogram(frame)
 
-            keyframe_phashes.append(compute_phash(frame))
-            color_histograms.append(compute_color_histogram(frame))
+                chunks.append(
+                    FingerprintChunk(
+                        start_time_ms=start_ms,
+                        end_time_ms=end_ms,
+                        phash_binary=phash,
+                        color_histogram=hist,
+                        frame_count=1,
+                    )
+                )
+
+            start_ms = end_ms
 
         cap.release()
 
+        # 向后兼容：聚合 keyframe_phashes / color_histograms
+        keyframe_phashes = [c.phash_binary for c in chunks]
+        color_histograms = [c.color_histogram for c in chunks]
+
         return VideoFingerprint(
             md5=md5_hash.hexdigest(),
             keyframe_phashes=keyframe_phashes,
             color_histograms=color_histograms,
             duration=duration,
             resolution=(width, height),
+            chunks=chunks,
         )
 
+    def _get_existing_chunks(self, video_id: str, session: Session) -> list[dict]:
+        """从 video_fingerprint_chunks 表读取分片数据。返回空列表表示无分片数据。"""
+        rows = (
+            session.query(VideoFingerprintChunkModel)
+            .filter(VideoFingerprintChunkModel.video_id == video_id)
+            .order_by(VideoFingerprintChunkModel.start_time_ms)
+            .all()
+        )
+        return [
+            {
+                "phash_binary": r.phash_binary,
+                "color_histogram": r.color_histogram,
+                "start_time_ms": r.start_time_ms,
+                "end_time_ms": r.end_time_ms,
+            }
+            for r in rows
+        ]
+
     def check_duplicate(self, fingerprint: VideoFingerprint, project_id: str, session: Session) -> Optional[dict]:
         """检查视频是否与项目中已有视频重复。
 
-        判定逻辑（按优先级）：
-        1. MD5 精确匹配：完全一致则 similarity=1.0，立即返回
-        2. pHash 相似度：计算新视频每帧 phash 与已有视频每帧 phash 的最小汉明距离，
-           取所有帧的平均值 avg_distance。若 avg_distance < PHASH_THRESHOLD(10)，
-           则判定为重复，similarity = 1.0 - (avg_distance / 64)
+        查重逻辑：
+        1. MD5 精确匹配 → similarity=1.0
+        2. pHash 相似度（优先从分片表读取，回退到 JSON 字段）
 
-        注意：返回第一个通过阈值的匹配（非最优匹配）。
+        判定阈值：avg_distance < PHASH_THRESHOLD(10)
 
         Args:
             fingerprint: 待检测视频的指纹
@@ -182,8 +286,15 @@ class VideoDeduplicator:
             if fingerprint.md5 == ef.get("md5"):
                 return {"duplicate": True, "duplicate_of": existing.id, "reason": "exact_md5_match", "similarity": 1.0}
 
-            # 感知哈希相似度
-            existing_phashes = ef.get("keyframe_phashes", [])
+            # 优先从分片表读取已有视频的分片 phash
+            existing_phashes = []
+            chunk_data = self._get_existing_chunks(existing.id, session)
+            if chunk_data:
+                existing_phashes = [c["phash_binary"] for c in chunk_data]
+            else:
+                # 回退：从 JSON 字段读取（存量旧视频）
+                existing_phashes = ef.get("keyframe_phashes", [])
+
             if not existing_phashes:
                 continue
 
@@ -247,7 +358,14 @@ class VideoDeduplicator:
                     "similarity": 1.0,
                 }
 
-            existing_phashes = ef.get("keyframe_phashes", [])
+            # 优先从分片表读取
+            existing_phashes = []
+            chunk_data = self._get_existing_chunks(existing.id, session)
+            if chunk_data:
+                existing_phashes = [c["phash_binary"] for c in chunk_data]
+            else:
+                existing_phashes = ef.get("keyframe_phashes", [])
+
             if not existing_phashes:
                 continue
 
@@ -372,8 +490,14 @@ class VideoDeduplicator:
             if fingerprint.md5 == ef.get("md5"):
                 return 100.0
 
-            # pHash 相似度
-            existing_phashes = ef.get("keyframe_phashes", [])
+            # 优先从分片表读取
+            existing_phashes = []
+            chunk_data = self._get_existing_chunks(existing.id, session)
+            if chunk_data:
+                existing_phashes = [c["phash_binary"] for c in chunk_data]
+            else:
+                existing_phashes = ef.get("keyframe_phashes", [])
+
             if not existing_phashes or not fingerprint.keyframe_phashes:
                 continue
 
@@ -388,6 +512,31 @@ class VideoDeduplicator:
         return round(max(max_similarity, 0.0), 2)
 
 
+def _save_fingerprint_chunks(
+    fingerprint: VideoFingerprint,
+    video_id: str,
+    project_id: str,
+    user_id: str,
+    session: Session,
+) -> None:
+    """将指纹分片数据批量写入 video_fingerprint_chunks 表。幂等：已有数据时跳过。"""
+    # 幂等检查：已有分片数据则跳过
+    existing_count = (
+        session.query(VideoFingerprintChunkModel).filter(VideoFingerprintChunkModel.video_id == video_id).count()
+    )
+    if existing_count > 0:
+        logger.debug("Fingerprint chunks already exist for video %s (%d chunks), skipping", video_id, existing_count)
+        return
+
+    if not fingerprint.chunks:
+        logger.warning("No chunks in fingerprint for video %s, skipping chunk save", video_id)
+        return
+
+    chunk_models = fingerprint.to_chunk_models(video_id, project_id, user_id)
+    session.bulk_save_objects(chunk_models)
+    logger.info("Saved %d fingerprint chunks for video %s", len(chunk_models), video_id)
+
+
 @celery_app.task(bind=True, max_retries=3, name="worker.check_duplicate")
 def check_duplicate_task(self: Task, generated_video_id: str) -> dict:
     """Celery task to check if generated video is a duplicate."""
@@ -421,6 +570,10 @@ def check_duplicate_task(self: Task, generated_video_id: str) -> dict:
             video.duplicate_of = None
 
         video_repo.update(video)
+
+        # 写入分片表
+        _save_fingerprint_chunks(fingerprint, generated_video_id, video.project_id, video.user_id, session)
+
         session.commit()
 
         logger.info(f"Duplicate check completed for video {generated_video_id}: is_duplicate={video.is_duplicate}")

apps/worker/video_processing/dedup_helpers.py

@@ -100,6 +100,14 @@ def create_video_record_and_dedup(
 
         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)
+
         # (a) 历史成片查重
         duplicate_result = deduplicator.check_duplicate(fingerprint, project_id, session)
 

packages/adapters/sqlalchemy_impl/models.py

@@ -620,3 +620,20 @@ class CoverTemplateModel(Base):
     config = Column(JSON, nullable=False, default=dict)
     created_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc))
     updated_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc))
+
+
+class VideoFingerprintChunkModel(Base):
+    """分片视频指纹 — 每个视频按时间分片存储 pHash + color_histogram."""
+
+    __tablename__ = "video_fingerprint_chunks"
+
+    id = Column(String(36), primary_key=True)
+    video_id = Column(String(36), nullable=False, index=True)
+    project_id = Column(String(36), nullable=False, index=True)
+    user_id = Column(String(36), nullable=False, index=True, default="")
+    start_time_ms = Column(Integer, nullable=False)
+    end_time_ms = Column(Integer, nullable=False)
+    phash_binary = Column(String(16), nullable=False)
+    color_histogram = Column(JSON, nullable=False)
+    frame_count = Column(Integer, nullable=False, default=1)
+    created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))

tests/unit/test_fingerprint_chunks.py

@@ -0,0 +1,313 @@
+"""分片指纹存储单元测试 — Issue #1657.
+
+覆盖：
+- 分片策略：60秒视频 → 30片，120秒视频 → 24片
+- VideoFingerprint.to_chunk_models() 输出正确
+- _save_fingerprint_chunks 幂等性（已有数据跳过）
+- to_dict() 向后兼容
+"""
+
+from __future__ import annotations
+
+import sys
+from unittest.mock import MagicMock
+
+
+def _mock_module(**attrs):
+    """Create a mock module with __spec__ to avoid AttributeError."""
+    m = MagicMock()
+    m.__spec__ = None
+    for k, v in attrs.items():
+        setattr(m, k, v)
+    return m
+
+
+# ── Module-level setup: mock deps, import dedup, then restore sys.modules ──
+_SAVED_MODULES_KEYS = set(sys.modules.keys())
+_SAVED_MODULES_VALUES = {
+    k: sys.modules.get(k)
+    for k in [
+        "cv2",
+        "celery",
+        "sqlalchemy",
+        "sqlalchemy.orm",
+        "sqlalchemy.engine",
+        "sqlalchemy.ext",
+        "sqlalchemy.ext.declarative",
+        "worker_app.db",
+        "worker_app.celery_app",
+        "worker_app.core.config",
+        "packages.adapters.sqlalchemy_impl.session",
+        "packages.adapters.sqlalchemy_impl.generated_video_repository",
+        "packages.adapters.sqlalchemy_impl.models",
+        "packages.shared.config",
+        "packages.shared.storage",
+    ]
+}
+
+# Set up mocks
+sys.modules["cv2"] = _mock_module()
+
+_mock_celery = MagicMock()
+_mock_celery.Task = MagicMock
+_mock_celery.Celery = MagicMock
+_mock_celery.__spec__ = None
+sys.modules["celery"] = _mock_celery
+
+_mock_sqla = MagicMock()
+_mock_sqla.__path__ = []
+_mock_sqla.__spec__ = None
+sys.modules["sqlalchemy"] = _mock_sqla
+
+_mock_sqla_orm = MagicMock()
+_mock_sqla_orm.__path__ = []
+_mock_sqla_orm.__spec__ = None
+_mock_sqla_orm.Session = MagicMock
+sys.modules["sqlalchemy.orm"] = _mock_sqla_orm
+sys.modules["sqlalchemy.engine"] = _mock_module()
+sys.modules["sqlalchemy.ext"] = _mock_module()
+sys.modules["sqlalchemy.ext.declarative"] = _mock_module()
+
+sys.modules["worker_app.db"] = _mock_module(SessionLocal=MagicMock())
+sys.modules["worker_app.celery_app"] = _mock_module(celery_app=MagicMock())
+sys.modules["worker_app.core.config"] = _mock_module(get_settings=MagicMock(return_value=MagicMock()))
+
+sys.modules["packages.adapters.sqlalchemy_impl.session"] = _mock_module(
+    Base=MagicMock(),
+    build_engine=MagicMock(),
+    build_session_factory=MagicMock(),
+    ensure_database_exists=MagicMock(),
+    initialize_database=MagicMock(),
+)
+sys.modules["packages.adapters.sqlalchemy_impl.generated_video_repository"] = _mock_module()
+
+
+# Mock VideoFingerprintChunkModel with class-level column attributes
+class _FakeChunkModel:
+    video_id = MagicMock()
+    project_id = MagicMock()
+    user_id = MagicMock()
+    start_time_ms = MagicMock()
+    end_time_ms = MagicMock()
+    phash_binary = MagicMock()
+    color_histogram = MagicMock()
+    frame_count = MagicMock()
+    created_at = MagicMock()
+
+    def __init__(self, **kwargs):
+        for k, v in kwargs.items():
+            setattr(self, k, v)
+
+
+sys.modules["packages.adapters.sqlalchemy_impl.models"] = _mock_module(
+    VideoFingerprintChunkModel=_FakeChunkModel,
+)
+sys.modules["packages.shared.config"] = _mock_module(get_shared_settings=MagicMock(return_value=MagicMock()))
+sys.modules["packages.shared.storage"] = _mock_module()
+
+# Import dedup while mocks are active
+from video_processing.dedup import (  # noqa: E402
+    FingerprintChunk,
+    VideoFingerprint,
+    _save_fingerprint_chunks,
+    compute_chunk_interval,
+)
+
+# ── Restore sys.modules immediately after import ──
+for _key in list(sys.modules.keys()):
+    if _key not in _SAVED_MODULES_KEYS:
+        del sys.modules[_key]
+for _key, _value in _SAVED_MODULES_VALUES.items():
+    if _value is not None:
+        sys.modules[_key] = _value
+    elif _key in sys.modules:
+        del sys.modules[_key]
+del _SAVED_MODULES_KEYS, _SAVED_MODULES_VALUES, _key, _value
+
+
+class TestChunkInterval:
+    """测试分片间隔策略。"""
+
+    def test_short_video_interval(self):
+        """短视频（≤60秒）每 2 秒一个分片。"""
+        assert compute_chunk_interval(0) == 2
+        assert compute_chunk_interval(30) == 2
+        assert compute_chunk_interval(60) == 2
+
+    def test_long_video_interval(self):
+        """长视频（>60秒）每 5 秒一个分片。"""
+        assert compute_chunk_interval(61) == 5
+        assert compute_chunk_interval(120) == 5
+        assert compute_chunk_interval(300) == 5
+
+    def test_chunk_count_60s_video(self):
+        """60秒视频 → 30 片（60/2=30）。"""
+        duration = 60
+        interval = compute_chunk_interval(duration)
+        expected_chunks = int(duration / interval)
+        assert expected_chunks == 30
+
+    def test_chunk_count_120s_video(self):
+        """120秒视频 → 24 片（120/5=24）。"""
+        duration = 120
+        interval = compute_chunk_interval(duration)
+        expected_chunks = int(duration / interval)
+        assert expected_chunks == 24
+
+
+class TestVideoFingerprintToChunkModels:
+    """测试 VideoFingerprint.to_chunk_models() 输出。"""
+
+    def test_to_chunk_models_output(self):
+        """to_chunk_models 返回正确的 Model 列表。"""
+        fp = VideoFingerprint(
+            md5="abc123",
+            keyframe_phashes=["a1b2", "c3d4"],
+            color_histograms=[[0.1] * 96, [0.2] * 96],
+            duration=10.0,
+            resolution=(1920, 1080),
+            chunks=[
+                FingerprintChunk(start_time_ms=0, end_time_ms=2000, phash_binary="a1b2", color_histogram=[0.1] * 96),
+                FingerprintChunk(start_time_ms=2000, end_time_ms=4000, phash_binary="c3d4", color_histogram=[0.2] * 96),
+            ],
+        )
+
+        models = fp.to_chunk_models(video_id="v1", project_id="p1", user_id="u1")
+
+        assert len(models) == 2
+        assert models[0].video_id == "v1"
+        assert models[0].project_id == "p1"
+        assert models[0].user_id == "u1"
+        assert models[0].start_time_ms == 0
+        assert models[0].end_time_ms == 2000
+        assert models[0].phash_binary == "a1b2"
+        assert models[1].start_time_ms == 2000
+        assert models[1].end_time_ms == 4000
+        assert models[1].phash_binary == "c3d4"
+
+    def test_to_chunk_models_empty_chunks(self):
+        """空 chunks 列表返回空 Model 列表。"""
+        fp = VideoFingerprint(
+            md5="abc",
+            keyframe_phashes=[],
+            color_histograms=[],
+            duration=0,
+            resolution=(0, 0),
+            chunks=[],
+        )
+
+        models = fp.to_chunk_models(video_id="v1", project_id="p1")
+        assert models == []
+
+
+class TestSaveFingerprintChunksIdempotent:
+    """测试 _save_fingerprint_chunks 幂等性。"""
+
+    def test_save_skips_existing(self):
+        """已有分片数据时跳过写入。"""
+        fp = VideoFingerprint(
+            md5="abc",
+            keyframe_phashes=["a1b2"],
+            color_histograms=[[0.1] * 96],
+            duration=5.0,
+            resolution=(1920, 1080),
+            chunks=[
+                FingerprintChunk(start_time_ms=0, end_time_ms=2000, phash_binary="a1b2", color_histogram=[0.1] * 96),
+            ],
+        )
+
+        session = MagicMock()
+        # Mock: 已有 1 条分片数据
+        session.query.return_value.filter.return_value.count.return_value = 1
+
+        _save_fingerprint_chunks(fp, video_id="v1", project_id="p1", user_id="u1", session=session)
+
+        # bulk_save_objects 不应被调用
+        session.bulk_save_objects.assert_not_called()
+
+    def test_save_writes_new(self):
+        """无分片数据时写入。"""
+        fp = VideoFingerprint(
+            md5="abc",
+            keyframe_phashes=["a1b2"],
+            color_histograms=[[0.1] * 96],
+            duration=5.0,
+            resolution=(1920, 1080),
+            chunks=[
+                FingerprintChunk(start_time_ms=0, end_time_ms=2000, phash_binary="a1b2", color_histogram=[0.1] * 96),
+            ],
+        )
+
+        session = MagicMock()
+        # Mock: 无分片数据
+        session.query.return_value.filter.return_value.count.return_value = 0
+
+        _save_fingerprint_chunks(fp, video_id="v1", project_id="p1", user_id="u1", session=session)
+
+        # bulk_save_objects 应被调用一次
+        session.bulk_save_objects.assert_called_once()
+        saved_models = session.bulk_save_objects.call_args[0][0]
+        assert len(saved_models) == 1
+        assert saved_models[0].video_id == "v1"
+        assert saved_models[0].phash_binary == "a1b2"
+
+    def test_save_skips_no_chunks(self):
+        """指纹无 chunks 时跳过。"""
+        fp = VideoFingerprint(
+            md5="abc",
+            keyframe_phashes=[],
+            color_histograms=[],
+            duration=0,
+            resolution=(0, 0),
+            chunks=[],
+        )
+
+        session = MagicMock()
+        session.query.return_value.filter.return_value.count.return_value = 0
+
+        _save_fingerprint_chunks(fp, video_id="v1", project_id="p1", user_id="u1", session=session)
+
+        # bulk_save_objects 不应被调用
+        session.bulk_save_objects.assert_not_called()
+
+
+class TestFingerprintToDictBackwardCompat:
+    """测试 to_dict() 向后兼容性。"""
+
+    def test_to_dict_includes_chunks(self):
+        """to_dict() 包含 chunks 字段。"""
+        fp = VideoFingerprint(
+            md5="abc123",
+            keyframe_phashes=["a1b2"],
+            color_histograms=[[0.1] * 96],
+            duration=5.0,
+            resolution=(1920, 1080),
+            chunks=[
+                FingerprintChunk(start_time_ms=0, end_time_ms=2000, phash_binary="a1b2", color_histogram=[0.1] * 96),
+            ],
+        )
+
+        d = fp.to_dict()
+
+        assert "chunks" in d
+        assert len(d["chunks"]) == 1
+        assert d["chunks"][0]["start_time_ms"] == 0
+        assert d["chunks"][0]["end_time_ms"] == 2000
+        assert d["chunks"][0]["phash_binary"] == "a1b2"
+
+    def test_to_dict_preserves_legacy_fields(self):
+        """to_dict() 保留 keyframe_phashes 和 color_histograms 字段。"""
+        fp = VideoFingerprint(
+            md5="abc",
+            keyframe_phashes=["a1b2", "c3d4"],
+            color_histograms=[[0.1] * 96, [0.2] * 96],
+            duration=10.0,
+            resolution=(1920, 1080),
+        )
+
+        d = fp.to_dict()
+
+        assert "keyframe_phashes" in d
+        assert "color_histograms" in d
+        assert len(d["keyframe_phashes"]) == 2
+        assert len(d["color_histograms"]) == 2