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

"""Video deduplication module - compute fingerprints and detect duplicates.

Dynamic keyframe detection + sliding window temporal matching (Issue #1659).
"""

import hashlib
import logging
import math
import os
import statistics
import tempfile
from dataclasses import dataclass, field
from typing import Optional
from uuid import uuid4

import cv2
import numpy as np
from celery import Task
from sqlalchemy.orm import Session
from worker_app.celery_app import celery_app
from worker_app.db import SessionLocal

from packages.adapters.sqlalchemy_impl.generated_video_repository import SQLAlchemyGeneratedVideoRepository
from packages.adapters.sqlalchemy_impl.models import VideoFingerprintChunkModel
from packages.shared.storage import get_storage_service

logger = logging.getLogger(__name__)

# ── 关键帧检测常量 ──────────────────────────────────────────────
SCENE_CHANGE_THRESHOLD = 30  # 灰度差异阈值
MIN_KEYFRAME_INTERVAL_SEC = 1.0  # 最小关键帧间隔（秒）
MAX_KEYFRAMES = 30  # 最大关键帧数
MIN_KEYFRAMES = 5  # 最小关键帧数
LONG_VIDEO_SEGMENT_SEC = 30  # 长视频每段秒数
LONG_VIDEO_DURATION_THRESHOLD_SEC = 180  # 3 分钟阈值
MIN_FRAMES_PER_SEGMENT = 2  # 长视频每段最少帧数

# ── 滑动窗口匹配常量 ────────────────────────────────────────────
SEGMENT_MATCH_THRESHOLD = 8  # 帧匹配汉明距离阈值
MIN_CONSECUTIVE_MATCHES = 5  # 最少连续匹配帧数
MAX_GAP = 2  # 允许的最大间隙帧数

# ── 融合判定常量 ────────────────────────────────────────────────
PHASH_WEIGHT = 0.7  # pHash 权重
HISTOGRAM_WEIGHT = 0.3  # 直方图权重
MATCH_RATIO_THRESHOLD = 0.7  # 至少 70% 帧匹配
DUPLICATE_THRESHOLD = 0.70  # 融合后相似度阈值


# ── 感知哈希 & 颜色直方图工具函数 ────────────────────────────────


def compute_phash(image: np.ndarray, hash_size: int = 8) -> str:
    """计算图像的感知哈希（pHash），基于 DCT（离散余弦变换）。

    算法步骤：
    1. 将图像缩放到 hash_size*4 × hash_size*4（默认 32×32）
    2. 转为灰度图，应用 2D DCT 提取频率分量
    3. 取左上角 hash_size×hash_size 的低频分量（默认 8×8 = 64 bit）
    4. 排除 DC 分量（[0,0] 位置），计算中位数
    5. 每个分量与中位数比较，生成二值 hash

    Args:
        image: BGR 格式的 numpy 图像数组
        hash_size: 哈希边长，默认 8（生成 64-bit hash）

    Returns:
        十六进制字符串表示的感知哈希
    """
    # Resize to 32x32 for DCT
    resized = cv2.resize(image, (hash_size * 4, hash_size * 4))
    gray = cv2.cvtColor(resized, cv2.COLOR_BGR2GRAY).astype(np.float32)
    # Apply 2D DCT
    dct = cv2.dct(gray)
    # Take top-left 8x8 low-frequency components
    dct_low = dct[:hash_size, :hash_size]
    # Compute median (excluding DC component at [0,0])
    dct_low[0, 0] = 0
    median = np.median(dct_low)
    # Generate hash based on comparison with median
    diff = (dct_low > median).astype(int)
    hash_str = "".join(str(b) for row in diff for b in row)
    return hex(int(hash_str, 2))[2:]


def hamming_distance(hash1: str, hash2: str) -> int:
    """计算两个十六进制哈希之间的汉明距离（不同 bit 位数）。

    使用 XOR 异或 + bit 计数：bin(h1 ^ h2).count("1")。
    例如：hamming_distance("00", "ff") = 8（8 个 bit 全不同）。

    Args:
        hash1: 十六进制字符串
        hash2: 十六进制字符串

    Returns:
        不同 bit 的数量
    """
    h1, h2 = int(hash1, 16), int(hash2, 16)
    return bin(h1 ^ h2).count("1")


def compute_color_histogram(image: np.ndarray, bins: int = 32) -> list[float]:
    """Compute color histogram for an image."""
    hist = []
    for i in range(3):
        h = cv2.calcHist([image], [i], None, [bins], [0, 256])
        h = cv2.normalize(h, h).flatten()
        hist.extend(h)
    return hist


# ── 关键帧检测 ──────────────────────────────────────────────────


