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  # 最小关键帧数
FINGERPRINT_SAMPLE_INTERVAL_SEC = 1.0  # 指纹采样间隔（秒）：密集均匀采样，保证两视频时序可对齐
FINGERPRINT_MAX_SAMPLES = 30  # 长视频采样数上限（超过后采样间隔自动放宽）
LONG_VIDEO_SEGMENT_SEC = 30  # 长视频每段秒数
LONG_VIDEO_DURATION_THRESHOLD_SEC = 180  # 3 分钟阈值
MIN_FRAMES_PER_SEGMENT = 2  # 长视频每段最少帧数

# ── 滑动窗口匹配常量（Issue #1702 重新校准） ─────────────────────
# 阈值经 staging 真实数据回归校准（2026-09-05，worker 容器内离线实验）：
#   - 同源成片对（20s/11s，各自 2-5% 随机边缘裁剪降重，1s 密集采样）：
#     全部帧对最小汉明距离 min=8，<=12 命中 10/31 帧（B->A 4/11）
#   - 异源成片对（4 个不同项目真实视频）：最小距离 24，<=16 命中 0 帧
# 8（#1658 旧值）会漏掉同源裁剪（自对照实验：同帧两次 2-5% 随机裁剪距离 4~10），
# 12 能检出同源/局部复用且与异源分布（>=24）间隔 12bit，无误报空间。
PHASH_THRESHOLD = 12
SEGMENT_MATCH_THRESHOLD = PHASH_THRESHOLD  # 片段匹配阈值与帧匹配统一（#1702：阈值常量统一来源）
MIN_CONSECUTIVE_MATCHES = 5  # 连续匹配默认门槛；短视频自适应 min(5, max(2, 分片数//2))
MAX_GAP = 2  # 允许的最大间隙帧数
NEIGHBOR_WINDOW = 1  # 分片时序对齐：允许 ±1 邻接偏移（1s 密集采样下即 ±1s，缓解切点不一致）

# ── 融合判定常量 ────────────────────────────────────────────────
PHASH_WEIGHT = 0.7  # pHash 权重
HISTOGRAM_WEIGHT = 0.3  # 直方图权重
MATCH_RATIO_THRESHOLD = 0.7  # 全片重复（is_duplicate）至少 70% 帧匹配
PARTIAL_COVERAGE_THRESHOLD = 0.5  # 局部复用覆盖率 >=50% 也判全片重复
DUPLICATE_THRESHOLD = 0.70  # 融合后相似度阈值

# ── 降重裁剪规避常量（Issue #1702） ─────────────────────────────
# 成片强制 2-5% random_edge_crop 降重只服务外部平台；自查重指纹取中心 90%
# 区域，使两次不同裁剪的同源画面 pHash 距离回到同分布。
FINGERPRINT_CENTER_CROP_RATIO = 0.90


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


