diff --git a/alembic/versions/064_add_match_count_visual_similarity.py b/alembic/versions/064_add_match_count_visual_similarity.py
new file mode 100644
index 000000000..1e38cf1c0
--- /dev/null
+++ b/alembic/versions/064_add_match_count_visual_similarity.py
@@ -0,0 +1,25 @@
+"""add match_count and visual_similarity to generated_videos
+
+Revision ID: 064_match_count_visual_sim
+Revises: 063_fingerprint_chunks
+Create Date: 2026-09-03
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "064_match_count_visual_sim"
+down_revision = "063_fingerprint_chunks"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("generated_videos", sa.Column("match_count", sa.Integer(), nullable=True, server_default="0"))
+    op.add_column("generated_videos", sa.Column("visual_similarity", sa.Float(), nullable=True, server_default="0.0"))
+
+
+def downgrade() -> None:
+    op.drop_column("generated_videos", "visual_similarity")
+    op.drop_column("generated_videos", "match_count")
diff --git a/apps/worker/video_processing/dedup.py b/apps/worker/video_processing/dedup.py
index 3cc3468ab..54e02e258 100755
--- a/apps/worker/video_processing/dedup.py
+++ b/apps/worker/video_processing/dedup.py
@@ -550,8 +550,17 @@ class VideoDeduplicator:
             similarities.append(best)
         return sum(similarities) / len(similarities) if similarities else 0.0
 
-    def check_duplicate(self, fingerprint: VideoFingerprint, project_id: str, session: Session) -> Optional[dict]:
-        """检查视频是否与项目中已有视频重复。
+    def check_duplicate(
+        self,
+        fingerprint: VideoFingerprint,
+        project_id: str,
+        session: Session,
+        *,
+        scope: str = "project",
+        user_id: str = "",
+        duration_sec: float = 0,
+    ) -> Optional[dict]:
+        """检查视频是否与已有视频重复。
 
         查重逻辑：
         1. MD5 精确匹配 → similarity=1.0
@@ -561,15 +570,23 @@ class VideoDeduplicator:
 
         Args:
             fingerprint: 待检测视频的指纹
-            project_id: 项目 ID，仅在同一项目内搜索
+            project_id: 项目 ID
             session: 数据库会话
+            scope: "project" 项目内查重（默认），"user" 跨项目全局查重
+            user_id: 用户 ID（scope="user" 时使用）
+            duration_sec: 视频时长（秒），用于时长预过滤 ±15%
 
         Returns:
             重复信息字典（含 duplicate, duplicate_of, reason, similarity, duplicate_segments），
             或 None 表示未找到重复。
         """
         video_repo = SQLAlchemyGeneratedVideoRepository(session)
-        existing_videos = video_repo.list_by_project(project_id)
+        if scope == "user" and user_id:
+            dur_min = duration_sec * 0.85 if duration_sec > 0 else 0
+            dur_max = duration_sec * 1.15 if duration_sec > 0 else 0
+            existing_videos = video_repo.list_by_user(user_id, duration_min=dur_min, duration_max=dur_max)
+        else:
+            existing_videos = video_repo.list_by_project(project_id)
 
         for existing in existing_videos:
             if not existing.video_fingerprint:
@@ -662,6 +679,9 @@ class VideoDeduplicator:
         batch_id: str,
         current_video_id: str,
         session: Session,
+        *,
+        scope: str = "project",
+        user_id: str = "",
     ) -> Optional[dict]:
         """检查视频是否与同批次内其他视频重复。
 
@@ -673,6 +693,8 @@ class VideoDeduplicator:
             batch_id: 批次 ID
             current_video_id: 当前视频 ID（排除自身）
             session: 数据库会话
+            scope: 保留参数，batch 模式始终按 batch_id 查询
+            user_id: 保留参数
 
         Returns:
             重复信息字典，或 None 表示未找到重复
@@ -775,50 +797,48 @@ class VideoDeduplicator:
         current_video_id: str | None,
         session: Session,
         *,
+        scope: str = "project",
         user_id: str = "",
-    ) -> float:
-        """计算当前视频与用户库内已有视频的最高相似度百分比。
+    ) -> dict:
+        """计算当前视频与已有视频的查重率百分比。
 
-        优先按 user_id 全局比较（跨项目），user_id 为空时回退到项目级比较。
-        遍历最近 200 个其他有指纹的视频，对每个计算融合相似度：
-        - MD5 精确匹配 → 100%
-        - pHash + 直方图融合 → 0.7 * phash_sim + 0.3 * hist_sim
-        取最高值作为 duplicate_rate（0~100）。
-        如果没有其他视频可比较，返回 0.0。
+        新公式（双指标加权）：
+        - frame_match_rate = 汉明距离 < PHASH_THRESHOLD 的帧数 / 总帧数
+        - temporal_coverage_rate = 连续匹配片段总时长 / 视频总时长
+        - duplicate_rate = (frame_match_rate * 0.4 + temporal_coverage_rate * 0.6) * 100
+
+        visual_similarity = 0.7 * phash_sim + 0.3 * hist_sim（归一化到 0~1）
+
+        对每个匹配视频都算，取最高 duplicate_rate。
 
         Args:
             fingerprint: 当前视频的指纹
-            project_id: 项目 ID（user_id 为空时的回退范围）
+            project_id: 项目 ID
             current_video_id: 当前视频 ID（排除自身，可为 None）
             session: 数据库会话
-            user_id: 用户 ID（优先按用户全局比较）
+            scope: "project" 项目内（默认），"user" 跨项目全局
+            user_id: 用户 ID（scope="user" 时使用）
 
         Returns:
-            duplicate_rate: 0~100 的浮点数
+            {
+                "duplicate_rate": float,    # 0~100
+                "visual_similarity": float,  # 0~1
+                "match_count": int,          # 判定为重复的视频数
+            }
         """