def detect_keyframe_timestamps(
    video_path: str,
    *,
    min_interval_sec: float = MIN_KEYFRAME_INTERVAL_SEC,
    max_frames: int = MAX_KEYFRAMES,
    min_frames: int = MIN_KEYFRAMES,
) -> list[float]:
    """检测视频中的场景切换点，返回关键帧时间戳列表（秒）。

    算法：
    1. 降采样到 320x240，逐帧转灰度
    2. 计算相邻帧灰度差异（像素均值差）
    3. 差异 > SCENE_CHANGE_THRESHOLD(30) 标记为候选关键帧
    4. 相邻关键帧间隔 < min_interval_sec 的，保留差异更大的那个
    5. 数量裁剪到 [min_frames, max_frames]

    对于长视频（>3分钟）：
    - 每 30 秒一个分段
    - 每个分段至少选 2 个关键帧（如果分段内无场景切换，均匀取 2 帧）
    """
    cap = cv2.VideoCapture(video_path)
    if not cap.isOpened():
        raise RuntimeError(f"Cannot open video: {video_path}")

    fps = cap.get(cv2.CAP_PROP_FPS)
    frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    duration = frame_count / fps if fps > 0 else 0

    if duration <= 0:
        cap.release()
        return []

    # 逐帧检测场景切换
    candidates: list[tuple[float, float]] = []  # (timestamp_sec, diff_score)
    prev_gray = None

    while True:
        ret, frame = cap.read()
        if not ret:
            break

        # 降采样 + 灰度
        small = cv2.resize(frame, (320, 240))
        gray = cv2.cvtColor(small, cv2.COLOR_BGR2GRAY).astype(np.float32)

        if prev_gray is not None:
            diff = float(np.mean(np.abs(gray - prev_gray)))
            if diff > SCENE_CHANGE_THRESHOLD:
                pos_ms = cap.get(cv2.CAP_PROP_POS_MSEC)
                candidates.append((pos_ms / 1000.0, diff))

        prev_gray = gray

    cap.release()

    # 按最小间隔过滤（保留差异更大的）
    filtered: list[tuple[float, float]] = []
    for ts, diff in sorted(candidates):
        if filtered and (ts - filtered[-1][0]) < min_interval_sec:
            if diff > filtered[-1][1]:
                filtered[-1] = (ts, diff)
        else:
            filtered.append((ts, diff))

    keyframe_times = [ts for ts, _ in filtered]

    # 数量不足 min_frames 时，在时间轴上均匀补充
    if len(keyframe_times) < min_frames:
        uniform = [duration * (i + 0.5) / min_frames for i in range(min_frames)]
        keyframe_times = sorted(set(uniform) | set(keyframe_times))
        # 如果合并后还不足 min_frames，直接用均匀分布
        if len(keyframe_times) < min_frames:
            keyframe_times = uniform

    # 数量超过 max_frames 时，均匀采样
    if len(keyframe_times) > max_frames:
        step = len(keyframe_times) / max_frames
        keyframe_times = [keyframe_times[int(i * step)] for i in range(max_frames)]

    # 长视频分段保底（>3分钟）
    if duration > LONG_VIDEO_DURATION_THRESHOLD_SEC:
        segment_count = int(duration / LONG_VIDEO_SEGMENT_SEC)
        for seg_idx in range(segment_count):
            seg_start = seg_idx * LONG_VIDEO_SEGMENT_SEC
            seg_end = min((seg_idx + 1) * LONG_VIDEO_SEGMENT_SEC, duration)
            seg_frames = [t for t in keyframe_times if seg_start <= t < seg_end]
            if len(seg_frames) < MIN_FRAMES_PER_SEGMENT:
                # 均匀补齐
                for i in range(MIN_FRAMES_PER_SEGMENT):
                    t = seg_start + LONG_VIDEO_SEGMENT_SEC * (i + 0.5) / MIN_FRAMES_PER_SEGMENT
                    if t not in keyframe_times and seg_start <= t < seg_end:
                        keyframe_times.append(t)
        keyframe_times.sort()

    return keyframe_times


# ── 数据类 ──────────────────────────────────────────────────────


@dataclass
class FingerprintChunk:
    """单个分片指纹数据。"""

    start_time_ms: int
    end_time_ms: int
    phash_binary: str
    color_histogram: list[float]
    frame_count: int = 1


@dataclass
class DuplicateSegment:
    """一段重复片段的描述。"""

    query_start_ms: int
    query_end_ms: int
    target_start_ms: int
    target_end_ms: int
    avg_distance: float  # 该段内帧的平均汉明距离


