xiaoxia-saas/apps/worker/video_processing/dedup_helpers.py

"""查重辅助函数 — 从 generation.py 提取的 GeneratedVideo 记录 + 查重逻辑.

供 generate_video 共同复用，
创建 GeneratedVideo 记录后计算指纹并执行项目级 + 批次内查重。

v2: 两阶段持久化 — 先计算所有查重数据，再一次性 commit，
避免中间异常导致 duplicate_rate 等字段缺失。
"""

from __future__ import annotations

import logging
from uuid import uuid4

from sqlalchemy.orm import Session

logger = logging.getLogger(__name__)


def create_video_record_and_dedup(
    *,
    generation_task_id: str,
    project_id: str,
    user_id: str = "",
    batch_id: str,
    file_url: str,
    file_size: int,
    duration: float,
    video_path: str,
    mode: str,
    session: Session,
    width: int = 1280,
    height: int = 720,
    fps: float = 25.0,
    name: str = "",
    thumbnail_url: str = "",
) -> int:
    """创建 GeneratedVideo 记录，计算指纹并执行查重（历史 + 批次）。

    采用两阶段持久化：先计算所有指纹/查重数据（内存），
    再一次性写入数据库并 commit。若指纹计算失败，
    视频记录仍会创建（无查重数据），但保证不会出现"写了记录却没 commit"的中间态。

    Returns:
        创建的视频记录数量（1 表示成功，0 表示失败）
    """
    from video_processing.dedup import VideoDeduplicator, _save_fingerprint_chunks

    from packages.adapters.sqlalchemy_impl.generated_video_repository import (
        SQLAlchemyGeneratedVideoRepository,
    )
    from packages.domain import GeneratedVideo

    try:
        video_id = uuid4().hex
        video_name = name.strip() if name else f"generated-{generation_task_id[:8]}.mp4"

        # ── Phase 1: 构建视频记录（内存，不 commit） ────────────────
        generated_video = GeneratedVideo(
            id=video_id,
            project_id=project_id,
            user_id=user_id,
            generation_task_id=generation_task_id,
            name=video_name,
            file_url=file_url,
            file_size=file_size,
            duration=duration,
            width=width,
            height=height,
            fps=fps,
            status="completed",
            generation_params={"mode": mode},
            thumbnail_url=thumbnail_url or None,
        )

        # ── Phase 2: 计算指纹 & 查重（全部在内存） ────────────────
        deduplicator = VideoDeduplicator()
        fingerprint = None

        try:
            fingerprint = deduplicator.compute_fingerprint(video_path)
        except Exception as fp_err:
            logger.warning("Fingerprint computation failed for %s: %s", video_id, fp_err)

        if fingerprint is not None:
            generated_video.video_fingerprint = fingerprint.to_dict()

            # 写入分片指纹表（失败不阻塞）
            try:
                _save_fingerprint_chunks(fingerprint, video_id, project_id, user_id, session)
            except Exception as chunk_err:
                logger.warning("Failed to save fingerprint chunks for %s: %s", video_id, chunk_err)

            # (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:
                duplicate_result = deduplicator.check_batch_duplicate(fingerprint, batch_id, video_id, session)

            if duplicate_result:
                generated_video.is_duplicate = True
                generated_video.duplicate_of = duplicate_result["duplicate_of"]
                logger.info(
                    "Duplicate detected: %s -> %s (reason=%s, similarity=%.3f)",
                    video_id,
                    duplicate_result["duplicate_of"],
                    duplicate_result["reason"],
                    duplicate_result["similarity"],
                )
            else:
                generated_video.is_duplicate = False
                generated_video.duplicate_of = None

            # 计算重复率百分比（跨项目全局）
            try:
                rate_result = deduplicator.compute_duplicate_rate(
                    fingerprint,
                    project_id,
                    video_id,
                    session,
                    scope="user",
                    user_id=user_id,
                )
                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

        # ── Phase 3: 一次性持久化 ─────────────────────────────────
        video_repo = SQLAlchemyGeneratedVideoRepository(session)
        video_repo.create(generated_video)

        if thumbnail_url:
            logger.info("Thumbnail set for video %s: %s", video_id, thumbnail_url[:80])

        session.commit()
        logger.info(
            "GeneratedVideo record created: %s (task=%s, dup=%s, rate=%s)",
            video_id,
            generation_task_id,
            generated_video.is_duplicate,
            generated_video.duplicate_rate,
        )
        return 1
    except Exception as e:
        logger.error(
            "Failed to create video record / dedup for task %s: %s",
            generation_task_id,
            e,
        )
        session.rollback()
        return 0