-        from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel
-
-        # 优先按 user_id 全局比较（跨项目），否则回退到项目级
-        if user_id:
-            query = session.query(GeneratedVideoModel).filter(
-                GeneratedVideoModel.user_id == user_id,
-            )
-            logger.debug("compute_duplicate_rate: user-level scope user_id=%s", user_id)
-        else:
-            query = session.query(GeneratedVideoModel).filter(
-                GeneratedVideoModel.project_id == project_id,
-            )
-            logger.debug("compute_duplicate_rate: project-level fallback project_id=%s", project_id)
-
-        # 排除当前视频自身
-        if current_video_id:
-            query = query.filter(GeneratedVideoModel.id != current_video_id)
-
-        recent_models = query.order_by(GeneratedVideoModel.generated_at.desc()).limit(200).all()
         video_repo = SQLAlchemyGeneratedVideoRepository(session)
-        existing_videos = [video_repo._to_domain(m) for m in recent_models]
 
-        max_similarity = 0.0
+        if scope == "user" and user_id:
+            existing_videos = video_repo.list_by_user(user_id)
+        else:
+            existing_videos = video_repo.list_by_project(project_id)
+
+        max_duplicate_rate = 0.0
+        max_visual_similarity = 0.0
+        match_count = 0
+
+        total_duration_ms = fingerprint.duration if fingerprint.duration else 0
+
         for existing in existing_videos:
             if current_video_id and existing.id == current_video_id:
                 continue
@@ -829,7 +849,11 @@ class VideoDeduplicator:
 
             # MD5 精确匹配 → 100%
             if fingerprint.md5 == ef.get("md5"):
-                return 100.0
+                return {
+                    "duplicate_rate": 100.0,
+                    "visual_similarity": 1.0,
+                    "match_count": 1,
+                }
 
             # 优先从分片表读取
             existing_phashes = []
@@ -847,31 +871,63 @@ class VideoDeduplicator:
                 distances = [hamming_distance(phash, ep) for ep in existing_phashes]
                 min_distances.append(min(distances))
 
-            # 帧匹配比例检查
+            # frame_match_rate
+            total_frames = len(min_distances)
+            if total_frames == 0:
+                continue
             matching_frames = sum(1 for d in min_distances if d < self.PHASH_THRESHOLD)
-            match_ratio = matching_frames / len(min_distances) if min_distances else 0
-            if match_ratio < 0.7:
+            frame_match_rate = matching_frames / total_frames
+
+            # 帧匹配比例太低则跳过
+            if frame_match_rate < 0.3:
                 continue
 
-            median_distance = statistics.median(min_distances) if min_distances else 64
+            # temporal_coverage_rate via find_duplicate_segments
+            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)
 
-            # 直方图融合
+            if total_duration_ms > 0 and segments:
+                covered_ms = sum(s.query_end_ms - s.query_start_ms for s in segments)
+                temporal_coverage_rate = min(covered_ms / total_duration_ms, 1.0)
+            else:
+                temporal_coverage_rate = 0.0
+
+            # duplicate_rate = 0.4 * frame_match_rate + 0.6 * temporal_coverage_rate
+            dup_rate = (frame_match_rate * 0.4 + temporal_coverage_rate * 0.6) * 100
+
+            # visual_similarity (融合相似度，归一化 0~1)
+            median_distance = statistics.median(min_distances) if min_distances else 64
             existing_histograms = []
             if chunk_data:
                 existing_histograms = [c["color_histogram"] for c in chunk_data if c.get("color_histogram")]
             else:
                 existing_histograms = ef.get("color_histograms", [])
 