@dataclass
class VideoFingerprint:
    """Video fingerprint containing multiple similarity metrics."""

    md5: str
    keyframe_phashes: list[str]
    color_histograms: list[list[float]]
    duration: float
    resolution: tuple[int, int]
    chunks: list[FingerprintChunk] = field(default_factory=list)

    def to_dict(self) -> dict:
        # 注意：color_histograms 里的值可能是 np.float32（来自 cv2.normalize），
        # 直接存进 dict 后 SQLAlchemy JSON 序列化会报 "float32 is not JSON serializable"。
        # 这里统一转成 Python 原生 float。
        native_histograms = [[float(v) for v in hist] for hist in self.color_histograms]
        return {
            "md5": self.md5,
            "keyframe_phashes": self.keyframe_phashes,
            "color_histograms": native_histograms,
            "duration": float(self.duration),
            "resolution": [int(self.resolution[0]), int(self.resolution[1])],
            "chunks": [
                {
                    "start_time_ms": c.start_time_ms,
                    "end_time_ms": c.end_time_ms,
                    "phash_binary": c.phash_binary,
                    "color_histogram": [float(v) for v in c.color_histogram],
                    "frame_count": c.frame_count,
                }
                for c in self.chunks
            ],
        }

    def to_chunk_models(self, video_id: str, project_id: str, user_id: str = "") -> list[VideoFingerprintChunkModel]:
        """将分片数据转为 SQLAlchemy Model 列表，用于批量写入 video_fingerprint_chunks 表。"""
        models = []
        for chunk in self.chunks:
            models.append(
                VideoFingerprintChunkModel(
                    id=uuid4().hex,
                    video_id=video_id,
                    project_id=project_id,
                    user_id=user_id,
                    start_time_ms=chunk.start_time_ms,
                    end_time_ms=chunk.end_time_ms,
                    phash_binary=chunk.phash_binary,
                    color_histogram=[float(v) for v in chunk.color_histogram],
                    frame_count=chunk.frame_count,
                )
            )
        return models


# ── 滑动窗口时序匹配 ────────────────────────────────────────────


def find_duplicate_segments(
    query_chunks: list,
    target_chunks: list,
    *,
    match_threshold: int = SEGMENT_MATCH_THRESHOLD,
    min_consecutive: int = MIN_CONSECUTIVE_MATCHES,
    max_gap: int = MAX_GAP,
) -> list[DuplicateSegment]:
    """滑动窗口时序匹配：找出两组分片之间的重复片段。

    算法：
    1. 对每个 query chunk，找到 target 中汉明距离最小的 chunk
    2. 距离 <= match_threshold 视为匹配
    3. 找连续匹配的 run（允许 max_gap 帧间隙）
    4. 连续匹配数 >= min_consecutive 的 run 报告为重复片段

    Args:
        query_chunks: 查询视频的分片列表（FingerprintChunk 或 dict）
        target_chunks: 目标视频的分片列表
        match_threshold: 汉明距离匹配阈值
        min_consecutive: 最少连续匹配帧数
        max_gap: 允许的最大间隙帧数

    Returns:
        DuplicateSegment 列表
    """
    if not query_chunks or not target_chunks:
        return []

    def _get_phash(chunk) -> str:
        if isinstance(chunk, dict):
            return chunk["phash_binary"]
        return chunk.phash_binary

    def _get_start(chunk) -> int:
        if isinstance(chunk, dict):
            return chunk["start_time_ms"]
        return chunk.start_time_ms

    def _get_end(chunk) -> int:
        if isinstance(chunk, dict):
            return chunk["end_time_ms"]
        return chunk.end_time_ms

    # Step 1: 逐帧匹配
    frame_matches: list[tuple[bool, int, int]] = []  # (is_match, min_dist, best_target_idx)
    for qc in query_chunks:
        qc_phash = _get_phash(qc)
        best_dist = 64
        best_idx = 0
        for j, tc in enumerate(target_chunks):
            d = hamming_distance(qc_phash, _get_phash(tc))
            if d < best_dist:
                best_dist = d
                best_idx = j
        frame_matches.append((best_dist <= match_threshold, best_dist, best_idx))

    # Step 2: 找连续匹配的 runs
    runs: list[tuple[int, int]] = []  # list of (start_idx, end_idx)
    run_start = None
    gap_count = 0

    for i, (is_match, _dist, _idx) in enumerate(frame_matches):
        if is_match:
            if run_start is None:
                run_start = i
            gap_count = 0  # 重置间隙
        else:
            if run_start is not None:
                gap_count += 1
                if gap_count > max_gap:
                    # 中断当前 run
                    run_end = i - gap_count  # 最后一个匹配帧的索引
                    # 计算 run 内的实际匹配帧数（总跨度 - 间隙数）
                    total_gaps = sum(1 for k in range(run_start, run_end + 1) if not frame_matches[k][0])
                    matching_count = (run_end - run_start + 1) - total_gaps
                    if matching_count >= min_consecutive:
                        runs.append((run_start, run_end))
                    run_start = None
                    gap_count = 0

    # 处理末尾 run
    if run_start is not None:
        last_idx = len(frame_matches) - 1
        # 回退找到最后一个匹配帧的位置（跳过尾部非匹配帧）
        while last_idx >= run_start and not frame_matches[last_idx][0]:
            last_idx -= 1
        if last_idx >= run_start:
            # 计算 run 内的总间隙数
            total_gaps = sum(1 for k in range(run_start, last_idx + 1) if not frame_matches[k][0])
            matching_count = (last_idx - run_start + 1) - total_gaps
            if matching_count >= min_consecutive:
                runs.append((run_start, last_idx))

    # Step 3: 构建 DuplicateSegment
    segments: list[DuplicateSegment] = []
    for start, end in runs:
        query_start = _get_start(query_chunks[start])
        query_end = _get_end(query_chunks[end])

        # 取目标范围（按最佳匹配的目标 chunk 时间范围）
        target_indices = [frame_matches[k][2] for k in range(start, end + 1) if frame_matches[k][0]]
        if target_indices:
            t_min = min(target_indices)
            t_max = max(target_indices)
            target_start = _get_start(target_chunks[t_min])
            target_end = _get_end(target_chunks[t_max])
        else:
            target_start = _get_start(target_chunks[0])
            target_end = _get_end(target_chunks[-1])

        avg_dist = sum(frame_matches[k][1] for k in range(start, end + 1)) / (end - start + 1)
        segments.append(
            DuplicateSegment(
                query_start_ms=query_start,
                query_end_ms=query_end,
                target_start_ms=target_start,
                target_end_ms=target_end,
                avg_distance=avg_dist,
            )
        )

    return segments


