"""手动查重任务（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() # 超过重试上限：标记 failed 并返回失败结果，不再 retry if "repo" in locals() and self.request.retries >= self.max_retries: try: 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() return {"ok": False, "record_id": record_id, "status": "failed", "error": str(e)} # 未达上限：60 秒后重试 raise self.retry(exc=e, countdown=60) from e return {"ok": False, "record_id": record_id, "status": "failed", "error": str(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)