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

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

import hashlib
import logging
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 = 10
    HISTOGRAM_THRESHOLD = 0.85

    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]
        return float(sum(np.sqrt(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

    def check_duplicate(self, fingerprint: VideoFingerprint, project_id: str, session: Session) -> Optional[dict]:
        """检查视频是否与项目中已有视频重复。

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

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

        Args:
            fingerprint: 待检测视频的指纹
            project_id: 项目 ID，仅在同一项目内搜索
            session: 数据库会话

        Returns:
            重复信息字典（含 duplicate, duplicate_of, reason, similarity, duplicate_segments），
            或 None 表示未找到重复。
        """
        video_repo = SQLAlchemyGeneratedVideoRepository(session)
        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}

            # 优先从分片表读取已有视频的分片 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 < 0.7:
                continue

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

            # 直方图融合
            existing_histograms = []
            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", [])

            phash_similarity = 1.0 - (median_distance / 64)
            hist_similarity = (
                self._compute_histogram_similarity(fingerprint.color_histograms, existing_histograms)
                if existing_histograms
                else 0.5
            )
            combined_score = 0.7 * phash_similarity + 0.3 * hist_similarity

            # DUPLICATE_THRESHOLD from module level
            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,
    ) -> Optional[dict]:
        """检查视频是否与同批次内其他视频重复。

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

        Args:
            fingerprint: 待检测视频的指纹
            batch_id: 批次 ID
            current_video_id: 当前视频 ID（排除自身）
            session: 数据库会话

        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 = []
            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 < 0.7:
                continue

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

            # 直方图融合
            existing_histograms = []
            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", [])

            phash_similarity = 1.0 - (median_distance / 64)
            hist_similarity = (
                self._compute_histogram_similarity(fingerprint.color_histograms, existing_histograms)
                if existing_histograms
                else 0.5
            )
            combined_score = 0.7 * phash_similarity + 0.3 * hist_similarity

            # DUPLICATE_THRESHOLD from module level
            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,
        *,
        user_id: str = "",
    ) -> float:
        """计算当前视频与用户库内已有视频的最高相似度百分比。

        优先按 user_id 全局比较（跨项目），user_id 为空时回退到项目级比较。
        遍历最近 200 个其他有指纹的视频，对每个计算融合相似度：
        - MD5 精确匹配 → 100%
        - pHash + 直方图融合 → 0.7 * phash_sim + 0.3 * hist_sim
        取最高值作为 duplicate_rate（0~100）。
        如果没有其他视频可比较，返回 0.0。

        Args:
            fingerprint: 当前视频的指纹
            project_id: 项目 ID（user_id 为空时的回退范围）
            current_video_id: 当前视频 ID（排除自身，可为 None）
            session: 数据库会话
            user_id: 用户 ID（优先按用户全局比较）

        Returns:
            duplicate_rate: 0~100 的浮点数
        """
        from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel

        # 优先按 user_id 全局比较（跨项目），否则回退到项目级
        if user_id:
            query = session.query(GeneratedVideoModel).filter(
                GeneratedVideoModel.user_id == user_id,
            )
            logger.debug("compute_duplicate_rate: user-level scope user_id=%s", user_id)
        else:
            query = session.query(GeneratedVideoModel).filter(
                GeneratedVideoModel.project_id == project_id,
            )
            logger.debug("compute_duplicate_rate: project-level fallback project_id=%s", project_id)

        # 排除当前视频自身
        if current_video_id:
            query = query.filter(GeneratedVideoModel.id != current_video_id)

        recent_models = query.order_by(GeneratedVideoModel.generated_at.desc()).limit(200).all()
        video_repo = SQLAlchemyGeneratedVideoRepository(session)
        existing_videos = [video_repo._to_domain(m) for m in recent_models]

        max_similarity = 0.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 100.0

            # 优先从分片表读取
            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))

            # 帧匹配比例检查
            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 < 0.7:
                continue

            median_distance = statistics.median(min_distances) if min_distances else 64

            # 直方图融合
            existing_histograms = []
            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", [])

            phash_similarity = (1.0 - median_distance / 64) * 100
            hist_similarity = (
                self._compute_histogram_similarity(fingerprint.color_histograms, existing_histograms) * 100
                if existing_histograms
                else 50.0
            )
            combined_score = 0.7 * phash_similarity + 0.3 * hist_similarity
            max_similarity = max(max_similarity, combined_score)

        return round(max(max_similarity, 0.0), 2)


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)

        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

        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,
            "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)