fix: correct extract-voice API path from /tts to /voices prefix

Backend route is mounted under /voices, not /tts
2026-09-03 20:40:48 +08:00
7 changed files with 26 additions and 738 deletions
@@ -1,46 +0,0 @@
-"""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")
@@ -1,174 +0,0 @@
-#!/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()
@@ -136,9 +136,8 @@ export const MaterialVoiceTab: React.FC<MaterialVoiceTabProps> = ({
            const material = mapAssetToMaterial(asset)
            // duration 优先取顶层（后端从 metadata 提取），兜底 metadata
            const cardDuration = asset.duration || material.duration || 0
-            // AI 生成素材标识：兼容旧素材（无 source 字段但有 tts_job_id）
-            const meta = asset.metadata as Record<string, unknown>
-            const isAiMaterial = meta?.source === "tts_job" || !!meta?.tts_job_id
+            // AI 生成素材标识（metadata.source === "tts_job"）
+            const isAiMaterial = (asset.metadata as Record<string, unknown>)?.source === "tts_job"
            const isPlaying = playingId === asset.id
            const isSelected = selectedIds.has(asset.id)
            // 播放中以 audio 真实时长为准，未播放显示卡片时长
@@ -4,9 +4,8 @@ import hashlib
 import logging
 import os
 import tempfile
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from typing import Optional
-from uuid import uuid4

 import cv2
 import numpy as np
@@ -16,16 +15,10 @@ 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（离散余弦变换）。
@@ -87,28 +80,6 @@ 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."""
@@ -118,7 +89,6 @@ 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），
@@ -131,37 +101,8 @@ 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."""
@@ -170,12 +111,7 @@ class VideoDeduplicator:
    HISTOGRAM_THRESHOLD = 0.85

    def compute_fingerprint(self, video_path: str) -> VideoFingerprint:
-        """Compute video fingerprint using MD5, pHash, and color histogram.
-
-        按时间分片抽帧：短视频（≤60s）每 2s 一片，长视频每 5s 一片。
-        每片取 1 帧计算 pHash + color_histogram。
-        同时保留 keyframe_phashes/color_histograms 聚合字段（向后兼容）。
-        """
+        """Compute video fingerprint using MD5, pHash, and color histogram."""
        cap = cv2.VideoCapture(video_path)
        if not cap.isOpened():
            raise RuntimeError(f"Cannot open video: {video_path}")
@@ -187,82 +123,42 @@ class VideoDeduplicator:
        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))

        md5_hash = hashlib.md5(usedforsecurity=False)
-        chunks: list[FingerprintChunk] = []
+        keyframe_phashes = []
+        color_histograms = []

-        # 分片间隔（秒）
-        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)
+        frame_interval = max(1, frame_count // 10)
+        for i in range(0, frame_count, frame_interval):
+            cap.set(cv2.CAP_PROP_POS_FRAMES, i)
            ret, frame = cap.read()
-            if ret:
-                # MD5 计算
-                _, buffer = cv2.imencode(".jpg", frame)
-                md5_hash.update(buffer)
+            if not ret:
+                continue

-                phash = compute_phash(frame)
-                hist = compute_color_histogram(frame)
+            _, buffer = cv2.imencode(".jpg", frame)
+            md5_hash.update(buffer)

-                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
+            keyframe_phashes.append(compute_phash(frame))
+            color_histograms.append(compute_color_histogram(frame))

        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 相似度（优先从分片表读取，回退到 JSON 字段）
+        判定逻辑（按优先级）：
+        1. MD5 精确匹配：完全一致则 similarity=1.0，立即返回
+        2. pHash 相似度：计算新视频每帧 phash 与已有视频每帧 phash 的最小汉明距离，
+           取所有帧的平均值 avg_distance。若 avg_distance < PHASH_THRESHOLD(10)，
+           则判定为重复，similarity = 1.0 - (avg_distance / 64)

-        判定阈值：avg_distance < PHASH_THRESHOLD(10)
+        注意：返回第一个通过阈值的匹配（非最优匹配）。

        Args:
            fingerprint: 待检测视频的指纹
@@ -286,15 +182,8 @@ class VideoDeduplicator:
            if fingerprint.md5 == ef.get("md5"):
                return {"duplicate": True, "duplicate_of": existing.id, "reason": "exact_md5_match", "similarity": 1.0}

-            # 优先从分片表读取已有视频的分片 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", [])
-
+            # 感知哈希相似度
+            existing_phashes = ef.get("keyframe_phashes", [])
            if not existing_phashes:
                continue

@@ -358,14 +247,7 @@ class VideoDeduplicator:
                    "similarity": 1.0,
                }

-            # 优先从分片表读取
-            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", [])
-
+            existing_phashes = ef.get("keyframe_phashes", [])
            if not existing_phashes:
                continue

@@ -490,14 +372,8 @@ class VideoDeduplicator:
            if fingerprint.md5 == ef.get("md5"):
                return 100.0

-            # 优先从分片表读取
-            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", [])
-
+            # pHash 相似度
+            existing_phashes = ef.get("keyframe_phashes", [])
            if not existing_phashes or not fingerprint.keyframe_phashes:
                continue

@@ -512,31 +388,6 @@ 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."""
@@ -570,10 +421,6 @@ 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}")
@@ -100,14 +100,6 @@ 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)

@@ -620,20 +620,3 @@ 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))
@@ -1,313 +0,0 @@
-"""分片指纹存储单元测试 — 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