xiaoxia-saas/apps/worker/worker_app/tasks/duplication_check.py

"""手动查重任务（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)