# ── VideoDeduplicator ───────────────────────────────────────────


class VideoDeduplicator:
    """Video deduplication using multiple fingerprint methods."""

    PHASH_THRESHOLD = 8  # Issue #1658: pHash 汉明距离阈值由 10 收紧到 8，降低不同视频误判率
    HISTOGRAM_THRESHOLD = 0.85

    @staticmethod
    def _is_bad_fingerprint(phashes: list[str]) -> bool:
        """检测指纹质量差的视频（黑屏/纯色视频）。

        当视频有多个关键帧但所有 phash 完全相同或极其相似时，
        说明视频内容无变化（如黑屏、纯色画面），这类指纹与任何视频
        比较都会得到虚假的"匹配"结果，应跳过。

        注意：单帧视频（只有 1 个 phash）不视为坏指纹，可能是短视频或抽帧不足。

        Args:
            phashes: 关键帧 phash 列表

        Returns:
            True 表示指纹无效，应跳过
        """
        if not phashes:
            return True
        # 单帧不视为坏指纹（短视频或抽帧不足）
        if len(phashes) == 1:
            return False
        # 多帧但所有 phash 完全相同 → 黑屏/纯色视频
        unique = set(phashes)
        if len(unique) == 1:
            return True
        # 多帧但所有 phash 之间的汉明距离都极小（<3）→ 近似黑屏
        phash_list = list(unique)
        if len(phash_list) >= 2:
            all_distances = []
            for i in range(len(phash_list)):
                for j in range(i + 1, len(phash_list)):
                    all_distances.append(hamming_distance(phash_list[i], phash_list[j]))
            if all_distances and max(all_distances) < 3:
                return True
        return False

    def compute_fingerprint(self, video_path: str) -> VideoFingerprint:
        """Compute video fingerprint using dynamic keyframe detection.

        使用 detect_keyframe_timestamps() 检测内容感知关键帧，
        在每个关键帧处取帧计算 pHash + color_histogram。
        同时保留 MD5 计算和分片数据结构。
        """
        cap = cv2.VideoCapture(video_path)
        if not cap.isOpened():
            raise RuntimeError(f"Cannot open video: {video_path}")

        fps = cap.get(cv2.CAP_PROP_FPS)
        frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
        duration = frame_count / fps if fps > 0 else 0
        width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))

        cap.release()

        # 1. 检测关键帧时间戳
        keyframe_times = detect_keyframe_timestamps(video_path)

        if not keyframe_times:
            return VideoFingerprint(
                md5="",
                keyframe_phashes=[],
                color_histograms=[],
                duration=duration,
                resolution=(width, height),
                chunks=[],
            )

        # 2. 打开视频，逐个关键帧取帧
        cap = cv2.VideoCapture(video_path)
        md5_hash = hashlib.md5(usedforsecurity=False)
        chunks: list[FingerprintChunk] = []

        for i, t_sec in enumerate(keyframe_times):
            seek_ms = t_sec * 1000
            cap.set(cv2.CAP_PROP_POS_MSEC, seek_ms)
            ret, frame = cap.read()
            if not ret:
                continue

            # MD5 计算
            _, buffer = cv2.imencode(".jpg", frame)
            md5_hash.update(buffer)

            phash = compute_phash(frame)
            hist = compute_color_histogram(frame)

            # 计算分片时间范围（从前一个关键帧到下一个关键帧的中点）
            prev_boundary = keyframe_times[i - 1] * 1000 if i > 0 else 0
            next_boundary = keyframe_times[i + 1] * 1000 if i < len(keyframe_times) - 1 else duration * 1000
            start_ms = int((prev_boundary + seek_ms) / 2)
            end_ms = int((seek_ms + next_boundary) / 2)

            chunks.append(
                FingerprintChunk(
                    start_time_ms=start_ms,
                    end_time_ms=end_ms,
                    phash_binary=phash,
                    color_histogram=hist,
                    frame_count=1,
                )
            )

        cap.release()

        # 向后兼容：聚合 keyframe_phashes / color_histograms
        keyframe_phashes = [c.phash_binary for c in chunks]
        color_histograms = [c.color_histogram for c in chunks]

        return VideoFingerprint(
            md5=md5_hash.hexdigest(),
            keyframe_phashes=keyframe_phashes,
            color_histograms=color_histograms,
            duration=duration,
            resolution=(width, height),
            chunks=chunks,
        )

    def _get_existing_chunks(self, video_id: str, session: Session) -> list[dict]:
        """从 video_fingerprint_chunks 表读取分片数据。返回空列表表示无分片数据。"""
        rows = (
            session.query(VideoFingerprintChunkModel)
            .filter(VideoFingerprintChunkModel.video_id == video_id)
            .order_by(VideoFingerprintChunkModel.start_time_ms)
            .all()
        )
        return [
            {
                "phash_binary": r.phash_binary,
                "color_histogram": r.color_histogram,
                "start_time_ms": r.start_time_ms,
                "end_time_ms": r.end_time_ms,
            }
            for r in rows
        ]

    @staticmethod
    def _bhattacharyya_coefficient(hist_a: list[float], hist_b: list[float]) -> float:
        """Bhattacharyya 系数：Σ √(a[i] * b[i])，范围 [0, 1]，1=完全相同。"""
        min_len = min(len(hist_a), len(hist_b))
        a = hist_a[:min_len]
        b = hist_b[:min_len]
        # 纯标准库计算（不依赖 numpy）；max(0.0, ...) 防御上游异常负值导致 sqrt domain error
        return float(sum(math.sqrt(max(0.0, ai * bi)) for ai, bi in zip(a, b, strict=False)))

    @staticmethod
    def _compute_histogram_similarity(
        histograms_a: list[list[float]],
        histograms_b: list[list[float]],
    ) -> float:
        """对每组直方图，找到最佳匹配的 Bhattacharyya 系数，取平均。"""
        if not histograms_a or not histograms_b:
            return 0.0
        similarities = []
        for ha in histograms_a:
            best = 0.0
            for hb in histograms_b:
                bc = VideoDeduplicator._bhattacharyya_coefficient(ha, hb)
                best = max(best, bc)
            similarities.append(best)
        return sum(similarities) / len(similarities) if similarities else 0.0

    @staticmethod
    def _compute_fusion_score(
        median_distance: float,
        histograms_a: list[list[float]],
        histograms_b: list[list[float]],
    ) -> float:
        """pHash 相似度与颜色直方图相似度的加权融合得分（Issue #1658）。

        - phash_similarity = 1.0 - median_distance / 64（64 为 64bit pHash 最大汉明距离）
        - hist_similarity = Bhattacharyya 系数均值；无直方图数据时回退中性值 0.5
        - 融合得分 = PHASH_WEIGHT * phash_similarity + HISTOGRAM_WEIGHT * hist_similarity

        返回 0~1 的原始得分，是否判重由调用方与 DUPLICATE_THRESHOLD 比较决定。
        """
        # DB 中 color_histograms 可能为 NULL（None），显式回退空列表而非 `or []`，
        # 以保留全黑视频的全零直方图（[0,0,...] 为有效数据，空列表才走 0.5 中性回退）。
        hist_a = histograms_a if histograms_a is not None else []
        hist_b = histograms_b if histograms_b is not None else []

        phash_similarity = 1.0 - (median_distance / 64)
        hist_similarity = VideoDeduplicator._compute_histogram_similarity(hist_a, hist_b) if hist_b else 0.5
        return PHASH_WEIGHT * phash_similarity + HISTOGRAM_WEIGHT * hist_similarity

    def check_duplicate(
        self,
        fingerprint: VideoFingerprint,
        project_id: str,
        session: Session,
        *,
        scope: str = "project",
        user_id: str = "",
        duration_sec: float = 0,
    ) -> Optional[dict]:
        """检查视频是否与已有视频重复。

        查重逻辑：
        1. MD5 精确匹配 → similarity=1.0
        2. pHash 中位数距离 + 帧匹配比例 + 直方图融合判定

        判定为重复后，调用 find_duplicate_segments() 获取具体重复片段。

        Args:
            fingerprint: 待检测视频的指纹
            project_id: 项目 ID
            session: 数据库会话
            scope: "project" 项目内查重（默认），"user" 跨项目全局查重
            user_id: 用户 ID（scope="user" 时使用）
            duration_sec: 视频时长（秒），用于时长预过滤 ±15%

        Returns:
            重复信息字典（含 duplicate, duplicate_of, reason, similarity, duplicate_segments），
            或 None 表示未找到重复。
        """
        video_repo = SQLAlchemyGeneratedVideoRepository(session)
        if scope == "user" and user_id:
            dur_min = duration_sec * 0.85 if duration_sec > 0 else 0
            dur_max = duration_sec * 1.15 if duration_sec > 0 else 0
            existing_videos = video_repo.list_by_user(user_id, duration_min=dur_min, duration_max=dur_max)
        else:
            existing_videos = video_repo.list_by_project(project_id)

        for existing in existing_videos:
            if not existing.video_fingerprint:
                continue

            ef = existing.video_fingerprint

            # 精确匹配：MD5 完全一致
            if fingerprint.md5 == ef.get("md5"):
                return {"duplicate": True, "duplicate_of": existing.id, "reason": "exact_md5_match", "similarity": 1.0}

            # 跳过指纹质量差的视频（黑屏/纯色视频）
            existing_phashes_for_check = ef.get("keyframe_phashes", [])
            if self._is_bad_fingerprint(existing_phashes_for_check):
                logger.debug("Skipping bad fingerprint video %s in check_duplicate", existing.id)
                continue

            # 优先从分片表读取已有视频的分片 phash
            existing_phashes = []
            chunk_data = self._get_existing_chunks(existing.id, session)
            if chunk_data:
                existing_phashes = [c["phash_binary"] for c in chunk_data]
            else:
                # 回退：从 JSON 字段读取（存量旧视频）
                existing_phashes = ef.get("keyframe_phashes", [])

            if not existing_phashes:
                continue

            # 计算每个新关键帧到已有关键帧的最小汉明距离
            min_distances = []
            for phash in fingerprint.keyframe_phashes:
                distances = [hamming_distance(phash, ep) for ep in existing_phashes]
                min_distances.append(min(distances))

            # 帧匹配比例检查
            matching_frames = sum(1 for d in min_distances if d < self.PHASH_THRESHOLD)
            match_ratio = matching_frames / len(min_distances) if min_distances else 0
            if match_ratio < MATCH_RATIO_THRESHOLD:
                continue

            # 中位数距离
            median_distance = statistics.median(min_distances) if min_distances else 64
            if median_distance >= self.PHASH_THRESHOLD:
                continue

            # 直方图融合（chunk 表优先，回退 JSON 字段；JSON NULL 显式回退空列表）
            if chunk_data:
                existing_histograms = [c["color_histogram"] for c in chunk_data if c.get("color_histogram")]
            else:
                existing_histograms = ef.get("color_histograms") or []

            combined_score = self._compute_fusion_score(
                median_distance, fingerprint.color_histograms, existing_histograms
            )

            if combined_score < DUPLICATE_THRESHOLD:
                continue

            # 滑动窗口时序匹配：获取具体重复片段
            existing_chunk_objects = (
                chunk_data
                if chunk_data
                else [{"phash_binary": p, "start_time_ms": 0, "end_time_ms": 0} for p in existing_phashes]
            )
            segments = find_duplicate_segments(fingerprint.chunks, existing_chunk_objects)

            return {
                "duplicate": True,
                "duplicate_of": existing.id,
                "reason": "phash_histogram_fusion",
                "similarity": combined_score,
                "duplicate_segments": [
                    {
                        "query_start_ms": s.query_start_ms,
                        "query_end_ms": s.query_end_ms,
                        "target_start_ms": s.target_start_ms,
                        "target_end_ms": s.target_end_ms,
                        "avg_distance": round(s.avg_distance, 2),
                    }
                    for s in segments
                ],
            }

        return None

    def check_batch_duplicate(
        self,
        fingerprint: VideoFingerprint,
        batch_id: str,
        current_video_id: str,
        session: Session,
        *,
        scope: str = "project",
        user_id: str = "",
    ) -> Optional[dict]:
        """检查视频是否与同批次内其他视频重复。

        逻辑与 check_duplicate 一致（MD5 + pHash + 直方图融合 + 时序匹配），
        但搜索范围限定为同 batch_id 的视频。

        Args:
            fingerprint: 待检测视频的指纹
            batch_id: 批次 ID
            current_video_id: 当前视频 ID（排除自身）
            session: 数据库会话
            scope: 保留参数，batch 模式始终按 batch_id 查询
            user_id: 保留参数

        Returns:
            重复信息字典，或 None 表示未找到重复
        """
        video_repo = SQLAlchemyGeneratedVideoRepository(session)
        batch_videos = video_repo.list_by_batch(batch_id)

        for existing in batch_videos:
            if existing.id == current_video_id:
                continue
            if not existing.video_fingerprint:
                continue

            ef = existing.video_fingerprint

            if fingerprint.md5 == ef.get("md5"):
                return {
                    "duplicate": True,
                    "duplicate_of": existing.id,
                    "reason": "batch_exact_md5_match",
                    "similarity": 1.0,
                }

            # 跳过指纹质量差的视频（黑屏/纯色视频）
            existing_phashes_batch = ef.get("keyframe_phashes", [])
            if self._is_bad_fingerprint(existing_phashes_batch):
                logger.debug("Skipping bad fingerprint video %s in check_batch_duplicate", existing.id)
                continue

            # 优先从分片表读取
            existing_phashes = []
            chunk_data = self._get_existing_chunks(existing.id, session)
            if chunk_data:
                existing_phashes = [c["phash_binary"] for c in chunk_data]
            else:
                existing_phashes = ef.get("keyframe_phashes", [])

            if not existing_phashes:
                continue

            min_distances = []
            for phash in fingerprint.keyframe_phashes:
                distances = [hamming_distance(phash, ep) for ep in existing_phashes]
                min_distances.append(min(distances))

            # 帧匹配比例检查
            matching_frames = sum(1 for d in min_distances if d < self.PHASH_THRESHOLD)
            match_ratio = matching_frames / len(min_distances) if min_distances else 0
            if match_ratio < MATCH_RATIO_THRESHOLD:
                continue

            median_distance = statistics.median(min_distances) if min_distances else 64
            if median_distance >= self.PHASH_THRESHOLD:
                continue

            # 直方图融合（chunk 表优先，回退 JSON 字段；JSON NULL 显式回退空列表）
            if chunk_data:
                existing_histograms = [c["color_histogram"] for c in chunk_data if c.get("color_histogram")]
            else:
                existing_histograms = ef.get("color_histograms") or []

            combined_score = self._compute_fusion_score(
                median_distance, fingerprint.color_histograms, existing_histograms
            )

            if combined_score < DUPLICATE_THRESHOLD:
                continue

            # 滑动窗口时序匹配
            existing_chunk_objects = (
                chunk_data
                if chunk_data
                else [{"phash_binary": p, "start_time_ms": 0, "end_time_ms": 0} for p in existing_phashes]
            )
            segments = find_duplicate_segments(fingerprint.chunks, existing_chunk_objects)

            return {
                "duplicate": True,
                "duplicate_of": existing.id,
                "reason": "batch_phash_histogram_fusion",
                "similarity": combined_score,
                "duplicate_segments": [
                    {
                        "query_start_ms": s.query_start_ms,
                        "query_end_ms": s.query_end_ms,
                        "target_start_ms": s.target_start_ms,
                        "target_end_ms": s.target_end_ms,
                        "avg_distance": round(s.avg_distance, 2),
                    }
                    for s in segments
                ],
            }

        return None

    def compute_duplicate_rate(
        self,
        fingerprint: VideoFingerprint,
        project_id: str,
        current_video_id: str | None,
        session: Session,
        *,
        scope: str = "project",
        user_id: str = "",
    ) -> dict:
        """计算当前视频与已有视频的查重率百分比。

        新公式（双指标加权）：
        - frame_match_rate = 汉明距离 < PHASH_THRESHOLD 的帧数 / 总帧数
        - temporal_coverage_rate = 连续匹配片段总时长 / 视频总时长
        - duplicate_rate = (frame_match_rate * 0.4 + temporal_coverage_rate * 0.6) * 100

        visual_similarity = 0.7 * phash_sim + 0.3 * hist_sim（归一化到 0~1）

        对每个匹配视频都算，取最高 duplicate_rate。

        Args:
            fingerprint: 当前视频的指纹
            project_id: 项目 ID
            current_video_id: 当前视频 ID（排除自身，可为 None）
            session: 数据库会话
            scope: "project" 项目内（默认），"user" 跨项目全局
            user_id: 用户 ID（scope="user" 时使用）

        Returns:
            {
                "duplicate_rate": float,    # 0~100
                "visual_similarity": float,  # 0~1
                "match_count": int,          # 判定为重复的视频数
            }
        """
        video_repo = SQLAlchemyGeneratedVideoRepository(session)

        if scope == "user" and user_id:
            existing_videos = video_repo.list_by_user(user_id)
        else:
            existing_videos = video_repo.list_by_project(project_id)

        max_duplicate_rate = 0.0
        max_visual_similarity = 0.0
        match_count = 0

        total_duration_ms = fingerprint.duration if fingerprint.duration else 0

        for existing in existing_videos:
            if current_video_id and existing.id == current_video_id:
                continue
            if not existing.video_fingerprint:
                continue

            ef = existing.video_fingerprint

            # MD5 精确匹配 → 100%
            if fingerprint.md5 == ef.get("md5"):
                return {
                    "duplicate_rate": 100.0,
                    "visual_similarity": 1.0,
                    "match_count": 1,
                }

            # 跳过指纹质量差的视频（黑屏/纯色视频）
            existing_phashes_check = ef.get("keyframe_phashes", [])
            if self._is_bad_fingerprint(existing_phashes_check):
                logger.debug("Skipping bad fingerprint video %s in compute_duplicate_rate", existing.id)
                continue

            # 优先从分片表读取
            existing_phashes = []
            chunk_data = self._get_existing_chunks(existing.id, session)
            if chunk_data:
                existing_phashes = [c["phash_binary"] for c in chunk_data]
            else:
                existing_phashes = ef.get("keyframe_phashes", [])

            if not existing_phashes or not fingerprint.keyframe_phashes:
                continue

            min_distances = []
            for phash in fingerprint.keyframe_phashes:
                distances = [hamming_distance(phash, ep) for ep in existing_phashes]
                min_distances.append(min(distances))

            # frame_match_rate
            total_frames = len(min_distances)
            if total_frames == 0:
                continue
            matching_frames = sum(1 for d in min_distances if d < self.PHASH_THRESHOLD)
            frame_match_rate = matching_frames / total_frames

            # 帧匹配比例太低则跳过
            if frame_match_rate < 0.3:
                continue

            # temporal_coverage_rate via find_duplicate_segments
            existing_chunk_objects = (
                chunk_data
                if chunk_data
                else [{"phash_binary": p, "start_time_ms": 0, "end_time_ms": 0} for p in existing_phashes]
            )
            segments = find_duplicate_segments(fingerprint.chunks, existing_chunk_objects)

            if total_duration_ms > 0 and segments:
                covered_ms = sum(s.query_end_ms - s.query_start_ms for s in segments)
                temporal_coverage_rate = min(covered_ms / total_duration_ms, 1.0)
            else:
                temporal_coverage_rate = 0.0

            # duplicate_rate = 0.4 * frame_match_rate + 0.6 * temporal_coverage_rate
            dup_rate = (frame_match_rate * 0.4 + temporal_coverage_rate * 0.6) * 100

            # visual_similarity (融合相似度，归一化 0~1)
            median_distance = statistics.median(min_distances) if min_distances else 64
            if chunk_data:
                existing_histograms = [c["color_histogram"] for c in chunk_data if c.get("color_histogram")]
            else:
                # JSON NULL 显式回退空列表
                existing_histograms = ef.get("color_histograms") or []

            visual_sim = self._compute_fusion_score(median_distance, fingerprint.color_histograms, existing_histograms)

            # 判定是否为重复（融合分数超过阈值）
            if visual_sim >= DUPLICATE_THRESHOLD:
                match_count += 1

            if dup_rate > max_duplicate_rate:
                max_duplicate_rate = dup_rate
                max_visual_similarity = visual_sim

        return {
            "duplicate_rate": round(max(max_duplicate_rate, 0.0), 2),
            "visual_similarity": round(max_visual_similarity, 4),
            "match_count": match_count,
        }