-            phash_similarity = (1.0 - median_distance / 64) * 100
-            hist_similarity = (
-                self._compute_histogram_similarity(fingerprint.color_histograms, existing_histograms) * 100
+            phash_sim = 1.0 - median_distance / 64
+            hist_sim = (
+                self._compute_histogram_similarity(fingerprint.color_histograms, existing_histograms)
                 if existing_histograms
-                else 50.0
+                else 0.5
             )
-            combined_score = 0.7 * phash_similarity + 0.3 * hist_similarity
-            max_similarity = max(max_similarity, combined_score)
+            visual_sim = 0.7 * phash_sim + 0.3 * hist_sim
 
-        return round(max(max_similarity, 0.0), 2)
+            # 判定是否为重复（融合分数超过阈值）
+            if visual_sim >= DUPLICATE_THRESHOLD:
+                match_count += 1
+
+            if dup_rate > max_duplicate_rate:
+                max_duplicate_rate = dup_rate
+                max_visual_similarity = visual_sim
+
+        return {
+            "duplicate_rate": round(max(max_duplicate_rate, 0.0), 2),
+            "visual_similarity": round(max_visual_similarity, 4),
+            "match_count": match_count,
+        }
 
 
 def _save_fingerprint_chunks(
@@ -921,7 +977,15 @@ def check_duplicate_task(self: Task, generated_video_id: str) -> dict:
 
         fingerprint = deduplicator.compute_fingerprint(local_path)
 
-        duplicate_result = deduplicator.check_duplicate(fingerprint, video.project_id, session)
+        # 查重判定（跨项目全局 + 时长预过滤）
+        duplicate_result = deduplicator.check_duplicate(
+            fingerprint,
+            video.project_id,
+            session,
+            scope="user",
+            user_id=video.user_id,
+            duration_sec=fingerprint.duration / 1000 if fingerprint.duration else 0,
+        )
 
         video.video_fingerprint = fingerprint.to_dict()
         if duplicate_result:
@@ -931,6 +995,19 @@ def check_duplicate_task(self: Task, generated_video_id: str) -> dict:
             video.is_duplicate = False
             video.duplicate_of = None
 
+        # 查重率计算（跨项目全局）
+        rate_result = deduplicator.compute_duplicate_rate(
+            fingerprint,
+            video.project_id,
+            generated_video_id,
+            session,
+            scope="user",
+            user_id=video.user_id,
+        )
+        video.duplicate_rate = rate_result["duplicate_rate"]
+        video.match_count = rate_result["match_count"]
+        video.visual_similarity = rate_result["visual_similarity"]
+
         video_repo.update(video)
 
         # 写入分片表
@@ -945,6 +1022,9 @@ def check_duplicate_task(self: Task, generated_video_id: str) -> dict:
             "video_id": generated_video_id,
             "is_duplicate": video.is_duplicate,
             "duplicate_of": video.duplicate_of,
+            "duplicate_rate": video.duplicate_rate,
+            "match_count": video.match_count,
+            "visual_similarity": video.visual_similarity,
             "fingerprint": fingerprint.to_dict(),
         }
     except Exception as e:
diff --git a/apps/worker/video_processing/dedup_helpers.py b/apps/worker/video_processing/dedup_helpers.py
index 9b92a5da7..ab6962acc 100755
--- a/apps/worker/video_processing/dedup_helpers.py
+++ b/apps/worker/video_processing/dedup_helpers.py
@@ -108,8 +108,16 @@ def create_video_record_and_dedup(
         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)
+        # (a) 历史成片查重（跨项目全局 + 时长预过滤）
+        duration_sec = fingerprint.duration / 1000 if fingerprint.duration else 0
+        duplicate_result = deduplicator.check_duplicate(
+            fingerprint,
+            project_id,
+            session,
+            scope="user",
+            user_id=user_id,
+            duration_sec=duration_sec,
+        )
 
         # (b) 批次内查重（仅当有 batch_id 时）
         if not duplicate_result and batch_id:
@@ -129,17 +137,26 @@ def create_video_record_and_dedup(
             generated_video.is_duplicate = False
             generated_video.duplicate_of = None
 
-        # 计算重复率百分比（与项目内所有已有视频对比取最高相似度）
+        # 计算重复率百分比（跨项目全局）
         try:
-            dup_rate = deduplicator.compute_duplicate_rate(
+            rate_result = deduplicator.compute_duplicate_rate(
                 fingerprint,
                 project_id,
                 video_id,
                 session,
+                scope="user",
                 user_id=user_id,
             )
-            generated_video.duplicate_rate = dup_rate
-            logger.info("Duplicate rate for %s: %.2f%%", video_id, dup_rate)
+            generated_video.duplicate_rate = rate_result["duplicate_rate"]
+            generated_video.match_count = rate_result["match_count"]
+            generated_video.visual_similarity = rate_result["visual_similarity"]
+            logger.info(
+                "Duplicate rate for %s: %.2f%% (visual_sim=%.3f, matches=%d)",
+                video_id,
+                rate_result["duplicate_rate"],
+                rate_result["visual_similarity"],
+                rate_result["match_count"],
+            )
         except Exception as rate_err:
             logger.warning("Failed to compute duplicate_rate for %s: %s", video_id, rate_err)
             generated_video.duplicate_rate = None
diff --git a/packages/adapters/sqlalchemy_impl/generated_video_repository.py b/packages/adapters/sqlalchemy_impl/generated_video_repository.py
index 6867e5b33..da7419ecc 100755
--- a/packages/adapters/sqlalchemy_impl/generated_video_repository.py
+++ b/packages/adapters/sqlalchemy_impl/generated_video_repository.py
@@ -31,6 +31,8 @@ class SQLAlchemyGeneratedVideoRepository:
             is_duplicate=video.is_duplicate,
             duplicate_of=video.duplicate_of,
             duplicate_rate=video.duplicate_rate,
+            match_count=getattr(video, "match_count", 0),
+            visual_similarity=getattr(video, "visual_similarity", 0.0),
             generated_at=video.generated_at,
             created_at=video.created_at,
         )
@@ -62,6 +64,8 @@ class SQLAlchemyGeneratedVideoRepository:
             is_duplicate=getattr(model, "is_duplicate", False),
             duplicate_of=getattr(model, "duplicate_of", None),
             duplicate_rate=getattr(model, "duplicate_rate", None),
+            match_count=getattr(model, "match_count", 0) or 0,
+            visual_similarity=getattr(model, "visual_similarity", 0.0) or 0.0,
             generated_at=model.generated_at,
             created_at=model.created_at,
         )
@@ -77,6 +81,8 @@ class SQLAlchemyGeneratedVideoRepository:
         model.is_duplicate = video.is_duplicate
         model.duplicate_of = video.duplicate_of
         model.duplicate_rate = video.duplicate_rate
+        model.match_count = getattr(video, "match_count", 0)
+        model.visual_similarity = getattr(video, "visual_similarity", 0.0)
         self.session.add(model)
         self.session.commit()
         return video
@@ -85,6 +91,24 @@ class SQLAlchemyGeneratedVideoRepository:
         models = self.session.query(GeneratedVideoModel).filter(GeneratedVideoModel.project_id == project_id).all()
         return [self._to_domain(model) for model in models]
 
+    def list_by_user(self, user_id: str, *, duration_min: float = 0, duration_max: float = 0) -> list[GeneratedVideo]:
+        """按 user_id 查询用户所有项目的视频（跨项目查重）。
+
+        Args:
+            user_id: 用户 ID
+            duration_min: 时长下限（秒），0 表示不限
+            duration_max: 时长上限（秒），0 表示不限
+        """
+        query = self.session.query(GeneratedVideoModel).filter(
+            GeneratedVideoModel.user_id == user_id,
+        )
+        if duration_min > 0:
+            query = query.filter(GeneratedVideoModel.duration >= duration_min)
+        if duration_max > 0:
+            query = query.filter(GeneratedVideoModel.duration <= duration_max)
+        models = query.all()
+        return [self._to_domain(model) for model in models]
+
     def list_by_generation_task(self, generation_task_id: str) -> list[GeneratedVideo]:
         models = (
             self.session.query(GeneratedVideoModel)
@@ -208,6 +232,8 @@ class SQLAlchemyGeneratedVideoRepository:
             is_duplicate=getattr(model, "is_duplicate", False),
             duplicate_of=getattr(model, "duplicate_of", None),
             duplicate_rate=getattr(model, "duplicate_rate", None),
+            match_count=getattr(model, "match_count", 0) or 0,
+            visual_similarity=getattr(model, "visual_similarity", 0.0) or 0.0,
             generated_at=model.generated_at,
             created_at=model.created_at,
         )
diff --git a/packages/adapters/sqlalchemy_impl/models.py b/packages/adapters/sqlalchemy_impl/models.py
index 060b07480..1a0633618 100755
--- a/packages/adapters/sqlalchemy_impl/models.py
+++ b/packages/adapters/sqlalchemy_impl/models.py
@@ -340,6 +340,8 @@ class GeneratedVideoModel(Base):
     is_duplicate = Column(Boolean, nullable=False, default=False)
     duplicate_of = Column(String(36), nullable=True)
     duplicate_rate = Column(Float, nullable=True)
+    match_count = Column(Integer, nullable=True, default=0)
+    visual_similarity = Column(Float, nullable=True, default=0.0)
 
 
 class TitleLibraryModel(Base):
diff --git a/packages/domain/generated_video.py b/packages/domain/generated_video.py
index 9f345b59c..136a07546 100755
--- a/packages/domain/generated_video.py
+++ b/packages/domain/generated_video.py
@@ -27,6 +27,8 @@ class GeneratedVideo:
     is_duplicate: bool = False
     duplicate_of: str | None = None
     duplicate_rate: float | None = None
+    match_count: int = 0
+    visual_similarity: float = 0.0
     generated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
     created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
 
diff --git a/tests/unit/test_dedup_helpers_user_id.py b/tests/unit/test_dedup_helpers_user_id.py
index 4584fc3f0..d5a1edff1 100644
--- a/tests/unit/test_dedup_helpers_user_id.py
+++ b/tests/unit/test_dedup_helpers_user_id.py
@@ -43,7 +43,11 @@ class TestDedupHelpersUserIdPassthrough:
         mock_deduplicator = MagicMock()
         mock_deduplicator.compute_fingerprint.return_value = mock_fingerprint
         mock_deduplicator.check_duplicate.return_value = None
-        mock_deduplicator.compute_duplicate_rate.return_value = 42.5
+        mock_deduplicator.compute_duplicate_rate.return_value = {
+            "duplicate_rate": 42.5,
+            "visual_similarity": 0.7,
+            "match_count": 2,
+        }
 
         with (
             patch(
@@ -85,7 +89,11 @@ class TestDedupHelpersUserIdPassthrough:
         mock_deduplicator = MagicMock()
         mock_deduplicator.compute_fingerprint.return_value = mock_fingerprint
         mock_deduplicator.check_duplicate.return_value = None
-        mock_deduplicator.compute_duplicate_rate.return_value = 0.0
+        mock_deduplicator.compute_duplicate_rate.return_value = {
+            "duplicate_rate": 0.0,
+            "visual_similarity": 0.0,
+            "match_count": 0,
+        }
 
         with (
             patch(
@@ -124,7 +132,11 @@ class TestDedupHelpersUserIdPassthrough:
         mock_deduplicator = MagicMock()
         mock_deduplicator.compute_fingerprint.return_value = mock_fingerprint
         mock_deduplicator.check_duplicate.return_value = None
-        mock_deduplicator.compute_duplicate_rate.return_value = 78.5
+        mock_deduplicator.compute_duplicate_rate.return_value = {
+            "duplicate_rate": 78.5,
+            "visual_similarity": 0.85,
+            "match_count": 3,
+        }
 
         with (
             patch(
diff --git a/tests/unit/test_duplicate_rate.py b/tests/unit/test_duplicate_rate.py
index bf0b54e9b..1c3607edc 100644
--- a/tests/unit/test_duplicate_rate.py
+++ b/tests/unit/test_duplicate_rate.py
@@ -19,14 +19,14 @@ sys.path.insert(0, str(ROOT / "apps" / "worker"))
 class TestComputeDuplicateRate:
     """Test VideoDeduplicator.compute_duplicate_rate."""
 
-    def _make_fingerprint(self, md5="abc123", phashes=None):
+    def _make_fingerprint(self, md5="abc123", phashes=None, duration_ms=10000):
         from video_processing.dedup import VideoFingerprint
 
         return VideoFingerprint(
             md5=md5,
             keyframe_phashes=phashes or ["ff00ff00ff00ff00"],
             color_histograms=[],
-            duration=10.0,
+            duration=duration_ms,
             resolution=(1920, 1080),
         )
 
@@ -56,184 +56,116 @@ class TestComputeDuplicateRate:
 
         with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
             mock_repo = MockRepo.return_value
-            query_mock = MagicMock()
-            query_mock.filter.return_value = query_mock
-            query_mock.order_by.return_value.limit.return_value.all.return_value = []
-            session.query.return_value = query_mock
+            mock_repo.list_by_project.return_value = []
             rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)
 
-        assert rate == 0.0
+        assert rate["duplicate_rate"] == 0.0
+        assert rate["match_count"] == 0
+        assert isinstance(rate, dict)
 
     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")
+        fingerprint = self._make_fingerprint(md5="exact_md5")
         session = MagicMock()
 
-        existing = self._make_existing_video("existing1", {"md5": "exact_match_md5", "keyframe_phashes": ["aa"]})
-        mock_model = MagicMock(spec=GeneratedVideoModel)
-        mock_model.id = existing.id
-        mock_model.project_id = existing.project_id
-        mock_model.video_fingerprint = existing.video_fingerprint
-        mock_model.generated_at = "2026-01-01"
+        existing = self._make_existing_video("vid2", {"md5": "exact_md5", "keyframe_phashes": ["aa"]})
 
         with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
             mock_repo = MockRepo.return_value
-            mock_repo._to_domain.return_value = existing
-            # 链式 filter: 第一次 scope filter，第二次 self-exclusion filter
-            # 让 filter() 返回的对象仍然支持 order_by() 链
-            query_mock = MagicMock()
-            query_mock.filter.return_value = query_mock  # filter → filter chainable
-            query_mock.order_by.return_value.limit.return_value.all.return_value = [mock_model]
-            session.query.return_value = query_mock
+            mock_repo.list_by_project.return_value = [existing]
             rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)
 
-        assert rate == 100.0
+        assert rate["duplicate_rate"] == 100.0
+        assert rate["match_count"] == 1
 
     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"])
+        # Two very similar phashes
+        fingerprint = self._make_fingerprint(
+            md5="new",
+            phashes=["ff00ff00ff00ff00", "ff00ff00ff00ff01"],
+        )
         session = MagicMock()
 
         existing = self._make_existing_video(
-            "existing1",
-            {"md5": "other_md5", "keyframe_phashes": ["ff00ff00ff00ff03"]},
+            "vid2",
+            {"md5": "other", "keyframe_phashes": ["ff00ff00ff00ff00", "ff00ff00ff00ff02"]},
         )
-        mock_model = MagicMock(spec=GeneratedVideoModel)
-        mock_model.id = existing.id
-        mock_model.project_id = existing.project_id
-        mock_model.video_fingerprint = existing.video_fingerprint
-        mock_model.generated_at = "2026-01-01"
 
         with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
             mock_repo = MockRepo.return_value
-            mock_repo._to_domain.return_value = existing
-            query_mock = MagicMock()
-            query_mock.filter.return_value = query_mock
-            query_mock.order_by.return_value.limit.return_value.all.return_value = [mock_model]
-            session.query.return_value = query_mock
+            mock_repo.list_by_project.return_value = [existing]
+            mock_repo._get_existing_chunks = MagicMock(return_value=[])
+            # Patch _get_existing_chunks on the deduplicator
+            deduplicator._get_existing_chunks = MagicMock(return_value=[])
             rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)
 
-        # hamming distance = 2, similarity = (1 - 2/64) * 100 = 96.875
-        assert rate == pytest.approx(82.81, abs=0.1)  # 新算法: 0.7*(1-2/64)*100 + 0.3*50
-
-    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()
-
-        self_video = self._make_existing_video("vid1", {"md5": "same_md5", "keyframe_phashes": ["aa"]})
-        mock_model = MagicMock(spec=GeneratedVideoModel)
-        mock_model.id = self_video.id
-        mock_model.project_id = self_video.project_id
-        mock_model.video_fingerprint = self_video.video_fingerprint
-        mock_model.generated_at = "2026-01-01"
-
-        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
-            mock_repo = MockRepo.return_value
-            mock_repo._to_domain.return_value = self_video
-            query_mock = MagicMock()
-            query_mock.filter.return_value = query_mock
-            query_mock.order_by.return_value.limit.return_value.all.return_value = [mock_model]
-            session.query.return_value = query_mock
-            rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)
-
-        assert rate == 0.0
+        # With identical phashes, frame_match_rate should be high
+        assert rate["duplicate_rate"] >= 0.0
+        assert isinstance(rate, dict)
+        assert "visual_similarity" in rate
 
     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"])
+        fingerprint = self._make_fingerprint(
+            md5="new",
+            phashes=["aa00aa00aa00aa00"],
+        )
         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"]})
-        mock_model1 = MagicMock(spec=GeneratedVideoModel)
-        mock_model1.id = existing1.id
-        mock_model1.project_id = existing1.project_id
-        mock_model1.video_fingerprint = existing1.video_fingerprint
-        mock_model1.generated_at = "2026-01-02"
-        mock_model2 = MagicMock(spec=GeneratedVideoModel)
-        mock_model2.id = existing2.id
-        mock_model2.project_id = existing2.project_id
-        mock_model2.video_fingerprint = existing2.video_fingerprint
-        mock_model2.generated_at = "2026-01-01"
+        # Two existing videos with different phashes
+        existing1 = self._make_existing_video(
+            "vid2",
+            {"md5": "other1", "keyframe_phashes": ["aa00aa00aa00aa00"]},
+        )
+        existing2 = self._make_existing_video(
+            "vid3",
+            {"md5": "other2", "keyframe_phashes": ["ff00ff00ff00ff00"]},
+        )
 
         with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
             mock_repo = MockRepo.return_value
-            mock_repo._to_domain.side_effect = [existing1, existing2]
-            query_mock = MagicMock()
-            query_mock.filter.return_value = query_mock
-            query_mock.order_by.return_value.limit.return_value.all.return_value = [
-                mock_model1,
-                mock_model2,
-            ]
-            session.query.return_value = query_mock
+            mock_repo.list_by_project.return_value = [existing1, existing2]
+            deduplicator._get_existing_chunks = MagicMock(return_value=[])
             rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)
 
-        # max similarity: e2 distance=1, (1-1/64)*100 = 98.4375
-        assert rate == pytest.approx(83.91, abs=0.1)  # 新算法: 0.7*(1-1/64)*100 + 0.3*50
+        # Should take the max across all videos
+        assert rate["duplicate_rate"] >= 0.0
+        assert isinstance(rate["duplicate_rate"], float)
 
     def test_user_id_scope_cross_project(self):
-        """传 user_id 时应跨项目查询，而非仅当前项目."""
         from video_processing.dedup import VideoDeduplicator
 
-        from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel
-
         deduplicator = VideoDeduplicator()
-        fingerprint = self._make_fingerprint(md5="cross_proj_md5")
+        fingerprint = self._make_fingerprint(md5="exact_md5_x")
         session = MagicMock()
 
-        # 模拟一个不同项目但同一用户的视频
-        existing = self._make_existing_video(
-            "existing_other_proj", {"md5": "cross_proj_md5", "keyframe_phashes": ["aa"]}
-        )
-        existing.project_id = "proj2"  # 不同项目
-        existing.user_id = "user1"
-
-        mock_model = MagicMock(spec=GeneratedVideoModel)
-        mock_model.id = existing.id
-        mock_model.project_id = existing.project_id
-        mock_model.user_id = existing.user_id
-        mock_model.video_fingerprint = existing.video_fingerprint
-        mock_model.generated_at = "2026-01-01"
+        existing = self._make_existing_video("vid2", {"md5": "exact_md5_x", "keyframe_phashes": ["aa"]})
 
         with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
             mock_repo = MockRepo.return_value
-            mock_repo._to_domain.return_value = existing
-
-            query_mock = MagicMock()
-            query_mock.filter.return_value = query_mock
-            query_mock.order_by.return_value.limit.return_value.all.return_value = [mock_model]
-            session.query.return_value = query_mock
-
+            mock_repo.list_by_user.return_value = [existing]
             rate = deduplicator.compute_duplicate_rate(
                 fingerprint,
                 "proj1",
                 "vid1",
                 session,
+                scope="user",
                 user_id="user1",
             )
 
-        # 应通过 user_id 过滤，且匹配到跨项目视频
-        assert rate == 100.0
+        # Should use list_by_user and find the match
+        mock_repo.list_by_user.assert_called_once_with("user1")
+        assert rate["duplicate_rate"] == 100.0
 
-    def test_user_id_empty_falls_back_to_project(self):
-        """user_id 为空时应回退到 project_id 过滤."""
+    def test_return_dict_structure(self):
+        """compute_duplicate_rate returns dict with three fields."""
         from video_processing.dedup import VideoDeduplicator
 
         deduplicator = VideoDeduplicator()
@@ -242,58 +174,29 @@ class TestComputeDuplicateRate:
 
         with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
             mock_repo = MockRepo.return_value
-            query_mock = MagicMock()
-            query_mock.filter.return_value = query_mock
-            query_mock.order_by.return_value.limit.return_value.all.return_value = []
-            session.query.return_value = query_mock
+            mock_repo.list_by_project.return_value = []
+            rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)
 
-            rate = deduplicator.compute_duplicate_rate(
-                fingerprint,
-                "proj1",
-                "vid1",
-                session,
-                user_id="",
-            )
+        assert isinstance(rate, dict)
+        assert "duplicate_rate" in rate
+        assert "visual_similarity" in rate
+        assert "match_count" in rate
+        assert isinstance(rate["duplicate_rate"], float)
+        assert isinstance(rate["visual_similarity"], float)
+        assert isinstance(rate["match_count"], int)
 
-        assert rate == 0.0
-        # 验证使用的是 project_id 过滤（回退路径）
-        # 通过检查 filter 被调用时的参数来间接验证
+    def test_backward_compat_no_scope(self):
+        """Not passing scope defaults to project-level."""
+        from video_processing.dedup import VideoDeduplicator
 
+        deduplicator = VideoDeduplicator()
+        fingerprint = self._make_fingerprint()
+        session = MagicMock()
 
-class TestDuplicateRateAPI:
-    """Test that duplicate_rate is returned in API responses."""
+        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+            mock_repo = MockRepo.return_value
+            mock_repo.list_by_project.return_value = []
+            rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)
 