def center_crop_frame(image: np.ndarray, ratio: float = FINGERPRINT_CENTER_CROP_RATIO) -> np.ndarray:
    """取画面中心 ratio 比例区域（裁除四边边缘）。

    查重指纹用：random_edge_crop 降重（2-5% 四边随机裁剪）会让同源画面 pHash
    位翻转 12-16，污染自查重（Issue #1702）。算 pHash/颜色直方图前先居中裁除
    边缘 10%，两次不同裁剪的同源画面中心区域基本重合，指纹不再被降重污染。
    降重只服务外部平台，不影响内部查重。
    """
    if image is None or image.size == 0:
        return image
    h, w = image.shape[:2]
    ch, cw = int(h * ratio), int(w * ratio)
    if ch <= 0 or cw <= 0 or (ch >= h and cw >= w):
        return image
    y0 = (h - ch) // 2
    x0 = (w - cw) // 2
    return image[y0 : y0 + ch, x0 : x0 + cw]


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 BGR color histogram for an image.

    Issue #1702: 每个通道独立做 NORM_L1 归一化（通道内 Σ=1，是概率分布），
    三通道拼接存储。Bhattacharyya 系数对拼接向量直接 Σ√(a*b) 会得到
    3 通道之和（范围 [0,3]，实测 ~14.9 是旧 L2 归一化的错误结果），
    消费方 _bhattacharyya_coefficient 按通道数平均归一到 [0,1]。
    """
    hist = []
    for i in range(3):
        h = cv2.calcHist([image], [i], None, [bins], [0, 256])
        h = cv2.normalize(h, h, norm_type=cv2.NORM_L1).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


def sample_fingerprint_timestamps(
    duration: float,
    *,
    interval_sec: float = FINGERPRINT_SAMPLE_INTERVAL_SEC,
    max_samples: int = FINGERPRINT_MAX_SAMPLES,
) -> list[float]:
    """指纹采样时间戳：固定间隔密集均匀采样（Issue #1702）。

    动态场景检测抽帧（#1659）在两个同源视频上会各自取到不同时刻，切点/取帧
    错位让对齐帧的 pHash 距离都很大（实测同源对最小距离 12 且配对时序错乱）。
    改为固定 1s 间隔均匀采样后，复用片段的帧时刻天然对齐，配合 ±1 邻接窗口
    即可检出同源/局部复用。长视频（>max_samples*interval）自动放宽间隔到
    duration/max_samples，保证分片数有上限。
    """
    if duration <= 0:
        return []
    step = interval_sec
    n_uniform = int(duration / step)
    if n_uniform > max_samples:
        step = duration / max_samples
    count = max(1, int(duration / step))
    return [step * (i + 0.5) for i in range(count)]


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


@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: Optional[int] = None,
    max_gap: int = MAX_GAP,
    neighbor_window: int = NEIGHBOR_WINDOW,
) -> list[DuplicateSegment]:
    """滑动窗口时序匹配：找出两组分片之间的重复片段（Issue #1702 重构）。

    算法：
    1. 构建 query×target 全量汉明距离矩阵；每个 query chunk 保留所有
       距离 <= match_threshold 的候选 target 分片（与帧匹配判定同一阈值）。
    2. 时序一致贪心对齐：沿 query 时序推进，run 内优先选择与上一匹配帧
       目标序号连贯（0 <= delta <= neighbor_window+1，允许 ±1 邻接窗口 /
       时序偏移对齐，缓解场景切割导致的切点、取帧错位）的候选；同距时
       偏好大索引，避免重复 hash 塌缩到 target 首帧。
    3. 连贯匹配中允许 <= max_gap 帧间隙桥接；断裂后另起新 run——天然
       支持局部片段复用（复用片段可出现在任意时序位置，各成独立片段）。
    4. 连续匹配帧数 >= min_consecutive 的 run 报为重复片段。短视频自适应：
       min_consecutive = min(5, max(2, len(query_chunks)//2))；n=1 时
       不形成片段，由调用方匹配帧回退兜底。

    Args:
        query_chunks: 查询视频的分片列表（FingerprintChunk 或 dict）
        target_chunks: 目标视频的分片列表
        match_threshold: 汉明距离匹配阈值（统一常量 PHASH_THRESHOLD）
        min_consecutive: 最少连续匹配帧数；None 时按短视频自适应
        max_gap: 允许的最大间隙帧数
        neighbor_window: 时序对齐允许的目标分片序号邻接窗口

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

    def _get(chunk, key):
        if isinstance(chunk, dict):
            return chunk[key]
        return getattr(chunk, key)

    n, m = len(query_chunks), len(target_chunks)
    q_ph = [_get(c, "phash_binary") for c in query_chunks]
    t_ph = [_get(c, "phash_binary") for c in target_chunks]

    # Step 1: 全量距离矩阵。每个 query chunk 保留所有 <= 阈值的候选 target，
    # 按距离升序；同距时小索引优先（取最早的对齐位置，贪心连贯推进时最保守，
    # 不会越过复用片段末端；重复 hash 的连续帧由 Step 2 的连贯性窗口约束）。
    candidates: list[list[tuple[int, int]]] = []  # 每 query 帧: [(target_idx, dist), ...]
    for i in range(n):
        dists = [hamming_distance(q_ph[i], t_ph[j]) for j in range(m)]
        cand = [(j, d) for j, d in enumerate(dists) if d <= match_threshold]
        cand.sort(key=lambda x: (x[1], x[0]))
        candidates.append(cand)

    # 短视频自适应连续匹配门槛（Issue #1702 工单公式）：
    # MIN_CONSECUTIVE_MATCHES = min(5, max(2, 分片数//2))。
    # n=1 时门槛为 2 不形成片段，由 _evaluate_candidate 的匹配帧回退
    # （temporal_coverage 按匹配帧占比估计）兜底检出，不回归。
    if min_consecutive is None:
        min_consecutive = min(MIN_CONSECUTIVE_MATCHES, max(2, n // 2))

    # Step 2: 时序一致贪心对齐。
    # run 内偏好与上一匹配帧目标序号连贯（0 <= delta <= neighbor_window+1，
    # 支持 ±1 邻接窗口/时序偏移对齐）的候选；无连贯候选时关闭旧 run。
    # 这天然支持局部片段复用：同一 query 视频中多个复用片段各自形成独立 run。
    frame_matches: list[tuple[bool, int, int]] = []
    runs: list[tuple[int, int]] = []
    run_start: Optional[int] = None
    run_last_t: Optional[int] = None
    gap_count = 0

    def _matching_count(a: int, b: int) -> int:
        return sum(1 for k in range(a, b + 1) if frame_matches[k][0])

    def _close_run(a: int, b: int) -> None:
        if b >= a and _matching_count(a, b) >= min_consecutive:
            runs.append((a, b))

    for i in range(n):
        cand = candidates[i]
        if run_last_t is None:
            chosen = cand[0] if cand else None
        else:
            chosen = next(
                (c for c in cand if 0 <= c[0] - run_last_t <= neighbor_window + 1),
                None,
            )

        if chosen is not None:
            tidx, dist = chosen
            frame_matches.append((True, dist, tidx))
            if run_start is None:
                run_start = i
            gap_count = 0
            run_last_t = tidx
        else:
            frame_matches.append((False, match_threshold + 1, -1))
            if run_start is not None:
                gap_count += 1
                if gap_count > max_gap:
                    # 非匹配帧从 i-gap_count+1 开始，run 结束于其前一帧
                    _close_run(run_start, i - gap_count)
                    run_start, run_last_t, gap_count = None, None, 0

    if run_start is not None:
        last_idx = n - 1
        while last_idx >= run_start and not frame_matches[last_idx][0]:
            last_idx -= 1
        _close_run(run_start, last_idx)

    # Step 3: 构建 DuplicateSegment
    segments: list[DuplicateSegment] = []
    for start, end in runs:
        target_indices = [frame_matches[k][2] for k in range(start, end + 1) if frame_matches[k][0]]
        t_min, t_max = min(target_indices), max(target_indices)
        avg_dist = sum(frame_matches[k][1] for k in range(start, end + 1) if frame_matches[k][0]) / len(target_indices)
        segments.append(
            DuplicateSegment(
                query_start_ms=_get(query_chunks[start], "start_time_ms"),
                query_end_ms=_get(query_chunks[end], "end_time_ms"),
                target_start_ms=_get(target_chunks[t_min], "start_time_ms"),
                target_end_ms=_get(target_chunks[t_max], "end_time_ms"),
                avg_distance=avg_dist,
            )
        )

    return segments


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


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

    # Issue #1702: 阈值统一来源为模块常量 PHASH_THRESHOLD（#1658 曾收紧到 8，
    # 后经 staging 真实同源/异源指纹分布重新校准，见 test_phash_threshold_calibration_1702）。
    PHASH_THRESHOLD = PHASH_THRESHOLD
    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
        # Issue #1702: 旧逻辑"所有 phash 完全相同即判黑屏"会误杀短视频——
        # 11s 视频只有几个不同镜头时，相邻 1s 采样帧可能 phash 完全一致（内容
        # 连续但非黑屏）。黑屏的特征是「大量帧全部无内容」，要求至少 8 帧
        # 且相同帧占比 >=80% 才判坏；短视频（<8 帧）只有真正单值时交给
        # _bhattacharyya/融合分兜底，不因"帧都一样"直接跳过。
        if len(phashes) < 8:
            return False
        unique = set(phashes)
        same_ratio = sum(1 for x in phashes if x == phashes[0]) / len(phashes)
        if len(unique) == 1 and same_ratio >= 0.8:
            return True
        # 多帧但所有唯一 phash 之间的汉明距离都极小（<3）且占比 >=80% → 近似黑屏
        phash_list = list(unique)
        if len(phash_list) >= 2 and same_ratio >= 0.8:
            all_distances = [
                hamming_distance(phash_list[i], phash_list[j])
                for i in range(len(phash_list))
                for j in range(i + 1, len(phash_list))
            ]
            if all_distances and max(all_distances) < 3:
                return True
        return False

    def compute_fingerprint(self, video_path: str) -> VideoFingerprint:
        """Compute video fingerprint using dense uniform sampling.

        Issue #1702: 使用 sample_fingerprint_timestamps() 固定 1s 间隔密集均匀
        采样（替代动态场景检测抽帧），保证两个同源视频复用片段的帧时刻天然
        对齐；每帧取中心 90% 区域（center_crop_frame）计算 pHash + color_histogram，
        绕开 random_edge_crop 降重裁剪污染；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. 固定间隔密集采样（Issue #1702：替代动态场景检测，保证跨视频时序对齐）
        keyframe_times = sample_fingerprint_timestamps(duration)

        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)

            # Issue #1702: pHash / 颜色直方图基于中心 90% 区域，绕开 random_edge_crop
            # 降重裁剪对指纹的污染（降重只服务外部平台，不污染自查重）。
            fp_frame = center_crop_frame(frame)
            phash = compute_phash(fp_frame)
            hist = compute_color_histogram(fp_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 系数（概率分布版，范围 [0,1]，1=完全相同）。

        Issue #1702: compute_color_histogram 输出 3 通道拼接、每通道独立 NORM_L1
        （单通道 Σ=1，三通道拼接向量 Σ=3）。旧实现直接 Σ√(a*b) 对三通道拼接向量
        算出 ~3（旧 L2 归一化更是算出 ~14.9），不是合法的概率系数。
        这里按两个直方图各自的总量归一：BC = Σ√(a*b) / √(Σa·Σb)。
        - 单通道概率分布（Σa=Σb=1）：分母 1，与旧测试/教科书定义一致；
        - 三通道拼接（Σa=Σb=3）：分母 3，结果在 [0,1]。
        """
        min_len = min(len(hist_a), len(hist_b))
        a = [max(0.0, float(x)) for x in hist_a[:min_len]]
        b = [max(0.0, float(x)) for x in hist_b[:min_len]]
        # max(0.0, ...) 防御上游异常负值导致 sqrt domain error
        coeff = sum(math.sqrt(ai * bi) for ai, bi in zip(a, b, strict=False))
        norm = math.sqrt(sum(a) * sum(b))
        return float(coeff / norm) if norm > 0 else 0.0

    @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

    @staticmethod
    def _evaluate_candidate(
        fingerprint: VideoFingerprint,
        existing_phashes: list[str],
        existing_histograms: list,
        existing_chunk_objects: list,
        *,
        query_duration_sec: float,
    ) -> dict:
        """评估新视频指纹与单个候选视频的相似度（Issue #1702 共享逻辑）。

        指标：
        - min_distances / frame_match_rate：每个新分片到候选视频全局最近邻的汉明距离，
          分母取两视频分片数的较小值（支持局部片段复用：短视频复用长视频片段时不被长视频分母稀释）。
        - temporal_coverage：时序一致连续匹配片段总时长 / 新视频时长（局部复用主指标）。
        - fusion：pHash 中位数距离 + 颜色直方图的加权融合分。

        Returns:
            {frame_match_rate, temporal_coverage, segments, median_distance,
             fusion, matching_frames, min_distances}
        """
        query_phashes = fingerprint.keyframe_phashes or []
        if not query_phashes or not existing_phashes:
            return {
                "frame_match_rate": 0.0,
                "temporal_coverage": 0.0,
                "segments": [],
                "median_distance": 64,
                "fusion": 0.0,
                "matching_frames": 0,
                "min_distances": [],
            }

        min_distances = [min(hamming_distance(ph, ep) for ep in existing_phashes) for ph in query_phashes]
        matching_frames = sum(1 for d in min_distances if d <= PHASH_THRESHOLD)
        # 分母取 min(两视频分片数)：局部复用时（如 B 的 5 片复用 A 9 片中的若干片）
        # 命中帧占比不因候选视频更长而被稀释。
        frame_match_rate = matching_frames / min(len(query_phashes), len(existing_phashes))

        segments = find_duplicate_segments(fingerprint.chunks, existing_chunk_objects)
        duration_ms = query_duration_sec * 1000 if query_duration_sec else 0
        if duration_ms > 0 and segments:
            covered_ms = sum(s.query_end_ms - s.query_start_ms for s in segments)
            temporal_coverage = min(covered_ms / duration_ms, 1.0)
        elif matching_frames > 0:
            # 无连续片段（时序连贯性不足）时，按匹配帧占比估计覆盖：
            # 密集 1s 采样下每个分片≈1s 等权时间片，匹配帧数≈命中秒数。
            temporal_coverage = min(frame_match_rate, 1.0)
        else:
            temporal_coverage = 0.0

        median_distance = statistics.median(min_distances) if min_distances else 64
        fusion = VideoDeduplicator._compute_fusion_score(
            median_distance, fingerprint.color_histograms, existing_histograms
        )

        return {
            "frame_match_rate": frame_match_rate,
            "temporal_coverage": temporal_coverage,
            "segments": segments,
            "median_distance": median_distance,
            "fusion": fusion,
            "matching_frames": matching_frames,
            "min_distances": min_distances,
        }

    def check_duplicate(
        self,
        fingerprint: VideoFingerprint,
        project_id: str,
        session: Session,
        *,
        scope: str = "project",
        user_id: str = "",
        duration_sec: float = 0,
        exclude_video_id: str | None = None,
    ) -> 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%
            exclude_video_id: 排除的视频 ID（查重自身时用）。recompute-dedup
                重算时视频记录已存在，不排除会自匹配（距离 0 分最高）导致
                duplicate_of 指向自己（Issue #1702 连带修复）。

        Returns:
            重复信息字典（含 duplicate, duplicate_of, reason, similarity, duplicate_segments），
            或 None 表示未找到重复。
        """
        video_repo = SQLAlchemyGeneratedVideoRepository(session)
        if scope == "user" and user_id:
            # Issue #1702: 不做 ±15% 时长预过滤。旧逻辑按 duration_sec 缩小候选窗口，
            # 但局部片段复用的两个视频时长必然不同（证据视频 20s vs 11s，差 42%），
            # ±15% 窗口让同源视频互相不可见 → is_duplicate 恒 False。
            # 全量遍历同用户视频（与 compute_duplicate_rate 口径一致），异源视频由
            # fusion/temporal_coverage 阈值天然过滤（校准：异源最小汉明距离 24）。
            existing_videos = video_repo.list_by_user(user_id)
        else:
            existing_videos = video_repo.list_by_project(project_id)

        best_score = 0.0
        best_result: Optional[dict] = None

        for existing in existing_videos:
            # 排除自身（recompute 时当前视频已在候选列表里，否则自匹配距离 0 必最高分）
            if exclude_video_id and existing.id == exclude_video_id:
                continue
            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

            # 直方图 / 分片对象（chunk 表优先，回退 JSON 字段；JSON NULL 显式回退空列表）
            if chunk_data:
                existing_histograms = [c["color_histogram"] for c in chunk_data if c.get("color_histogram")]
                existing_chunk_objects = chunk_data
            else:
                existing_histograms = ef.get("color_histograms") or []
                existing_chunk_objects = [
                    {"phash_binary": pp, "start_time_ms": 0, "end_time_ms": 0} for pp in existing_phashes
                ]

            # Issue #1702: 统一评估每个候选（含局部片段复用），不再用
            # "frame_match_rate<0.7 整条跳过" 的硬门槛——局部复用（如 B 结尾 2s
            # ≈ A 中间 2s）帧比例天然低，但 coverage 能检出。
            ev = self._evaluate_candidate(
                fingerprint,
                existing_phashes,
                existing_histograms,
                existing_chunk_objects,
                query_duration_sec=fingerprint.duration,
            )
            logger.debug(
                "check_duplicate candidate=%s min_distances=%s frame_match_rate=%.3f "
                "temporal_coverage=%.3f median=%.1f fusion=%.3f segments=%d",
                existing.id,
                ev["min_distances"],
                ev["frame_match_rate"],
                ev["temporal_coverage"],
                ev["median_distance"],
                ev["fusion"],
                len(ev["segments"]),
            )

            # 全片重复判定：融合分过阈 且（帧匹配比例 >=70% 或 局部覆盖 >=50%）
            is_full_duplicate = ev["fusion"] >= DUPLICATE_THRESHOLD and (
                ev["frame_match_rate"] >= MATCH_RATIO_THRESHOLD or ev["temporal_coverage"] >= PARTIAL_COVERAGE_THRESHOLD
            )
            if is_full_duplicate and ev["fusion"] > best_score:
                best_score = ev["fusion"]
                best_result = {
                    "duplicate": True,
                    "duplicate_of": existing.id,
                    "reason": "phash_histogram_fusion",
                    "similarity": ev["fusion"],
                    "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 ev["segments"]
                    ],
                }

        if best_result:
            return best_result
        logger.info(
            "check_duplicate no match (project=%s scope=%s): %d candidates evaluated, best_fusion=%.3f",
            project_id,
            scope,
            len(existing_videos),
            best_score,
        )
        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)

        best_score = 0.0
        best_result: Optional[dict] = None

        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

            if chunk_data:
                existing_histograms = [c["color_histogram"] for c in chunk_data if c.get("color_histogram")]
                existing_chunk_objects = chunk_data
            else:
                existing_histograms = ef.get("color_histograms") or []
                existing_chunk_objects = [
                    {"phash_binary": pp, "start_time_ms": 0, "end_time_ms": 0} for pp in existing_phashes
                ]

            ev = self._evaluate_candidate(
                fingerprint,
                existing_phashes,
                existing_histograms,
                existing_chunk_objects,
                query_duration_sec=fingerprint.duration,
            )
            logger.debug(
                "check_batch_duplicate candidate=%s min_distances=%s frame_match_rate=%.3f "
                "temporal_coverage=%.3f median=%.1f fusion=%.3f segments=%d",
                existing.id,
                ev["min_distances"],
                ev["frame_match_rate"],
                ev["temporal_coverage"],
                ev["median_distance"],
                ev["fusion"],
                len(ev["segments"]),
            )

            is_full_duplicate = ev["fusion"] >= DUPLICATE_THRESHOLD and (
                ev["frame_match_rate"] >= MATCH_RATIO_THRESHOLD or ev["temporal_coverage"] >= PARTIAL_COVERAGE_THRESHOLD
            )
            if is_full_duplicate and ev["fusion"] > best_score:
                best_score = ev["fusion"]
                best_result = {
                    "duplicate": True,
                    "duplicate_of": existing.id,
                    "reason": "batch_phash_histogram_fusion",
                    "similarity": ev["fusion"],
                    "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 ev["segments"]
                    ],
                }

        if best_result:
            return best_result
        logger.info("check_batch_duplicate no match (batch=%s): best_fusion=%.3f", batch_id, best_score)
        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
        evaluated = 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

            # 直方图 / 分片对象（chunk 表优先，回退 JSON 字段；JSON NULL 显式回退空列表）
            if chunk_data:
                existing_histograms = [c["color_histogram"] for c in chunk_data if c.get("color_histogram")]
                existing_chunk_objects = chunk_data
            else:
                existing_histograms = ef.get("color_histograms") or []
                existing_chunk_objects = [
                    {"phash_binary": pp, "start_time_ms": 0, "end_time_ms": 0} for pp in existing_phashes
                ]

            # Issue #1702: 统一评估；frame_match_rate 分母为 min(两视频分片数)，
            # temporal_coverage 时长量纲在 _evaluate_candidate 内统一为毫秒。
            ev = self._evaluate_candidate(
                fingerprint,
                existing_phashes,
                existing_histograms,
                existing_chunk_objects,
                query_duration_sec=fingerprint.duration,
            )
            evaluated += 1
            logger.debug(
                "compute_duplicate_rate candidate=%s min_distances=%s frame_match_rate=%.3f "
                "temporal_coverage=%.3f median=%.1f fusion=%.3f segments=%d",
                existing.id,
                ev["min_distances"],
                ev["frame_match_rate"],
                ev["temporal_coverage"],
                ev["median_distance"],
                ev["fusion"],
                len(ev["segments"]),
            )

            # Issue #1702: 去掉 "frame_match_rate<0.3 整条跳过" 硬门槛——
            # 局部片段复用帧比例天然低；coverage 为主指标，0 匹配自然得 0 分。
            # duplicate_rate = 0.4 * frame_match_rate + 0.6 * temporal_coverage
            dup_rate = (min(ev["frame_match_rate"], 1.0) * 0.4 + ev["temporal_coverage"] * 0.6) * 100

            # 全片重复计数与 check_duplicate 判定口径一致
            if ev["fusion"] >= DUPLICATE_THRESHOLD and (
                ev["frame_match_rate"] >= MATCH_RATIO_THRESHOLD or ev["temporal_coverage"] >= PARTIAL_COVERAGE_THRESHOLD
            ):
                match_count += 1

            if dup_rate > max_duplicate_rate:
                max_duplicate_rate = dup_rate
                max_visual_similarity = ev["fusion"]

        logger.info(
            "compute_duplicate_rate done (project=%s scope=%s): evaluated=%d max_rate=%.2f%% "
            "max_visual_sim=%.3f matches=%d",
            project_id,
            scope,
            evaluated,
            max_duplicate_rate,
            max_visual_similarity,
            match_count,
        )
        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 表。幂等：已有数据时跳过。"""
    if not fingerprint.chunks:
        logger.warning("No chunks in fingerprint for video %s, skipping chunk save", video_id)
        return

    # Issue #1702: recompute-dedup 重算时指纹算法已变（中心裁剪 + 新阈值），
    # 旧分片必须替换而非跳过（旧实现"有数据就跳过"导致重算不刷新分片表）。
    deleted = (
        session.query(VideoFingerprintChunkModel)
        .filter(VideoFingerprintChunkModel.video_id == video_id)
        .delete(synchronize_session=False)
    )
    if deleted:
        logger.info("Replaced %d stale fingerprint chunks for video %s", deleted, video_id)

    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")
        # Issue #1702: recompute 走的是 OSS 重新下载路径（正常生成流程用本地渲染文件，
        # 不经此任务）。成片真实 OSS key 是生成时的
        # generated/projects/{pid}/tasks/{task_id}/rendered_*.mp4（见 generation.py
        # _upload_and_record），旧代码硬编码 projects/{pid}/generated/{vid}/{vid}.mp4
        # 这个从不存在的 key，导致所有 recompute 任务下载 404、查重数据永远无法重算。
        # 优先从 file_url 解析真实 key，旧 key 模式仅作回退。
        download_key = getattr(video, "file_url", "") or ""
        if not download_key:
            download_key = f"projects/{video.project_id}/generated/{generated_video_id}/{generated_video_id}.mp4"
            logger.warning("video %s has no file_url, falling back to legacy key %s", generated_video_id, download_key)
        storage_service.download_file(download_key, 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,
            # Issue #1702: fingerprint.duration 单位已经是秒，旧代码 /1000 导致
            # ±15% 时长预过滤窗口缩到 ~0.013s，scope=user 的跨项目查重永远返回 None。
            duration_sec=fingerprint.duration if fingerprint.duration else 0,
            exclude_video_id=generated_video_id,
        )

        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)