From a5e56939a5db863457a7abb72b5aca0fff9355e6 Mon Sep 17 00:00:00 2001 From: saas-backend-agent Date: Fri, 4 Sep 2026 01:11:04 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat(worker):=20=E6=89=8B=E5=8A=A8=E6=9F=A5?= =?UTF-8?q?=E9=87=8D=20worker=20task=20+=20visual=5Fsimilarity/match=5Fcou?= =?UTF-8?q?nt=20=E5=AD=97=E6=AE=B5=20#1661?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 Celery task worker.process_duplication_check： OSS 下载上传视频 → 动态抽帧指纹 → 跨项目与用户所有成片比对 (compute_duplicate_rate + find_duplicate_segments) → 更新 DuplicationRecord - 状态流转 pending → processing → completed/failed，失败重试 3 次间隔 60s， 重试耗尽标记 failed，临时文件 finally 清理 - 片段映射：worker 分片 DuplicateSegment(ms/汉明距离) → 领域 DuplicateSegment(秒/0-100 相似度) - DuplicationRecord 实体新增 visual_similarity/match_count（可选 None，向后兼容）， mark_completed()/reset_for_retry() 同步 - migration 065: duplication_records 表加两列 - repository/model/schemas/routes 读写映射与响应透传 - 上传接口与重试接口 enqueue task（此前记录一直 pending 的根因） - 11 个新单测：成功/跳过/未找到/重试/失败标记/临时文件清理/片段映射/schema --- ...sual_similarity_match_count_duplication.py | 25 ++ apps/api/app/api/routes/duplication.py | 9 + apps/api/app/schemas/duplication.py | 3 + apps/worker/worker_app/celery_app.py | 1 + .../worker_app/tasks/duplication_check.py | 196 +++++++++ .../sqlalchemy_impl/duplication_repository.py | 6 + packages/adapters/sqlalchemy_impl/models.py | 3 + packages/domain/duplication.py | 17 +- tests/unit/test_duplication_check_worker.py | 379 ++++++++++++++++++ 9 files changed, 638 insertions(+), 1 deletion(-) create mode 100644 alembic/versions/065_add_visual_similarity_match_count_duplication.py create mode 100644 apps/worker/worker_app/tasks/duplication_check.py create mode 100644 tests/unit/test_duplication_check_worker.py diff --git a/alembic/versions/065_add_visual_similarity_match_count_duplication.py b/alembic/versions/065_add_visual_similarity_match_count_duplication.py new file mode 100644 index 000000000..617902f55 --- /dev/null +++ b/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") diff --git a/apps/api/app/api/routes/duplication.py b/apps/api/app/api/routes/duplication.py index 48eb3c92d..2b1cda323 100644 --- a/apps/api/app/api/routes/duplication.py +++ b/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, diff --git a/apps/api/app/schemas/duplication.py b/apps/api/app/schemas/duplication.py index eb422e718..c0a72f7c1 100644 --- a/apps/api/app/schemas/duplication.py +++ b/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 diff --git a/apps/worker/worker_app/celery_app.py b/apps/worker/worker_app/celery_app.py index 751e4da05..d953c23f4 100755 --- a/apps/worker/worker_app/celery_app.py +++ b/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", diff --git a/apps/worker/worker_app/tasks/duplication_check.py b/apps/worker/worker_app/tasks/duplication_check.py new file mode 100644 index 000000000..dcfa49c17 --- /dev/null +++ b/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) diff --git a/packages/adapters/sqlalchemy_impl/duplication_repository.py b/packages/adapters/sqlalchemy_impl/duplication_repository.py index b8a1216b1..d0df08d63 100644 --- a/packages/adapters/sqlalchemy_impl/duplication_repository.py +++ b/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, diff --git a/packages/adapters/sqlalchemy_impl/models.py b/packages/adapters/sqlalchemy_impl/models.py index 1a0633618..8e71fcb32 100755 --- a/packages/adapters/sqlalchemy_impl/models.py +++ b/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)) diff --git a/packages/domain/duplication.py b/packages/domain/duplication.py index 3b3c33de3..cc6c6d1df 100644 --- a/packages/domain/duplication.py +++ b/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 diff --git a/tests/unit/test_duplication_check_worker.py b/tests/unit/test_duplication_check_worker.py new file mode 100644 index 000000000..884e95b54 --- /dev/null +++ b/tests/unit/test_duplication_check_worker.py @@ -0,0 +1,379 @@ +"""#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 -- 2.54.0 From 77267ac1017c3cbd5610f6cadf78451902401825 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Thu, 3 Sep 2026 17:19:59 +0000 Subject: [PATCH 2/2] style: auto-format with black + isort + prettier [skip ci-format-check] --- tests/unit/test_duplication_check_worker.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/unit/test_duplication_check_worker.py b/tests/unit/test_duplication_check_worker.py index 884e95b54..b6dd9f901 100644 --- a/tests/unit/test_duplication_check_worker.py +++ b/tests/unit/test_duplication_check_worker.py @@ -221,7 +221,6 @@ class TestProcessDuplicationCheckFailure: def test_temp_dir_cleaned_after_failure(self): import os - import tempfile from worker_app.tasks import duplication_check as mod -- 2.54.0