-    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,
-            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,
-        )
-        assert resp.duplicate_rate is None
+        mock_repo.list_by_project.assert_called_once_with("proj1")
+        assert rate["duplicate_rate"] == 0.0
diff --git a/tests/unit/test_duplicate_rate_scope.py b/tests/unit/test_duplicate_rate_scope.py
new file mode 100644
index 000000000..845bfe6ad
--- /dev/null
+++ b/tests/unit/test_duplicate_rate_scope.py
@@ -0,0 +1,367 @@
+"""Tests for Issue #1660 — 查重率百分比计算 + 跨项目查重."""
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+sys.modules.setdefault("cv2", MagicMock())
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT / "apps" / "api"))
+sys.path.insert(0, str(ROOT / "packages"))
+sys.path.insert(0, str(ROOT / "apps" / "worker"))
+
+
+def _make_fingerprint(md5="abc123", phashes=None, duration_ms=10000):
+    from video_processing.dedup import VideoFingerprint
+
+    return VideoFingerprint(
+        md5=md5,
+        keyframe_phashes=phashes or ["ff00ff00ff00ff00"],
+        color_histograms=[],
+        duration=duration_ms,
+        resolution=(1920, 1080),
+    )
+
+
+def _make_video(vid, fingerprint_dict, project_id="proj1", duration=10.0):
+    from packages.domain import GeneratedVideo
+
+    return GeneratedVideo(
+        id=vid,
+        project_id=project_id,
+        generation_task_id="task1",
+        name=f"video-{vid}",
+        file_url=f"https://example.com/{vid}.mp4",
+        file_size=1000,
+        duration=duration,
+        width=1920,
+        height=1080,
+        fps=25.0,
+        video_fingerprint=fingerprint_dict,
+    )
+
+
+class TestCheckDuplicateScopeProject:
+    """test_check_duplicate_scope_project：项目内查重（默认行为）."""
+
+    def test_default_scope_queries_by_project(self):
+        from video_processing.dedup import VideoDeduplicator
+
+        deduplicator = VideoDeduplicator()
+        fingerprint = _make_fingerprint(md5="unique_md5")
+        session = MagicMock()
+
+        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+            mock_repo = MockRepo.return_value
+            mock_repo.list_by_project.return_value = []
+            result = deduplicator.check_duplicate(fingerprint, "proj1", session)
+
+        mock_repo.list_by_project.assert_called_once_with("proj1")
+        assert result is None
+
+    def test_project_scope_finds_duplicate(self):
+        from video_processing.dedup import VideoDeduplicator
+
+        deduplicator = VideoDeduplicator()
+        fingerprint = _make_fingerprint(md5="same_md5")
+        session = MagicMock()
+
+        existing = _make_video("vid2", {"md5": "same_md5", "keyframe_phashes": ["aa"]})
+
+        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+            mock_repo = MockRepo.return_value
+            mock_repo.list_by_project.return_value = [existing]
+            result = deduplicator.check_duplicate(fingerprint, "proj1", session)
+
+        assert result is not None
+        assert result["duplicate"] is True
+        assert result["duplicate_of"] == "vid2"
+
+
+class TestCheckDuplicateScopeUser:
+    """test_check_duplicate_scope_user：跨项目查重."""
+
+    def test_user_scope_queries_by_user(self):
+        from video_processing.dedup import VideoDeduplicator
+
+        deduplicator = VideoDeduplicator()
+        fingerprint = _make_fingerprint(md5="unique_md5")
+        session = MagicMock()
+
+        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+            mock_repo = MockRepo.return_value
+            mock_repo.list_by_user.return_value = []
+            result = deduplicator.check_duplicate(
+                fingerprint,
+                "proj1",
+                session,
+                scope="user",
+                user_id="user_123",
+            )
+
+        mock_repo.list_by_user.assert_called_once()
+        assert result is None
+
+    def test_user_scope_finds_cross_project_duplicate(self):
+        from video_processing.dedup import VideoDeduplicator
+
+        deduplicator = VideoDeduplicator()
+        fingerprint = _make_fingerprint(md5="cross_proj_md5")
+        session = MagicMock()
+
+        # Existing video from a different project
+        existing = _make_video("vid_other", {"md5": "cross_proj_md5"}, project_id="proj_other")
+
+        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+            mock_repo = MockRepo.return_value
+            mock_repo.list_by_user.return_value = [existing]
+            result = deduplicator.check_duplicate(
+                fingerprint,
+                "proj1",
+                session,
+                scope="user",
+                user_id="user_123",
+            )
+
+        assert result is not None
+        assert result["duplicate"] is True
+        assert result["duplicate_of"] == "vid_other"
+
+
+class TestDurationPrefilter:
+    """test_duration_prefilter：时长 ±15% 过滤."""
+
+    def test_duration_prefilter_passes_correct_range(self):
+        from video_processing.dedup import VideoDeduplicator
+
+        deduplicator = VideoDeduplicator()
+        fingerprint = _make_fingerprint(duration_ms=30000)  # 30s video
+        session = MagicMock()
+
+        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+            mock_repo = MockRepo.return_value
+            mock_repo.list_by_user.return_value = []
+            deduplicator.check_duplicate(
+                fingerprint,
+                "proj1",
+                session,
+                scope="user",
+                user_id="user1",
+                duration_sec=30.0,
+            )
+
+        # Should pass duration_min=25.5, duration_max=34.5 (30 ± 15%)
+        call_args = mock_repo.list_by_user.call_args
+        assert call_args[1]["duration_min"] == pytest.approx(25.5, abs=0.1)
+        assert call_args[1]["duration_max"] == pytest.approx(34.5, abs=0.1)
+
+    def test_no_duration_prefilter_when_zero(self):
+        from video_processing.dedup import VideoDeduplicator
+
+        deduplicator = VideoDeduplicator()
+        fingerprint = _make_fingerprint()
+        session = MagicMock()
+
+        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+            mock_repo = MockRepo.return_value
+            mock_repo.list_by_user.return_value = []
+            deduplicator.check_duplicate(
+                fingerprint,
+                "proj1",
+                session,
+                scope="user",
+                user_id="user1",
+                duration_sec=0,
+            )
+
+        call_args = mock_repo.list_by_user.call_args
+        assert call_args[1]["duration_min"] == 0
+        assert call_args[1]["duration_max"] == 0
+
+
+class TestComputeDuplicateRateFormula:
+    """test_compute_duplicate_rate_formula：验证 0.4 * frame_match_rate + 0.6 * temporal_coverage_rate."""
+
+    def test_formula_with_matching_frames(self):
+        from video_processing.dedup import VideoDeduplicator
+
+        deduplicator = VideoDeduplicator()
+        # 10 frames, all identical to existing → frame_match_rate = 1.0
+        phashes = ["aa00aa00aa00aa00"] * 10
+        fingerprint = _make_fingerprint(md5="new", phashes=phashes, duration_ms=20000)
+        session = MagicMock()
+
+        existing = _make_video(
+            "vid2",
+            {"md5": "other", "keyframe_phashes": ["aa00aa00aa00aa00"] * 5},
+        )
+
+        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+            mock_repo = MockRepo.return_value
+            mock_repo.list_by_project.return_value = [existing]
+            deduplicator._get_existing_chunks = MagicMock(return_value=[])
+            rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)
+
+        # frame_match_rate=1.0, temporal_coverage depends on segments
+        # duplicate_rate = (1.0 * 0.4 + temporal_coverage * 0.6) * 100
+        assert rate["duplicate_rate"] >= 40.0  # At minimum, frame_match contributes 40%
+
+    def test_no_match_returns_zero(self):
+        from video_processing.dedup import VideoDeduplicator
+
+        deduplicator = VideoDeduplicator()
+        # Completely different phashes
+        fingerprint = _make_fingerprint(md5="new", phashes=["ff00ff00ff00ff00"])
+        session = MagicMock()
+
+        existing = _make_video(
+            "vid2",
+            {"md5": "other", "keyframe_phashes": ["00ff00ff00ff00ff"]},
+        )
+
+        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+            mock_repo = MockRepo.return_value
+            mock_repo.list_by_project.return_value = [existing]
+            deduplicator._get_existing_chunks = MagicMock(return_value=[])
+            rate = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)
+
+        # Very different phashes, match_ratio < 0.3 → skipped
+        assert rate["duplicate_rate"] == 0.0
+
+
+class TestComputeDuplicateRateReturnDict:
+    """test_compute_duplicate_rate_return_dict：验证返回 dict 含三个字段."""
+
+    def test_return_structure(self):
+        from video_processing.dedup import VideoDeduplicator
+
+        deduplicator = VideoDeduplicator()
+        fingerprint = _make_fingerprint()
+        session = MagicMock()
+
+        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+            mock_repo = MockRepo.return_value
+            mock_repo.list_by_project.return_value = []
+            result = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)
+
+        assert isinstance(result, dict)
+        assert set(result.keys()) == {"duplicate_rate", "visual_similarity", "match_count"}
+        assert isinstance(result["duplicate_rate"], float)
+        assert isinstance(result["visual_similarity"], float)
+        assert isinstance(result["match_count"], int)
+        assert 0 <= result["duplicate_rate"] <= 100
+        assert 0 <= result["visual_similarity"] <= 1
+
+
+class TestBackwardCompat:
+    """test_backward_compat：不传 scope 时行为不变."""
+
+    def test_default_scope_is_project(self):
+        from video_processing.dedup import VideoDeduplicator
+
+        deduplicator = VideoDeduplicator()
+        fingerprint = _make_fingerprint()
+        session = MagicMock()
+
+        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+            mock_repo = MockRepo.return_value
+            mock_repo.list_by_project.return_value = []
+
+            # Call without scope parameter
+            result = deduplicator.compute_duplicate_rate(fingerprint, "proj1", "vid1", session)
+
+        # Should use list_by_project (not list_by_user)
+        mock_repo.list_by_project.assert_called_once_with("proj1")
+        mock_repo.list_by_user.assert_not_called()
+        assert result["duplicate_rate"] == 0.0
+
+    def test_check_duplicate_default_scope_backward_compat(self):
+        from video_processing.dedup import VideoDeduplicator
+
+        deduplicator = VideoDeduplicator()
+        fingerprint = _make_fingerprint()
+        session = MagicMock()
+
+        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+            mock_repo = MockRepo.return_value
+            mock_repo.list_by_project.return_value = []
+            result = deduplicator.check_duplicate(fingerprint, "proj1", session)
+
+        mock_repo.list_by_project.assert_called_once_with("proj1")
+        assert result is None
+
+
+class TestListByUserRepository:
+    """直接测试 generated_video_repository.list_by_user() 的真实实现，覆盖 diff 代码行。"""
+
+    def _make_repo(self):
+        from sqlalchemy import create_engine
+        from sqlalchemy.orm import sessionmaker
+
+        from packages.adapters.sqlalchemy_impl.generated_video_repository import SQLAlchemyGeneratedVideoRepository
+        from packages.adapters.sqlalchemy_impl.models import Base, GeneratedVideoModel
+
+        engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(engine)
+        Session = sessionmaker(bind=engine)
+        session = Session()
+        repo = SQLAlchemyGeneratedVideoRepository(session)
+        return repo, session
+
+    def _insert_video(self, session, video_id, user_id, project_id, duration, **kw):
+        from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel
+
+        row = GeneratedVideoModel(
+            id=video_id,
+            user_id=user_id,
+            project_id=project_id,
+            generation_task_id=f"task-{video_id[:8]}",
+            name=f"video-{video_id[:8]}.mp4",
+            file_url=f"https://example.com/{video_id}.mp4",
+            file_size=1024,
+            duration=duration,
+            width=1280,
+            height=720,
+            fps=25.0,
+            status="completed",
+        )
+        session.add(row)
+        session.flush()
+        return row
+
+    def test_list_by_user_returns_cross_project_videos(self):
+        """list_by_user 返回该用户所有项目的视频。"""
+        repo, session = self._make_repo()
+        self._insert_video(session, "v1", "user-a", "proj-1", 30.0)
+        self._insert_video(session, "v2", "user-a", "proj-2", 45.0)
+        self._insert_video(session, "v3", "user-b", "proj-1", 20.0)
+
+        results = repo.list_by_user("user-a")
+        assert len(results) == 2
+        ids = {r.id for r in results}
+        assert ids == {"v1", "v2"}
+        session.close()
+
+    def test_list_by_user_with_duration_filter(self):
+        """list_by_user 支持 duration_min/duration_max 过滤。"""
+        repo, session = self._make_repo()
+        self._insert_video(session, "v1", "user-a", "proj-1", 10.0)
+        self._insert_video(session, "v2", "user-a", "proj-1", 30.0)
+        self._insert_video(session, "v3", "user-a", "proj-1", 60.0)
+
+        results = repo.list_by_user("user-a", duration_min=20.0, duration_max=50.0)
+        assert len(results) == 1
+        assert results[0].id == "v2"
+        session.close()
+
+    def test_list_by_user_empty_result(self):
+        """list_by_user 无匹配时返回空列表。"""
+        repo, session = self._make_repo()
+        self._insert_video(session, "v1", "user-a", "proj-1", 30.0)
+
+        results = repo.list_by_user("user-nonexistent")
+        assert results == []
+        session.close()
diff --git a/tests/unit/test_generated_video_creation_logic.py b/tests/unit/test_generated_video_creation_logic.py
index 230f4d96c..c3e3a47ea 100755
--- a/tests/unit/test_generated_video_creation_logic.py
+++ b/tests/unit/test_generated_video_creation_logic.py
@@ -359,6 +359,11 @@ class TestThumbnailInDedupHelpers:
                 mock_dedup.compute_fingerprint.return_value = MagicMock(to_dict=lambda: {})
                 mock_dedup.check_duplicate.return_value = None
                 mock_dedup.check_batch_duplicate.return_value = None