def _save_fingerprint_chunks(
    fingerprint: VideoFingerprint,
    video_id: str,
    project_id: str,
    user_id: str,
    session: Session,
) -> None:
    """将指纹分片数据批量写入 video_fingerprint_chunks 表。幂等：已有数据时跳过。"""
    # 幂等检查：已有分片数据则跳过
    existing_count = (
        session.query(VideoFingerprintChunkModel).filter(VideoFingerprintChunkModel.video_id == video_id).count()
    )
    if existing_count > 0:
        logger.debug("Fingerprint chunks already exist for video %s (%d chunks), skipping", video_id, existing_count)
        return

    if not fingerprint.chunks:
        logger.warning("No chunks in fingerprint for video %s, skipping chunk save", video_id)
        return

    chunk_models = fingerprint.to_chunk_models(video_id, project_id, user_id)
    session.bulk_save_objects(chunk_models)
    logger.info("Saved %d fingerprint chunks for video %s", len(chunk_models), video_id)


@celery_app.task(bind=True, max_retries=3, name="worker.check_duplicate")
def check_duplicate_task(self: Task, generated_video_id: str) -> dict:
    """Celery task to check if generated video is a duplicate."""
    session = SessionLocal()
    temp_dir = tempfile.mkdtemp()

    try:
        video_repo = SQLAlchemyGeneratedVideoRepository(session)
        storage_service = get_storage_service()
        deduplicator = VideoDeduplicator()

        video = video_repo.get(generated_video_id)
        if video is None:
            raise ValueError(f"Generated video {generated_video_id} not found")

        local_path = os.path.join(temp_dir, f"{generated_video_id}.mp4")
        storage_service.download_file(
            f"projects/{video.project_id}/generated/{generated_video_id}/{generated_video_id}.mp4", local_path
        )

        fingerprint = deduplicator.compute_fingerprint(local_path)

        # 查重判定（跨项目全局 + 时长预过滤）
        duplicate_result = deduplicator.check_duplicate(
            fingerprint,
            video.project_id,
            session,
            scope="user",
            user_id=video.user_id,
            duration_sec=fingerprint.duration / 1000 if fingerprint.duration else 0,
        )

        video.video_fingerprint = fingerprint.to_dict()
        if duplicate_result:
            video.is_duplicate = True
            video.duplicate_of = duplicate_result["duplicate_of"]
        else:
            video.is_duplicate = False
            video.duplicate_of = None

        # 查重率计算（跨项目全局）
        rate_result = deduplicator.compute_duplicate_rate(
            fingerprint,
            video.project_id,
            generated_video_id,
            session,
            scope="user",
            user_id=video.user_id,
        )
        video.duplicate_rate = rate_result["duplicate_rate"]
        video.match_count = rate_result["match_count"]
        video.visual_similarity = rate_result["visual_similarity"]

        video_repo.update(video)

        # 写入分片表
        _save_fingerprint_chunks(fingerprint, generated_video_id, video.project_id, video.user_id, session)

        session.commit()

        logger.info(f"Duplicate check completed for video {generated_video_id}: is_duplicate={video.is_duplicate}")

        return {
            "ok": True,
            "video_id": generated_video_id,
            "is_duplicate": video.is_duplicate,
            "duplicate_of": video.duplicate_of,
            "duplicate_rate": video.duplicate_rate,
            "match_count": video.match_count,
            "visual_similarity": video.visual_similarity,
            "fingerprint": fingerprint.to_dict(),
        }
    except Exception as e:
        logger.error(f"Duplicate check failed for {generated_video_id}: {str(e)}")
        session.rollback()
        raise self.retry(exc=e, countdown=60) from e
    finally:
        session.close()
        import shutil

        shutil.rmtree(temp_dir, ignore_errors=True)