alembic/versions/065_add_visual_similarity_match_count_duplication.py

@@ -0,0 +1,25 @@
+"""add visual_similarity and match_count to duplication_records
+
+Revision ID: 065_dup_record_sim_match
+Revises: 064_match_count_visual_sim
+Create Date: 2026-09-04
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "065_dup_record_sim_match"
+down_revision = "064_match_count_visual_sim"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("duplication_records", sa.Column("visual_similarity", sa.Float(), nullable=True))
+    op.add_column("duplication_records", sa.Column("match_count", sa.Integer(), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column("duplication_records", "match_count")
+    op.drop_column("duplication_records", "visual_similarity")

apps/api/app/api/routes/duplication.py

@@ -7,6 +7,7 @@ from typing import Any
 from uuid import uuid4
 
 from app.auth import AuthenticatedUser, get_current_user
+from app.core.celery_app import celery_app
 from app.core.storage import OSSStorageService, get_storage_service
 from app.dependencies import get_duplication_repository
 from app.schemas.duplication import (
@@ -76,6 +77,8 @@ def _to_record_response(record: DuplicationRecord) -> DuplicationRecordResponse:
         status=record.status,
         duplicate_rate=record.duplicate_rate,
         duplicate_count=record.duplicate_count,
+        visual_similarity=getattr(record, "visual_similarity", None),
+        match_count=getattr(record, "match_count", None),
         created_at=record.created_at.isoformat(),
         updated_at=record.updated_at.isoformat(),
     )
@@ -90,6 +93,8 @@ def _to_detail_response(record: DuplicationRecord) -> DuplicationDetailResponse:
         status=record.status,
         duplicate_rate=record.duplicate_rate,
         duplicate_count=record.duplicate_count,
+        visual_similarity=getattr(record, "visual_similarity", None),
+        match_count=getattr(record, "match_count", None),
         created_at=record.created_at.isoformat(),
         updated_at=record.updated_at.isoformat(),
         segments=[
@@ -192,6 +197,8 @@ async def upload_for_duplication(
         authenticated_user.user.id,
     )
 
+    celery_app.send_task("worker.process_duplication_check", args=[record.id])
+
     return DuplicationUploadResponse(
         id=record.id,
         status=record.status,
@@ -296,6 +303,8 @@ def retry_duplication(
             detail=f"查重记录 {record_id} 不存在",
         )
 
+    celery_app.send_task("worker.process_duplication_check", args=[updated.id])
+
     return DuplicationUploadResponse(
         id=updated.id,
         status=updated.status,

apps/api/app/schemas/duplication.py

@@ -28,6 +28,9 @@ class DuplicationRecordResponse(BaseModel):
     status: str = "pending"
     duplicate_rate: float | None = None
     duplicate_count: int = 0
+    # #1661 视觉相似度（归一化 0~1）/ 匹配视频数
+    visual_similarity: float | None = None
+    match_count: int | None = None
     created_at: str
     updated_at: str
 

apps/worker/worker_app/celery_app.py

@@ -15,6 +15,7 @@ celery_app.conf.imports = (
     "worker_app.tasks.voice_clone",
     "worker_app.tasks.tts_synthesis",
     "worker_app.tasks.batch_download",
+    "worker_app.tasks.duplication_check",
     "worker_app.tasks._startup",
     "apps.worker.video_processing.dedup",
     "worker_app.tasks.cleanup",

apps/worker/worker_app/tasks/duplication_check.py

@@ -0,0 +1,196 @@
+"""手动查重任务（Issue #1661）。
+
+流程：
+1. 从 OSS 下载用户上传的待查重视频
+2. 动态抽帧计算指纹（复用 VideoDeduplicator.compute_fingerprint）
+3. 跨项目与用户所有已有成片比对（compute_duplicate_rate + find_duplicate_segments）
+4. 更新 DuplicationRecord：status / duplicate_rate / duplicate_count / segments
+   同时写入 visual_similarity / match_count
+5. 失败重试 3 次、间隔 60 秒，最终失败标记 failed；临时文件始终清理
+"""
+
+import logging
+import os
+import shutil
+import tempfile
+
+from celery import Task
+from celery.exceptions import Retry
+from video_processing.dedup import (
+    VideoDeduplicator,
+    find_duplicate_segments,
+)
+from worker_app.celery_app import celery_app
+from worker_app.db import SessionLocal
+
+from packages.adapters.sqlalchemy_impl.duplication_repository import (
+    SQLAlchemyDuplicationRecordRepository,
+)
+from packages.adapters.sqlalchemy_impl.generated_video_repository import (
+    SQLAlchemyGeneratedVideoRepository,
+)
+from packages.domain.duplication import DuplicateSegment
+from packages.shared.storage import get_storage_service
+
+logger = logging.getLogger(__name__)
+
+
+def _build_domain_segments(
+    fingerprint,
+    session,
+    deduplicator: VideoDeduplicator,
+    user_id: str,
+) -> tuple[list[DuplicateSegment], int]:
+    """对用户所有已有视频做分片级时序匹配，构建领域片段列表。
+
+    Returns:
+        (segments, duplicate_count) — segments 为 query 视频中的重复片段，
+        duplicate_count 为存在重复片段的匹配视频数。
+    """
+    video_repo = SQLAlchemyGeneratedVideoRepository(session)
+    existing_videos = video_repo.list_by_user(user_id)
+
+    segments_out: list[DuplicateSegment] = []
+    duplicate_count = 0
+
+    for existing in existing_videos:
+        if not existing.video_fingerprint:
+            continue
+
+        chunk_data = deduplicator._get_existing_chunks(existing.id, session)
+        if not chunk_data:
+            # 老视频无分片数据，时序定位不可靠，跳过片段级匹配
+            continue
+
+        raw_segments = find_duplicate_segments(fingerprint.chunks, chunk_data)
+        if not raw_segments:
+            continue
+
+        duplicate_count += 1
+        for raw in raw_segments:
+            avg_sim = 1.0 - raw.avg_distance / 64.0
+            segments_out.append(
+                DuplicateSegment.create(
+                    source_start=round(raw.query_start_ms / 1000.0, 2),
+                    source_end=round(raw.query_end_ms / 1000.0, 2),
+                    matched_video_id=existing.id,
+                    matched_video_name=existing.name,
+                    matched_start=round(raw.target_start_ms / 1000.0, 2),
+                    matched_end=round(raw.target_end_ms / 1000.0, 2),
+                    similarity=round(max(0.0, min(1.0, avg_sim)) * 100, 1),
+                )
+            )
+
+    # 按 query 起始时间排序，片段时间轴稳定
+    segments_out.sort(key=lambda s: (s.source_start, s.source_end))
+    return segments_out, duplicate_count
+
+
+@celery_app.task(bind=True, max_retries=3, name="worker.process_duplication_check")
+def process_duplication_check(self: Task, record_id: str) -> dict:
+    """处理一次手动查重请求。
+
+    Args:
+        record_id: DuplicationRecord ID
+
+    Returns:
+        dict: {"ok": True, "record_id": ..., "duplicate_rate": ..., ...}
+    """
+    session = None
+    temp_dir = None
+    try:
+        session = SessionLocal()
+        repo = SQLAlchemyDuplicationRecordRepository(session)
+        storage_service = get_storage_service()
+        deduplicator = VideoDeduplicator()
+
+        record = repo.get(record_id)
+        if record is None:
+            raise ValueError(f"Duplication record {record_id} not found")
+
+        if record.status not in ("pending", "processing"):
+            logger.info("Duplication record %s already %s, skip", record_id, record.status)
+            return {"ok": True, "record_id": record_id, "status": record.status, "skipped": True}
+
+        record.mark_processing()
+        repo.update(record)
+        session.commit()
+
+        temp_dir = tempfile.mkdtemp(prefix="dup_check_")
+        suffix = os.path.splitext(record.filename)[1] or ".mp4"
+        local_path = os.path.join(temp_dir, f"{record_id}{suffix}")
+
+        storage_service.download_file(record.storage_key, local_path)
+
+        fingerprint = deduplicator.compute_fingerprint(local_path)
+        record.duration_seconds = round(fingerprint.duration, 2) if fingerprint.duration else 0.0
+        record.video_fingerprint = fingerprint.to_dict()
+
+        # 跨项目与用户所有已有视频比对（current_video_id=None：上传视频不在成片表中）
+        rate_result = deduplicator.compute_duplicate_rate(
+            fingerprint,
+            project_id="",
+            current_video_id=None,
+            session=session,
+            scope="user",
+            user_id=record.user_id,
+        )
+
+        # 分片级时序匹配 → 重复片段
+        segments, segment_match_count = _build_domain_segments(fingerprint, session, deduplicator, record.user_id)
+
+        record.mark_completed(
+            duplicate_rate=rate_result["duplicate_rate"],
+            duplicate_count=segment_match_count,
+            segments=segments,
+            visual_similarity=rate_result["visual_similarity"],
+            match_count=rate_result["match_count"],
+        )
+        repo.update(record)
+        session.commit()
+
+        logger.info(
+            "Duplication check completed: record=%s rate=%.2f%% matches=%d segments=%d",
+            record_id,
+            record.duplicate_rate,
+            record.match_count,
+            len(segments),
+        )
+
+        return {
+            "ok": True,
+            "record_id": record_id,
+            "status": "completed",
+            "duplicate_rate": record.duplicate_rate,
+            "duplicate_count": record.duplicate_count,
+            "visual_similarity": record.visual_similarity,
+            "match_count": record.match_count,
+            "segments": len(segments),
+        }
+
+    except Retry:
+        raise
+
+    except Exception as e:
+        logger.error("Duplication check failed for record %s: %s", record_id, e, exc_info=True)
+        if session is not None:
+            session.rollback()
+            # 本次是最后一次执行机会（retries 从 0 计数，达到 max_retries 说明重试已耗尽），
+            # 标记 failed；否则保持 pending 由 Celery 60 秒后重试
+            try:
+                if "repo" in locals() and self.request.retries >= self.max_retries:
+                    failed_record = repo.get(record_id)
+                    if failed_record is not None and failed_record.status != "failed":
+                        failed_record.mark_failed(f"查重失败（已重试{self.max_retries}次）: {e}")
+                        repo.update(failed_record)
+                        session.commit()
+            except Exception as inner:
+                logger.error("Failed to mark duplication record %s as failed: %s", record_id, inner)
+                session.rollback()
+        raise self.retry(exc=e, countdown=60) from e
+
+    finally:
+        if session is not None:
+            session.close()
+        if temp_dir and os.path.isdir(temp_dir):
+            shutil.rmtree(temp_dir, ignore_errors=True)

packages/adapters/sqlalchemy_impl/duplication_repository.py

@@ -25,6 +25,8 @@ class SQLAlchemyDuplicationRecordRepository:
             status=record.status,
             duplicate_rate=record.duplicate_rate,
             duplicate_count=record.duplicate_count,
+            visual_similarity=record.visual_similarity,
+            match_count=record.match_count,
             video_fingerprint=json.dumps(record.video_fingerprint) if record.video_fingerprint else None,
             error_message=record.error_message,
             created_at=record.created_at,
@@ -58,6 +60,8 @@ class SQLAlchemyDuplicationRecordRepository:
         model.status = record.status
         model.duplicate_rate = record.duplicate_rate
         model.duplicate_count = record.duplicate_count
+        model.visual_similarity = record.visual_similarity
+        model.match_count = record.match_count
         model.video_fingerprint = json.dumps(record.video_fingerprint) if record.video_fingerprint else None
         model.error_message = record.error_message
         model.updated_at = record.updated_at
@@ -121,6 +125,8 @@ class SQLAlchemyDuplicationRecordRepository:
             status=model.status,
             duplicate_rate=model.duplicate_rate,
             duplicate_count=int(model.duplicate_count or 0),
+            visual_similarity=getattr(model, "visual_similarity", None),
+            match_count=getattr(model, "match_count", None),
             video_fingerprint=json.loads(fp_raw) if fp_raw else None,
             error_message=getattr(model, "error_message", ""),
             segments=segments,

packages/adapters/sqlalchemy_impl/models.py

@@ -417,6 +417,9 @@ class DuplicationRecordModel(Base):
     status = Column(String(20), nullable=False, default="pending", index=True)
     duplicate_rate = Column(Float, nullable=True)
     duplicate_count = Column(Integer, nullable=False, default=0)
+    # #1661 手动查重：视觉相似度（0~1）/ 匹配视频数
+    visual_similarity = Column(Float, nullable=True)
+    match_count = Column(Integer, nullable=True)
     video_fingerprint = Column(Text, nullable=True)
     error_message = Column(Text, nullable=False, default="")
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))

packages/domain/duplication.py

@@ -63,6 +63,9 @@ class DuplicationRecord:
     status: str = "pending"  # pending / processing / completed / failed
     duplicate_rate: float | None = None  # 0-100
     duplicate_count: int = 0
+    # #1661 手动查重：视觉相似度（归一化 0~1）/ 匹配视频数
+    visual_similarity: float | None = None
+    match_count: int | None = None
     video_fingerprint: dict[str, Any] | None = None
     error_message: str = ""
     segments: list[DuplicateSegment] = field(default_factory=list)
@@ -98,13 +101,23 @@ class DuplicationRecord:
         self.status = "processing"
         self.updated_at = datetime.now(timezone.utc)
 
-    def mark_completed(self, duplicate_rate: float, duplicate_count: int, segments: list[DuplicateSegment]) -> None:
+    def mark_completed(
+        self,
+        duplicate_rate: float,
+        duplicate_count: int,
+        segments: list[DuplicateSegment],
+        *,
+        visual_similarity: float | None = None,
+        match_count: int | None = None,
+    ) -> None:
         if not 0 <= duplicate_rate <= 100:
             raise ValueError("duplicate_rate must be between 0 and 100")
         self.status = "completed"
         self.duplicate_rate = duplicate_rate
         self.duplicate_count = duplicate_count
         self.segments = segments
+        self.visual_similarity = visual_similarity
+        self.match_count = match_count
         self.updated_at = datetime.now(timezone.utc)
 
     def mark_failed(self, error_message: str) -> None:
@@ -133,6 +146,8 @@ class DuplicationRecord:
         self.status = "pending"
         self.duplicate_rate = None
         self.duplicate_count = 0
+        self.visual_similarity = None
+        self.match_count = None
         self.error_message = ""
         self.segments = []
         self.video_fingerprint = None

tests/unit/test_duplication_check_worker.py

@@ -0,0 +1,378 @@
+"""#1661 手动查重 worker task 测试：成功/失败/重试/片段映射/schema 字段。"""
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+# cv2/numpy 在测试环境不可用，提前 mock
+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 _get_task(mod):
+    """返回 (run_callable, real_task)。
+
+    - celery task 环境：run 是 bound method（self 已绑定），retry 用 patch.object 打桩
+    - 原始函数环境：用一个 mock_self 作为 self
+    """
+    task_obj = mod.process_duplication_check
+    real = task_obj._get_current_object() if hasattr(task_obj, "_get_current_object") else task_obj
+    if hasattr(real, "run") and hasattr(real, "retry"):
+        return real.run, real, True  # bound
+    return real, None, False
+
+
+def _run(mod, record_id, retries=0):
+    """执行 task，返回 (result_or_None, raised_exc, mock_self_or_None)。"""
+    from celery.exceptions import Retry as CeleryRetry
+
+    func, real_task, bound = _get_task(mod)
+    raised = None
+    result = None
+    if bound:
+        mock_retry = MagicMock(side_effect=CeleryRetry("retry"))
+        with patch.object(real_task, "retry", mock_retry):
+            real_task.request.retries = retries
+            real_task.max_retries = 3
+            try:
+                result = func(record_id)
+            except CeleryRetry as e:
+                raised = e
+        return result, raised, None
+    mock_self = MagicMock()
+    mock_self.request.retries = retries
+    mock_self.max_retries = 3
+    mock_self.retry = MagicMock(side_effect=CeleryRetry("retry"))
+    try:
+        result = func(mock_self, record_id)
+    except CeleryRetry as e:
+        raised = e
+    return result, raised, mock_self
+
+
+def _make_record(status="pending"):
+    from packages.domain.duplication import DuplicationRecord
+
+    record = DuplicationRecord.create(
+        user_id="user-1",
+        filename="query.mp4",
+        file_size=1024,
+        storage_key="duplication/abc/query.mp4",
+    )
+    if status != "pending":
+        record.status = status
+    return record
+
+
+def _make_fingerprint():
+    from video_processing.dedup import FingerprintChunk, VideoFingerprint
+
+    chunks = [
+        FingerprintChunk(start_time_ms=0, end_time_ms=2000, phash_binary="0" * 16, color_histogram=[], frame_count=1),
+        FingerprintChunk(
+            start_time_ms=2000, end_time_ms=4000, phash_binary="1" * 16, color_histogram=[], frame_count=1
+        ),
+    ]
+    return VideoFingerprint(
+        md5="qmd5",
+        keyframe_phashes=[c.phash_binary for c in chunks],
+        color_histograms=[],
+        duration=10000.0,
+        resolution=(720, 1280),
+        chunks=chunks,
+    )
+
+
+def _patch_common(record, storage=None, dedup=None, session=None):
+    from worker_app.tasks import duplication_check as mod
+
+    fake_repo = MagicMock()
+    fake_repo.get.return_value = record
+    return [
+        patch.object(mod, "SessionLocal", return_value=session or MagicMock()),
+        patch.object(mod, "SQLAlchemyDuplicationRecordRepository", return_value=fake_repo),
+        patch.object(mod, "get_storage_service", return_value=storage or MagicMock()),
+        patch.object(mod, "VideoDeduplicator", return_value=dedup or MagicMock()),
+    ], fake_repo
+
+
+class TestProcessDuplicationCheckSuccess:
+    def test_success_flow_updates_record(self):
+        from worker_app.tasks import duplication_check as mod
+
+        record = _make_record()
+        fake_session = MagicMock()
+        fake_storage = MagicMock()
+        fake_dedup = MagicMock()
+        fake_dedup.compute_fingerprint.return_value = _make_fingerprint()
+        fake_dedup.compute_duplicate_rate.return_value = {
+            "duplicate_rate": 42.5,
+            "visual_similarity": 0.83,
+            "match_count": 1,
+        }
+        patches, fake_repo = _patch_common(record, storage=fake_storage, dedup=fake_dedup, session=fake_session)
+        patches.append(patch.object(mod, "_build_domain_segments", return_value=(["SEG"], 1)))
+        for p in patches:
+            p.start()
+        try:
+            result, raised, _ = _run(mod, record.id)
+        finally:
+            for p in patches:
+                p.stop()
+
+        assert raised is None
+        assert result["ok"] is True
+        assert result["status"] == "completed"
+        assert result["duplicate_rate"] == 42.5
+        assert result["visual_similarity"] == 0.83
+        assert result["match_count"] == 1
+        assert result["segments"] == 1
+
+        assert record.status == "completed"
+        assert record.duplicate_rate == 42.5
+        assert record.visual_similarity == 0.83
+        assert record.match_count == 1
+        assert record.duplicate_count == 1
+        assert record.segments == ["SEG"]
+
+        fake_storage.download_file.assert_called_once()
+        fake_dedup.compute_fingerprint.assert_called_once()
+        _, kwargs = fake_dedup.compute_duplicate_rate.call_args
+        assert kwargs["scope"] == "user"
+        assert kwargs["user_id"] == "user-1"
+        assert kwargs["current_video_id"] is None
+        assert fake_repo.update.call_count >= 2
+        fake_session.commit.assert_called()
+        fake_session.close.assert_called()
+
+    def test_already_completed_is_skipped(self):
+        from worker_app.tasks import duplication_check as mod
+
+        record = _make_record(status="completed")
+        patches, fake_repo = _patch_common(record)
+        for p in patches:
+            p.start()
+        try:
+            result, raised, _ = _run(mod, record.id)
+        finally:
+            for p in patches:
+                p.stop()
+        assert raised is None
+        assert result.get("skipped") is True
+        fake_repo.update.assert_not_called()
+
+
+class TestProcessDuplicationCheckFailure:
+    def test_record_not_found_raises(self):
+        from worker_app.tasks import duplication_check as mod
+
+        fake_repo = MagicMock()
+        fake_repo.get.return_value = None
+        patches = [
+            patch.object(mod, "SessionLocal", return_value=MagicMock()),
+            patch.object(mod, "SQLAlchemyDuplicationRecordRepository", return_value=fake_repo),
+            patch.object(mod, "get_storage_service", return_value=MagicMock()),
+        ]
+        for p in patches:
+            p.start()
+        try:
+            _result, raised, _ = _run(mod, "nope", retries=0)
+        finally:
+            for p in patches:
+                p.stop()
+        # 找不到记录触发异常 → retry（第一次）
+        assert raised is not None
+
+    def test_download_failure_retries_then_marks_failed(self):
+        from worker_app.tasks import duplication_check as mod
+
+        # 第一次失败（retries=0）：保持 pending
+        record = _make_record()
+        fake_storage = MagicMock()
+        fake_storage.download_file.side_effect = RuntimeError("oss network down")
+        patches, _ = _patch_common(record, storage=fake_storage)
+        for p in patches:
+            p.start()
+        try:
+            _, raised, _ = _run(mod, record.id, retries=0)
+        finally:
+            for p in patches:
+                p.stop()
+        assert raised is not None
+        assert record.status == "processing", "首次失败不应标记 failed（已进入 processing 等待重试）"
+
+        # 最后一次（retries==max_retries=3）：标记 failed
+        record2 = _make_record()
+        patches2, fake_repo2 = _patch_common(record2, storage=fake_storage)
+        for p in patches2:
+            p.start()
+        try:
+            _run(mod, record2.id, retries=3)
+        finally:
+            for p in patches2:
+                p.stop()
+        assert record2.status == "failed"
+        assert "查重失败" in record2.error_message
+        fake_repo2.update.assert_called()
+
+    def test_temp_dir_cleaned_after_failure(self):
+        import os
+        import tempfile
+
+        from worker_app.tasks import duplication_check as mod
+
+        record = _make_record()
+        fake_storage = MagicMock()
+        fake_storage.download_file.side_effect = RuntimeError("boom")
+
+        created_dirs = []
+        real_mkdtemp = tempfile.mkdtemp
+
+        def fake_mkdtemp(prefix=None):
+            d = real_mkdtemp(prefix=prefix)
+            created_dirs.append(d)
+            return d
+
+        patches, _ = _patch_common(record, storage=fake_storage)
+        patches.append(patch.object(mod.tempfile, "mkdtemp", fake_mkdtemp))
+        for p in patches:
+            p.start()
+        try:
+            _run(mod, record.id, retries=0)
+        finally:
+            for p in patches:
+                p.stop()
+
+        assert created_dirs, "mkdtemp should have been called"
+        assert not os.path.isdir(created_dirs[0]), "temp dir should be removed in finally"
+
+
+class TestBuildDomainSegments:
+    def test_maps_worker_segments_to_domain_with_seconds_and_percent(self):
+        from video_processing.dedup import DuplicateSegment as WorkerSegment
+        from worker_app.tasks import duplication_check as mod
+
+        fingerprint = _make_fingerprint()
+
+        from packages.domain import GeneratedVideo
+
+        existing = GeneratedVideo(
+            id="vid-1",
+            project_id="proj-1",
+            generation_task_id="t1",
+            name="成片A",
+            file_url="oss://x",
+            file_size=1,
+            duration=10.0,
+            width=720,
+            height=1280,
+            fps=30.0,
+            video_fingerprint={"md5": "x"},
+        )
+        fake_video_repo = MagicMock()
+        fake_video_repo.list_by_user.return_value = [existing]
+
+        fake_dedup = MagicMock()
+        fake_dedup._get_existing_chunks.return_value = [
+            {"phash_binary": "0" * 16, "start_time_ms": 0, "end_time_ms": 2000, "color_histogram": []},
+        ]
+        worker_seg = WorkerSegment(
+            query_start_ms=1000,
+            query_end_ms=3000,
+            target_start_ms=5000,
+            target_end_ms=7000,
+            avg_distance=6.0,
+        )
+
+        with (
+            patch.object(mod, "SQLAlchemyGeneratedVideoRepository", return_value=fake_video_repo),
+            patch.object(mod, "find_duplicate_segments", return_value=[worker_seg]),
+        ):
+            segments, dup_count = mod._build_domain_segments(fingerprint, MagicMock(), fake_dedup, "user-1")
+
+        assert dup_count == 1
+        assert len(segments) == 1
+        seg = segments[0]
+        assert seg.source_start == 1.0
+        assert seg.source_end == 3.0
+        assert seg.matched_start == 5.0
+        assert seg.matched_end == 7.0
+        assert seg.matched_video_id == "vid-1"
+        assert seg.matched_video_name == "成片A"
+        assert abs(seg.similarity - 90.6) < 0.2
+
+    def test_skips_videos_without_chunks(self):
+        from worker_app.tasks import duplication_check as mod
+
+        fingerprint = _make_fingerprint()
+        from packages.domain import GeneratedVideo
+
+        existing = GeneratedVideo(
+            id="vid-2",
+            project_id="p",
+            generation_task_id="t",
+            name="老视频",
+            file_url="oss://x",
+            file_size=1,
+            duration=5.0,
+            width=720,
+            height=1280,
+            fps=30.0,
+            video_fingerprint={"md5": "old"},
+        )
+        fake_video_repo = MagicMock()
+        fake_video_repo.list_by_user.return_value = [existing]
+        fake_dedup = MagicMock()
+        fake_dedup._get_existing_chunks.return_value = []
+
+        with patch.object(mod, "SQLAlchemyGeneratedVideoRepository", return_value=fake_video_repo):
+            segments, dup_count = mod._build_domain_segments(fingerprint, MagicMock(), fake_dedup, "u")
+        assert segments == []
+        assert dup_count == 0
+
+
+class TestDuplicationSchemaAndDomainNewFields:
+    def test_record_response_includes_new_fields(self):
+        from app.schemas.duplication import DuplicationRecordResponse
+
+        resp = DuplicationRecordResponse(
+            id="r1",
+            filename="f.mp4",
+            file_size=1,
+            status="completed",
+            duplicate_rate=10.0,
+            duplicate_count=1,
+            visual_similarity=0.5,
+            match_count=2,
+            created_at="2026-09-04T00:00:00",
+            updated_at="2026-09-04T00:00:00",
+        )
+        assert resp.visual_similarity == 0.5
+        assert resp.match_count == 2
+
+    def test_record_response_new_fields_default_none(self):
+        from app.schemas.duplication import DuplicationRecordResponse
+
+        resp = DuplicationRecordResponse(id="r1", filename="f.mp4", file_size=1, created_at="x", updated_at="y")
+        assert resp.visual_similarity is None
+        assert resp.match_count is None
+
+    def test_domain_mark_completed_accepts_new_fields(self):
+        record = _make_record()
+        record.mark_completed(33.0, 2, [], visual_similarity=0.77, match_count=3)
+        assert record.status == "completed"
+        assert record.visual_similarity == 0.77
+        assert record.match_count == 3
+
+    def test_reset_for_retry_clears_new_fields(self):
+        record = _make_record()
+        record.mark_completed(10.0, 1, [], visual_similarity=0.5, match_count=1)
+        record.status = "failed"
+        record.reset_for_retry()
+        assert record.status == "pending"
+        assert record.visual_similarity is None
+        assert record.match_count is None