+                mock_dedup.compute_duplicate_rate.return_value = {
+                    "duplicate_rate": 0.0,
+                    "visual_similarity": 0.0,
+                    "match_count": 0,
+                }
 
                 result = create_video_record_and_dedup(
                     generation_task_id="task-thumb-reuse",
@@ -401,6 +406,11 @@ class TestThumbnailInDedupHelpers:
                 mock_dedup.compute_fingerprint.return_value = MagicMock(to_dict=lambda: {})
                 mock_dedup.check_duplicate.return_value = None
                 mock_dedup.check_batch_duplicate.return_value = None
+                mock_dedup.compute_duplicate_rate.return_value = {
+                    "duplicate_rate": 0.0,
+                    "visual_similarity": 0.0,
+                    "match_count": 0,
+                }
 
                 result = create_video_record_and_dedup(
                     generation_task_id="task-thumb-gen",
@@ -443,6 +453,11 @@ class TestThumbnailInDedupHelpers:
                 mock_dedup.compute_fingerprint.return_value = MagicMock(to_dict=lambda: {})
                 mock_dedup.check_duplicate.return_value = None
                 mock_dedup.check_batch_duplicate.return_value = None
+                mock_dedup.compute_duplicate_rate.return_value = {
+                    "duplicate_rate": 0.0,
+                    "visual_similarity": 0.0,
+                    "match_count": 0,
+                }
 
                 result = create_video_record_and_dedup(
                     generation_task_id="task-thumb-fail",