fix(dedup): recompute任务从file_url派生OSS下载key，修复重算404 (#1702) (#1705)

Co-authored-by: xiaoxia <dev@xiaoxiajianji.com>
Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>

apps/api/app/api/routes/generation_preview.py

@@ -23,6 +23,7 @@ from app.dependencies import (
     get_generation_task_repository,
 )
 from app.schemas.generation_task import (
+    BatchPreviewGenerationTaskResponse,
     CreatePreviewGenerationTaskRequest,
     PreviewGenerationTaskResponse,
 )
@@ -193,11 +194,19 @@ def _to_preview_response(task, generated_videos: list | None = None) -> PreviewG
     if started_at and completed_at:
         generate_duration = (completed_at - started_at).total_seconds()
 
+    title_cfg = getattr(task, "title_config", None)
+    title_cfg = title_cfg if isinstance(title_cfg, dict) else {}
+    extra_meta = getattr(task, "extra_meta", None)
+    extra_meta = extra_meta if isinstance(extra_meta, dict) else {}
+    voice_library_id = getattr(task, "voice_library_id", "") or ""
+    if not isinstance(voice_library_id, str):
+        voice_library_id = str(voice_library_id) if voice_library_id else ""
     return PreviewGenerationTaskResponse(
         task_id=task.id,
         status=task.status.value if hasattr(task.status, "value") else str(task.status),
         progress=float(task.progress or 0.0),
         is_preview=bool(getattr(task, "is_preview", True)),
+        variant_index=int(extra_meta.get("variant_index", 0) or 0),
         resolution=getattr(task, "resolution", "") or "",
         video_url=video_url,
         duration=duration,
@@ -206,6 +215,8 @@ def _to_preview_response(task, generated_videos: list | None = None) -> PreviewG
         transition_count=transition_count,
         material_usage=material_usage,
         error_message=task.error_message or "",
+        title_text=str(title_cfg.get("text", "") or ""),
+        voice_library_id=voice_library_id,
         created_at=task.created_at,
         started_at=started_at,
         finished_at=completed_at,
@@ -213,45 +224,95 @@ def _to_preview_response(task, generated_videos: list | None = None) -> PreviewG
     )
 
 
-@router.post("/preview", response_model=PreviewGenerationTaskResponse, status_code=201)
+def _resolve_preview_edit_plan_id(
+    *,
+    request: CreatePreviewGenerationTaskRequest,
+    task,
+    db: Session,
+    user_id: str,
+) -> str:
+    """确定任务关联的编辑计划ID：优先前端传入，否则按 template_id+user 兜底查找。"""
+    if task.source_edit_plan_id:
+        return task.source_edit_plan_id
+    if not request.template_id:
+        return ""
+    try:
+        from packages.adapters.sqlalchemy_impl.edit_plan_repository import (
+            SQLAlchemyEditPlanRepository,
+        )
+
+        _plan_repo = SQLAlchemyEditPlanRepository(db)
+        _plans = _plan_repo.list_by_template(request.template_id, limit=20)
+        for _p in _plans:
+            if (_p.created_by_user_id or "") == user_id:
+                logger.info(
+                    "[预览生成] 自动关联编辑计划: task_id=%s plan_id=%s",
+                    task.id,
+                    _p.id,
+                )
+                return _p.id
+    except Exception:
+        logger.warning(
+            "[预览生成] 查找关联编辑计划失败（不影响主流程）: task_id=%s",
+            task.id,
+            exc_info=True,
+        )
+    return ""
+
+
+def _variant_value(values: list[str], index: int, fallback: str = "") -> str:
+    """从变体数组中取值：长度1=共用，长度>N=按索引，空数组=回退 fallback。"""
+    if not values:
+        return fallback
+    if len(values) == 1:
+        return values[0]
+    return values[index] if index < len(values) else fallback
+
+
+@router.post("/preview", response_model=BatchPreviewGenerationTaskResponse, status_code=201)
 def create_preview_generation_task(
     request: CreatePreviewGenerationTaskRequest,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     generation_task_repository=Depends(get_generation_task_repository),
     db: Session = Depends(get_db_session),
     asset_repo=Depends(get_asset_repository),
-) -> PreviewGenerationTaskResponse:
-    """创建预览生成任务。
+) -> BatchPreviewGenerationTaskResponse:
+    """创建预览生成任务（支持批量）。
 
-    预览渲染品质与正式生成一致（1080p, CRF 23, medium preset），确认生成时可直接复用预览产物。
-
-    Args:
-        request: 预览任务创建请求（template_id + asset_ids 等）
+    preview_count=1 时行为与旧版完全一致（创建 1 个任务）；
+    preview_count=N 时一次创建 N 个独立变体任务：
+    - 每个变体克隆独立编辑计划（独立 clips、独立随机素材起点），N 个预览内容互不相同
+    - 每个变体拥有独立 task_id / 状态 / 预览视频 URL，前端按 task_id 分别轮询
+    - 标题样式（font/color/position 等）全局共用；标题文字/配音/封面可按变体独立
+      （titles[] / voice_library_ids[] / cover_urls[]，长度1=共用，长度N=独立）
 
     Returns:
-        201 + 预览任务详情
+        201 + 变体任务数组 {items: [...], total: N}
     """
     user_id = authenticated_user.user.id
+    count = max(1, request.preview_count)
     logger.info(
         "[预览生成] 接收请求: user_id=%s, template_id=%s, asset_count=%d, preview_count=%d",
         user_id,
         request.template_id,
         len(request.asset_ids),
-        request.preview_count,
+        count,
     )
 
-    # 预检查队列限流
+    # 预检查队列限流（按变体总数计）
     try:
         user_pending = generation_task_repository.count_pending_by_user(user_id)
         global_pending = generation_task_repository.count_pending_total()
-        if user_pending + 1 > USER_PENDING_LIMIT:
-            raise UserPendingLimitExceeded(user_id=user_id, pending_count=user_pending + 1, limit=USER_PENDING_LIMIT)
-        if global_pending + 1 > GLOBAL_PENDING_LIMIT:
-            raise GlobalQueueFull(pending_count=global_pending + 1, limit=GLOBAL_PENDING_LIMIT)
+        if user_pending + count > USER_PENDING_LIMIT:
+            raise UserPendingLimitExceeded(
+                user_id=user_id, pending_count=user_pending + count, limit=USER_PENDING_LIMIT
+            )
+        if global_pending + count > GLOBAL_PENDING_LIMIT:
+            raise GlobalQueueFull(pending_count=global_pending + count, limit=GLOBAL_PENDING_LIMIT)
     except UserPendingLimitExceeded as e:
         raise HTTPException(
             status_code=429,
-            detail=f"您的待处理任务过多（当前 {e.pending_count - 1}/{e.limit}），请等待后再提交",
+            detail=f"您的待处理任务过多（当前 {e.pending_count - count}/{e.limit}，本次提交 {count} 个），请等待后再提交",
         ) from e
     except GlobalQueueFull as e:
         raise HTTPException(
@@ -273,14 +334,11 @@ def create_preview_generation_task(
                 w, h = int(parts[0]), int(parts[1])
                 base = 1920
                 if w < h:
-                    # 竖屏
                     output_width = round(base * w / h)
                     output_height = base
                 else:
-                    # 横屏
                     output_width = base
                     output_height = round(base * h / w)
-                # 对齐到偶数
                 output_width = output_width - output_width % 2
                 output_height = output_height - output_height % 2
             except (ValueError, ZeroDivisionError):
@@ -289,42 +347,71 @@ def create_preview_generation_task(
 
     logger.info(
         "[预览生成] 分辨率: video_ratio=%s → %s (%dx%d)",
-        video_ratio, resolution, output_width, output_height,
+        video_ratio,
+        resolution,
+        output_width,
+        output_height,
     )
 
-    # 从模板读取 editing_mode / mode 作为 strategy_id（渲染 pipeline 的 mode 参数）
     strategy_id = _resolve_strategy_id_from_template(request.template_id, db, user_id)
-
-    title_config = request.title_config or {}
+    base_title_config = request.title_config or {}
 
     use_case = CreateGenerationTaskUseCase(generation_task_repository)
 
+    # ── 预创建第一个任务，仅用于解析源编辑计划（不落库为最终任务）──
+    # 先创建一个临时任务拿到 task 对象上下文，实际 N 个任务在循环中统一创建；
+    # 为保持与旧版一致的源 plan 解析逻辑，先创建任务0、解析源 plan，
+    # 再预克隆 N 个变体 plan，最后重建任务关联。
+    # 简化实现：直接创建全部任务，plan 关联在创建后、入队前完成。
+
+    created_tasks: list = []
+    variant_plan_ids: list[str] = []  # 每个变体最终关联的 plan_id（按变体顺序）
+
     try:
-        task = use_case.execute(
-            CreateGenerationTaskCommand(
-                project_id="",
-                asset_library_id="",
-                strategy_id=strategy_id,
-                voice_library_id=request.voice_library_id,
-                template_id=request.template_id,
-                asset_ids=list(request.asset_ids),
-                title_ids=list(request.title_ids),
-                voice_ids=list(request.voice_ids),
-                created_by_user_id=user_id,
-                source_edit_plan_id=request.source_edit_plan_id,
-                asset_select_mode="",
-                batch_id="",
-                video_title=request.video_title,
-                resolution=resolution,
-                bgm_config=request.bgm_config or {},
-                auto_retry_enabled=False,
-                auto_retry_max=0,
-                is_preview=True,
-                title_config=title_config,
-                output_width=output_width,
-                output_height=output_height,
+        for variant_index in range(count):
+            # 变体独立标题文字：titles[] 覆盖 title_config.text
+            variant_title_text = _variant_value(request.titles, variant_index, "")
+            variant_title_config = dict(base_title_config)
+            if variant_title_text.strip():
+                variant_title_config["text"] = variant_title_text.strip()
+
+            # 变体独立配音
+            variant_voice_library_id = _variant_value(
+                request.voice_library_ids, variant_index, request.voice_library_id
             )
-        )
+
+            task = use_case.execute(
+                CreateGenerationTaskCommand(
+                    project_id="",
+                    asset_library_id="",
+                    strategy_id=strategy_id,
+                    voice_library_id=variant_voice_library_id,
+                    template_id=request.template_id,
+                    asset_ids=list(request.asset_ids),
+                    title_ids=list(request.title_ids),
+                    voice_ids=list(request.voice_ids),
+                    created_by_user_id=user_id,
+                    source_edit_plan_id=request.source_edit_plan_id,
+                    asset_select_mode="",
+                    batch_id="",
+                    video_title=request.video_title,
+                    resolution=resolution,
+                    bgm_config=request.bgm_config or {},
+                    auto_retry_enabled=False,
+                    auto_retry_max=0,
+                    is_preview=True,
+                    title_config=variant_title_config,
+                    output_width=output_width,
+                    output_height=output_height,
+                )
+            )
+            task.extra_meta["variant_index"] = variant_index
+
+            # 解析源编辑计划（前端传入或按模板兜底查找）
+            source_plan_id = _resolve_preview_edit_plan_id(request=request, task=task, db=db, user_id=user_id)
+            task.source_edit_plan_id = source_plan_id
+            generation_task_repository.update(task)
+            created_tasks.append(task)
     except ValueError as e:
         logger.warning("[预览生成] 创建失败: %s", e)
         raise HTTPException(status_code=400, detail=str(e)) from e
@@ -332,93 +419,121 @@ def create_preview_generation_task(
         logger.error("[预览生成] 创建失败: %s", e, exc_info=True)
         raise HTTPException(status_code=500, detail="创建预览生成任务失败，请稍后再试") from e
 
-    # 关联编辑计划：如果前端未传 source_edit_plan_id，通过 template_id + user_id 查找
-    if not task.source_edit_plan_id and request.template_id:
-        try:
-            from packages.adapters.sqlalchemy_impl.edit_plan_repository import (
-                SQLAlchemyEditPlanRepository,
-            )
-
-            _plan_repo = SQLAlchemyEditPlanRepository(db)
-            _plans = _plan_repo.list_by_template(request.template_id, limit=20)
-            for _p in _plans:
-                if (_p.created_by_user_id or "") == user_id:
-                    task.source_edit_plan_id = _p.id
-                    generation_task_repository.update(task)
-                    logger.info(
-                        "[预览生成] 自动关联编辑计划: task_id=%s plan_id=%s",
-                        task.id,
-                        _p.id,
-                    )
-                    break
-        except Exception:
-            logger.warning(
-                "[预览生成] 查找关联编辑计划失败（不影响主流程）: task_id=%s",
-                task.id,
-                exc_info=True,
-            )
-
-    # 每条预览都关联独立克隆 plan：多预览前端为 N 次并发调用，若共用同一 plan
-    # 则 N 条预览片段完全相同；克隆时片段起点按持久化历史区间重算（含受控复用），
-    # 保证各预览版本内容不同
-    if task.source_edit_plan_id:
+    # ── 克隆独立变体 plan：N 个预览全部克隆（预览不污染源 plan）──
+    # 源 plan 不存在（无编辑历史）时各任务走自身随机选片流程，不克隆。
+    source_plan_id = created_tasks[0].source_edit_plan_id if created_tasks else ""
+    if source_plan_id:
         try:
             from app.services.edit_plan_service import EditPlanService
 
             _plan_svc = EditPlanService(db)
-            _preview_plan = _plan_svc.clone_plan_for_variant(
-                task.source_edit_plan_id,
-                created_by_user_id=user_id,
-                name_suffix="预览变体",
-            )
-            task.source_edit_plan_id = _preview_plan.id
-            generation_task_repository.update(task)
-            logger.info(
-                "[预览生成] 预览关联独立克隆 plan: task_id=%s clone_plan_id=%s",
-                task.id,
-                _preview_plan.id,
-            )
-        except Exception as clone_err:
-            # 不退回共用原 plan（否则多条预览内容相同，违反去重诉求）：
-            # 标记任务失败并中断，前端可重新发起预览
-            logger.error(
-                "[预览生成] 克隆预览变体 plan 失败，任务标记失败: task_id=%s error=%s",
-                task.id,
-                clone_err,
-                exc_info=True,
-            )
-            _mark_task_failed(generation_task_repository, task, "预览变体计划创建失败")
+            for variant_index in range(count):
+                last_err: Exception | None = None
+                variant_plan = None
+                for _attempt in range(2):  # 1 次重试，抗 DB 瞬时抖动
+                    try:
+                        variant_plan = _plan_svc.clone_plan_for_variant(
+                            source_plan_id,
+                            created_by_user_id=user_id,
+                            name_suffix=f"预览变体{variant_index + 1}" if count > 1 else "预览变体",
+                        )
+                        break
+                    except Exception as clone_err:  # noqa: PERF203
+                        last_err = clone_err
+                        logger.warning(
+                            "[预览生成] 克隆变体 plan 失败(尝试%d/2): variant=%d error=%s",
+                            _attempt + 1,
+                            variant_index,
+                            clone_err,
+                            exc_info=True,
+                        )
+                if variant_plan is None:
+                    logger.error(
+                        "[预览生成] 克隆预览变体 plan 重试仍失败: variant=%d source=%s",
+                        variant_index,
+                        source_plan_id,
+                        exc_info=last_err,
+                    )
+                    # 标记已创建任务失败
+                    for t in created_tasks:
+                        _mark_task_failed(generation_task_repository, t, "预览变体计划创建失败")
+                    raise HTTPException(
+                        status_code=500,
+                        detail="创建预览任务失败：无法生成独立剪辑计划，请重试",
+                    ) from last_err
+                variant_plan_ids.append(variant_plan.id)
+        except HTTPException:
+            raise
+        except Exception as e:
+            logger.error("[预览生成] 克隆变体 plan 异常: %s", e, exc_info=True)
+            for t in created_tasks:
+                _mark_task_failed(generation_task_repository, t, "预览变体计划创建失败")
             raise HTTPException(
                 status_code=500,
                 detail="创建预览任务失败：无法生成独立剪辑计划，请重试",
-            ) from clone_err
+            ) from e
 
-    # 入队执行；若入队失败则标记任务为 failed 避免僵尸数据
-    try:
-        if not safe_enqueue_generation_task(
-            task,
-            generation_task_repository,
-            user_id=user_id,
-            log_prefix="[预览生成]",
-            log_task_status=True,
-        ):
-            logger.warning("[预览生成] 任务入队失败: task_id=%s", task.id)
-            _mark_task_failed(generation_task_repository, task, "任务入队失败")
-            raise HTTPException(status_code=500, detail="任务入队失败，请稍后重试")
-    except UserPendingLimitExceeded as e:
-        _mark_task_failed(generation_task_repository, task, "待处理任务超限")
-        raise HTTPException(
-            status_code=429,
-            detail=f"您的待处理任务过多（当前 {e.pending_count - 1}/{e.limit}），请等待后再提交",
-        ) from None
-    except GlobalQueueFull:
-        _mark_task_failed(generation_task_repository, task, "系统队列已满")
-        raise HTTPException(
-            status_code=503,
-            detail="系统繁忙，请稍后再试",
-        ) from None
+    # 关联变体 plan 并回写标题配置
+    for variant_index, task in enumerate(created_tasks):
+        if variant_plan_ids:
+            task.source_edit_plan_id = variant_plan_ids[variant_index]
+            generation_task_repository.update(task)
+        # 回写变体标题到 plan config（worker 渲染时从 plan 读取 title 配置）
+        if task.source_edit_plan_id and (task.title_config or {}).get("text", "").strip():
+            try:
+                from app.api.routes.generation_tasks import _writeback_edit_plan_config
 
-    return _to_preview_response(task)
+                _writeback_edit_plan_config(
+                    plan_id=task.source_edit_plan_id,
+                    task_id=task.id,
+                    title_config=task.title_config,
+                    db=db,
+                )
+            except Exception:
+                logger.warning(
+                    "[预览生成] 回写标题配置失败（不影响主流程）: task_id=%s",
+                    task.id,
+                    exc_info=True,
+                )
+
+    # ── 入队 ──
+    responses: list[PreviewGenerationTaskResponse] = []
+    for variant_index, task in enumerate(created_tasks):
+        try:
+            enqueued = safe_enqueue_generation_task(
+                task,
+                generation_task_repository,
+                user_id=user_id,
+                log_prefix=f"[预览生成][变体{variant_index + 1}]",
+                log_task_status=True,
+            )
+            if not enqueued:
+                logger.warning("[预览生成] 任务入队失败: task_id=%s", task.id)
+                _mark_task_failed(generation_task_repository, task, "任务入队失败")
+        except UserPendingLimitExceeded:
+            _mark_task_failed(generation_task_repository, task, "待处理任务超限")
+        except GlobalQueueFull:
+            _mark_task_failed(generation_task_repository, task, "系统队列已满")
+        except Exception:
+            logger.exception("[预览生成] 入队异常: task_id=%s", task.id)
+            _mark_task_failed(generation_task_repository, task, "任务入队异常")
+        # enqueue 会原地更新 task 状态/进度，直接用 task 构造响应
+        responses.append(_to_preview_response(task))
+
+    # 队列满/限流时若全部失败，返回明确错误码
+    if all(r.status == "failed" for r in responses):
+        first_err = next((r.error_message for r in responses if r.error_message), "")
+        if "待处理任务" in first_err:
+            raise HTTPException(status_code=429, detail=first_err or "待处理任务超限")
+        if "队列" in first_err:
+            raise HTTPException(status_code=503, detail=first_err or "系统繁忙，请稍后再试")
+
+    logger.info(
+        "[预览生成] 创建完成: %d 个变体任务, task_ids=%s",
+        len(responses),
+        [r.task_id for r in responses],
+    )
+    return BatchPreviewGenerationTaskResponse(items=responses, total=len(responses))
 
 
 @router.get("/preview/{task_id}", response_model=PreviewGenerationTaskResponse)

apps/api/app/api/routes/generation_tasks.py

@@ -47,6 +47,15 @@ logger = logging.getLogger(__name__)
 router = APIRouter()
 
 
+def _variant_value(values: list[str], index: int, fallback: str = "") -> str:
+    """从变体数组中取值：长度1=共用，长度>N=按索引，空数组=回退 fallback。"""
+    if not values:
+        return fallback
+    if len(values) == 1:
+        return values[0]
+    return values[index] if index < len(values) else fallback
+
+
 def _to_generation_task_response(task) -> GenerationTaskResponse:
     return GenerationTaskResponse(
         id=task.id,
@@ -472,12 +481,21 @@ def create_generation_task(
             if task_index > 0 and variant_plan_ids:
                 effective_plan_id = variant_plan_ids[task_index - 1]
 
+            # 变体级独立配置：titles[]/voice_library_ids[]/cover_urls[]
+            # 长度1=所有变体共用，长度=count=每个变体独立，空数组=回退单值字段
+            variant_title_text = _variant_value(request.titles, task_index, "")
+            variant_title_config = dict(request.title_config or {})
+            if variant_title_text.strip():
+                variant_title_config["text"] = variant_title_text.strip()
+            variant_voice_library_id = _variant_value(request.voice_library_ids, task_index, request.voice_library_id)
+            variant_cover_url = _variant_value(request.cover_urls, task_index, request.cover_url)
+
             task = use_case.execute(
                 CreateGenerationTaskCommand(
                     project_id=project_id,
                     asset_library_id=asset_library_id,
                     strategy_id=effective_strategy_id,
-                    voice_library_id=request.voice_library_id,
+                    voice_library_id=variant_voice_library_id,
                     template_id=request.template_id,
                     asset_ids=resolved_asset_ids,
                     title_ids=request.title_ids,
@@ -495,10 +513,12 @@ def create_generation_task(
                     source_task_id=request.source_task_id,
                     output_width=request.output_width,
                     output_height=request.output_height,
-                    cover_url=request.cover_url,
-                    title_config=request.title_config or {},
+                    cover_url=variant_cover_url,
+                    title_config=variant_title_config,
                 )
             )
+            # 变体序号写入 extra_meta（响应/排查时可辨识）
+            task.extra_meta["variant_index"] = task_index
             try:
                 # 兜底关联编辑计划：前端未传 source_edit_plan_id 时，
                 # 通过 template_id + user_id 在 DB 层直接查找最新的 plan。
@@ -533,13 +553,13 @@ def create_generation_task(
 
                 # 回写 plan.config：必须在 enqueue 之前执行，
                 # 确保 worker 读取 plan 时 config 中已包含 generation_task_id。
-                # 只在首个任务时回写一次，避免批量生成时循环覆盖。
+                # 批量场景下每个变体关联独立 plan，需各自回写自己的变体标题配置。
                 _effective_plan_id = task.source_edit_plan_id
-                if _effective_plan_id and len(created_tasks) == 0:
+                if _effective_plan_id:
                     _writeback_edit_plan_config(
                         plan_id=_effective_plan_id,
                         task_id=task.id,
-                        title_config=request.title_config,
+                        title_config=variant_title_config,
                         db=db,
                     )
 

apps/api/app/api/routes/videos.py

@@ -252,6 +252,10 @@ class RecomputeDedupRequest(BaseModel):
         None,
         description="指定视频 ID 列表。为空则对当前用户所有缺少查重数据的视频重新计算。",
     )
+    force: bool = Field(
+        False,
+        description="强制重算：即使视频已有查重数据也重新入队（#1702 查重算法升级后用于存量视频重算）。",
+    )
 
 
 class RecomputeDedupResponse(BaseModel):
@@ -291,15 +295,15 @@ def recompute_dedup(
     skipped = 0
 
     for video in target_videos:
-        # 已有完整查重数据的跳过
-        if video.duplicate_rate is not None and video.video_fingerprint:
+        # 已有完整查重数据的跳过（force=True 时强制重算，#1702 算法升级后存量视频需要重算指纹/分片）
+        if not request.force and video.duplicate_rate is not None and video.video_fingerprint:
             skipped += 1
             continue
 
         # 触发异步查重任务
         celery_app.send_task("worker.check_duplicate", args=[video.id])
         enqueued += 1
-        logger.info("Enqueued re-dedup for video %s (user=%s)", video.id, user_id)
+        logger.info("Enqueued re-dedup for video %s (user=%s, force=%s)", video.id, user_id, request.force)
 
     return RecomputeDedupResponse(
         enqueued=enqueued,

apps/api/app/schemas/generation_task.py

@@ -25,6 +25,12 @@ class CreateGenerationTaskRequest(BaseModel):
     asset_library_id: str = ""
     strategy_id: str = ""
     voice_library_id: str = ""
+    # ── 多变体独立配音（批量生成）──
+    # 长度 1 = 所有变体共用；长度 = count = 每个变体独立配音；空数组 = 回退 voice_library_id
+    voice_library_ids: list[str] = Field(
+        default_factory=list,
+        description="各变体独立配音素材库ID数组：长度1=共用，长度=count=独立。为空时回退 voice_library_id",
+    )
     created_by_user_id: str = ""
     # ── 模板模式新增字段 ──
     template_id: str = ""
@@ -75,6 +81,27 @@ class CreateGenerationTaskRequest(BaseModel):
     output_width: int = Field(default=1280, description="输出视频宽度")
     output_height: int = Field(default=720, description="输出视频高度")
     cover_url: str = Field(default="", description="封面图片 URL")
+    # ── 多变体独立封面（批量生成）──
+    # 长度 1 = 所有变体共用；长度 = count = 每个变体独立封面；空数组 = 回退 cover_url
+    cover_urls: list[str] = Field(
+        default_factory=list,
+        description="各变体独立封面URL数组：长度1=共用，长度=count=独立。为空时回退 cover_url",
+    )
+    # ── 多变体独立标题文字（批量生成）──
+    # 长度 1 = 所有变体共用；长度 = count = 每个变体独立标题文字；空数组 = 使用 title_config.text
+    titles: list[str] = Field(
+        default_factory=list,
+        description="各变体独立标题文字数组：长度1=共用，长度=count=独立。为空时使用 title_config.text",
+    )
+
+    @model_validator(mode="after")
+    def _check_variant_arrays(self) -> "CreateGenerationTaskRequest":
+        """变体数组字段长度校验：空数组（回退单值）、长度 1（共用）、或长度 = count（独立）。"""
+        for name in ("voice_library_ids", "cover_urls", "titles"):
+            arr = getattr(self, name)
+            if arr and len(arr) != 1 and len(arr) != self.count:
+                raise ValueError(f"{name} 长度必须为 1（共用）或 {self.count}（与 count 一致），当前为 {len(arr)}")
+        return self
 
     @model_validator(mode="after")
     def _check_at_least_one_mode(self) -> "CreateGenerationTaskRequest":
@@ -185,8 +212,33 @@ class CreatePreviewGenerationTaskRequest(BaseModel):
     )
     title_config: dict = Field(
         default_factory=dict,
-        description="标题配置（可选），渲染时烧录到预览视频中。支持字段: text/font/font_size/font_color/position/bold/stroke/shadow",
+        description="标题配置（可选），渲染时烧录到预览视频中。支持字段: text/font/font_size/font_color/position/bold/stroke/shadow。N个变体时样式全局共用",
     )
+    # ── 多变体独立配置（preview_count > 1）──
+    # 长度 1 = 所有变体共用；长度 = preview_count = 每个变体独立；空数组 = 回退单值字段
+    titles: list[str] = Field(
+        default_factory=list,
+        description="各变体独立标题文字数组：长度1=共用，长度=preview_count=独立。为空时使用 title_config.text",
+    )
+    voice_library_ids: list[str] = Field(
+        default_factory=list,
+        description="各变体独立配音素材库ID数组：长度1=共用，长度=preview_count=独立。为空时回退 voice_library_id",
+    )
+    cover_urls: list[str] = Field(
+        default_factory=list,
+        description="各变体独立封面URL数组：长度1=共用，长度=preview_count=独立（预览阶段通常为空）",
+    )
+
+    @model_validator(mode="after")
+    def _check_variant_arrays(self) -> "CreatePreviewGenerationTaskRequest":
+        """变体数组字段长度校验：空数组（回退单值）、长度 1（共用）、或长度 = preview_count（独立）。"""
+        for name in ("titles", "voice_library_ids", "cover_urls"):
+            arr = getattr(self, name)
+            if arr and len(arr) != 1 and len(arr) != self.preview_count:
+                raise ValueError(
+                    f"{name} 长度必须为 1（共用）或 {self.preview_count}（与 preview_count 一致），当前为 {len(arr)}"
+                )
+        return self
 
     @model_validator(mode="after")
     def _check_template_id(self) -> "CreatePreviewGenerationTaskRequest":
@@ -202,7 +254,7 @@ class CreatePreviewGenerationTaskRequest(BaseModel):
 
 
 class PreviewGenerationTaskResponse(BaseModel):
-    """预览生成任务响应。
+    """单个预览变体任务响应。
 
     包含任务状态、进度、分辨率、生成结果 URL 等关键字段。
     """
@@ -211,6 +263,7 @@ class PreviewGenerationTaskResponse(BaseModel):
     status: str
     progress: float
     is_preview: bool = True
+    variant_index: int = 0
     resolution: str = ""
     video_url: str = ""
     duration: float = 0.0
@@ -219,7 +272,21 @@ class PreviewGenerationTaskResponse(BaseModel):
     transition_count: int = 0
     material_usage: dict = Field(default_factory=dict)
     error_message: str = ""
+    title_text: str = ""
+    voice_library_id: str = ""
     created_at: datetime | None = None
     started_at: datetime | None = None
     finished_at: datetime | None = None
     generate_duration: float = 0.0
+
+
+class BatchPreviewGenerationTaskResponse(BaseModel):
+    """批量预览任务响应：preview_count=N 时返回 N 个独立变体任务。
+
+    - items: 变体任务数组，按 variant_index 顺序排列，每个含独立 task_id/状态/预览视频URL
+    - total: 变体总数（= preview_count）
+    - 前端按 items[i].task_id 分别轮询 GET /preview/{task_id} 获取进度与结果
+    """
+
+    items: list[PreviewGenerationTaskResponse]
+    total: int

apps/worker/video_processing/dedup.py

@@ -31,25 +31,60 @@ 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  # 长视频每段最少帧数
 
-# ── 滑动窗口匹配常量 ────────────────────────────────────────────
-SEGMENT_MATCH_THRESHOLD = 8  # 帧匹配汉明距离阈值
-MIN_CONSECUTIVE_MATCHES = 5  # 最少连续匹配帧数
+# ── 滑动窗口匹配常量（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  # 至少 70% 帧匹配
+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（离散余弦变换）。
 
@@ -101,11 +136,17 @@ def hamming_distance(hash1: str, hash2: str) -> int:
 
 
 def compute_color_histogram(image: np.ndarray, bins: int = 32) -> list[float]:
-    """Compute color histogram for an image."""
+    """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).flatten()
+        h = cv2.normalize(h, h, norm_type=cv2.NORM_L1).flatten()
         hist.extend(h)
     return hist
 
@@ -210,6 +251,30 @@ def detect_keyframe_timestamps(
     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)]
+
+
 # ── 数据类 ──────────────────────────────────────────────────────
 
 
@@ -297,23 +362,32 @@ def find_duplicate_segments(
     target_chunks: list,
     *,
     match_threshold: int = SEGMENT_MATCH_THRESHOLD,
-    min_consecutive: int = MIN_CONSECUTIVE_MATCHES,
+    min_consecutive: Optional[int] = None,
     max_gap: int = MAX_GAP,
+    neighbor_window: int = NEIGHBOR_WINDOW,
 ) -> list[DuplicateSegment]:
-    """滑动窗口时序匹配：找出两组分片之间的重复片段。
+    """滑动窗口时序匹配：找出两组分片之间的重复片段（Issue #1702 重构）。
 
     算法：
-    1. 对每个 query chunk，找到 target 中汉明距离最小的 chunk
-    2. 距离 <= match_threshold 视为匹配
-    3. 找连续匹配的 run（允许 max_gap 帧间隙）
-    4. 连续匹配数 >= min_consecutive 的 run 报告为重复片段
+    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: 汉明距离匹配阈值
-        min_consecutive: 最少连续匹配帧数
+        match_threshold: 汉明距离匹配阈值（统一常量 PHASH_THRESHOLD）
+        min_consecutive: 最少连续匹配帧数；None 时按短视频自适应
         max_gap: 允许的最大间隙帧数
+        neighbor_window: 时序对齐允许的目标分片序号邻接窗口
 
     Returns:
         DuplicateSegment 列表
@@ -321,95 +395,93 @@ def find_duplicate_segments(
     if not query_chunks or not target_chunks:
         return []
 
-    def _get_phash(chunk) -> str:
+    def _get(chunk, key):
         if isinstance(chunk, dict):
-            return chunk["phash_binary"]
-        return chunk.phash_binary
+            return chunk[key]
+        return getattr(chunk, key)
 
-    def _get_start(chunk) -> int:
-        if isinstance(chunk, dict):
-            return chunk["start_time_ms"]
-        return chunk.start_time_ms
+    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]
 
-    def _get_end(chunk) -> int:
-        if isinstance(chunk, dict):
-            return chunk["end_time_ms"]
-        return chunk.end_time_ms
+    # 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)
 
-    # 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))
+    # 短视频自适应连续匹配门槛（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: 找连续匹配的 runs
-    runs: list[tuple[int, int]] = []  # list of (start_idx, end_idx)
-    run_start = None
+    # 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
 
-    for i, (is_match, _dist, _idx) in enumerate(frame_matches):
-        if is_match:
+    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  # 重置间隙
+            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:
-                    # 中断当前 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
+                    # 非匹配帧从 i-gap_count+1 开始，run 结束于其前一帧
+                    _close_run(run_start, i - gap_count)
+                    run_start, run_last_t, gap_count = None, None, 0
 
-    # 处理末尾 run
     if run_start is not None:
-        last_idx = len(frame_matches) - 1
-        # 回退找到最后一个匹配帧的位置（跳过尾部非匹配帧）
+        last_idx = n - 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))
+        _close_run(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)
+        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=query_start,
-                query_end_ms=query_end,
-                target_start_ms=target_start,
-                target_end_ms=target_end,
+                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,
             )
         )
@@ -423,7 +495,9 @@ def find_duplicate_segments(
 class VideoDeduplicator:
     """Video deduplication using multiple fingerprint methods."""
 
-    PHASH_THRESHOLD = 8  # Issue #1658: pHash 汉明距离阈值由 10 收紧到 8，降低不同视频误判率
+    # Issue #1702: 阈值统一来源为模块常量 PHASH_THRESHOLD（#1658 曾收紧到 8，
+    # 后经 staging 真实同源/异源指纹分布重新校准，见 test_phash_threshold_calibration_1702）。
+    PHASH_THRESHOLD = PHASH_THRESHOLD
     HISTOGRAM_THRESHOLD = 0.85
 
     @staticmethod
@@ -447,27 +521,36 @@ class VideoDeduplicator:
         # 单帧不视为坏指纹（短视频或抽帧不足）
         if len(phashes) == 1:
             return False
-        # 多帧但所有 phash 完全相同 → 黑屏/纯色视频
+        # Issue #1702: 旧逻辑"所有 phash 完全相同即判黑屏"会误杀短视频——
+        # 11s 视频只有几个不同镜头时，相邻 1s 采样帧可能 phash 完全一致（内容
+        # 连续但非黑屏）。黑屏的特征是「大量帧全部无内容」，要求至少 8 帧
+        # 且相同帧占比 >=80% 才判坏；短视频（<8 帧）只有真正单值时交给
+        # _bhattacharyya/融合分兜底，不因"帧都一样"直接跳过。
+        if len(phashes) < 8:
+            return False
         unique = set(phashes)
-        if len(unique) == 1:
+        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）→ 近似黑屏
+        # 多帧但所有唯一 phash 之间的汉明距离都极小（<3）且占比 >=80% → 近似黑屏
         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 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 dynamic keyframe detection.
+        """Compute video fingerprint using dense uniform sampling.
 
-        使用 detect_keyframe_timestamps() 检测内容感知关键帧，
-        在每个关键帧处取帧计算 pHash + color_histogram。
-        同时保留 MD5 计算和分片数据结构。
+        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():
@@ -481,8 +564,8 @@ class VideoDeduplicator:
 
         cap.release()
 
-        # 1. 检测关键帧时间戳
-        keyframe_times = detect_keyframe_timestamps(video_path)
+        # 1. 固定间隔密集采样（Issue #1702：替代动态场景检测，保证跨视频时序对齐）
+        keyframe_times = sample_fingerprint_timestamps(duration)
 
         if not keyframe_times:
             return VideoFingerprint(
@@ -506,12 +589,15 @@ class VideoDeduplicator:
             if not ret:
                 continue
 
-            # MD5 计算
+            # MD5 计算（基于原始帧，指纹文件级去重不受裁剪影响）
             _, buffer = cv2.imencode(".jpg", frame)
             md5_hash.update(buffer)
 
-            phash = compute_phash(frame)
-            hist = compute_color_histogram(frame)
+            # 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
@@ -564,12 +650,22 @@ class VideoDeduplicator:
 
     @staticmethod
     def _bhattacharyya_coefficient(hist_a: list[float], hist_b: list[float]) -> float:
-        """Bhattacharyya 系数：Σ √(a[i] * b[i])，范围 [0, 1]，1=完全相同。"""
+        """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 = 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)))
+        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(
@@ -611,6 +707,72 @@ class VideoDeduplicator:
         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,
@@ -649,6 +811,9 @@ class VideoDeduplicator:
         else:
             existing_videos = video_repo.list_by_project(project_id)
 
+        best_score = 0.0
+        best_result: Optional[dict] = None
+
         for existing in existing_videos:
             if not existing.video_fingerprint:
                 continue
@@ -677,61 +842,70 @@ class VideoDeduplicator:
             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 显式回退空列表）
+            # 直方图 / 分片对象（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
+                ]
 
-            combined_score = self._compute_fusion_score(
-                median_distance, fingerprint.color_histograms, existing_histograms
+            # 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"]),
             )
 
-            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]
+            # 全片重复判定：融合分过阈 且（帧匹配比例 >=70% 或 局部覆盖 >=50%）
+            is_full_duplicate = ev["fusion"] >= DUPLICATE_THRESHOLD and (
+                ev["frame_match_rate"] >= MATCH_RATIO_THRESHOLD or ev["temporal_coverage"] >= PARTIAL_COVERAGE_THRESHOLD
             )
-            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
-                ],
-            }
+            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(
@@ -763,6 +937,9 @@ class VideoDeduplicator:
         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
@@ -796,59 +973,59 @@ class VideoDeduplicator:
             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")]
+                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
+                ]
 
-            combined_score = self._compute_fusion_score(
-                median_distance, fingerprint.color_histograms, existing_histograms
+            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"]),
             )
 
-            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]
+            is_full_duplicate = ev["fusion"] >= DUPLICATE_THRESHOLD and (
+                ev["frame_match_rate"] >= MATCH_RATIO_THRESHOLD or ev["temporal_coverage"] >= PARTIAL_COVERAGE_THRESHOLD
             )
-            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
-                ],
-            }
+            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(
@@ -897,8 +1074,7 @@ class VideoDeduplicator:
         max_duplicate_rate = 0.0
         max_visual_similarity = 0.0
         match_count = 0
-
-        total_duration_ms = fingerprint.duration if fingerprint.duration else 0
+        evaluated = 0
 
         for existing in existing_videos:
             if current_video_id and existing.id == current_video_id:
@@ -933,57 +1109,63 @@ class VideoDeduplicator:
             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
+            # 直方图 / 分片对象（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:
-                # JSON NULL 显式回退空列表
                 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
+                ]
 
-            visual_sim = self._compute_fusion_score(median_distance, fingerprint.color_histograms, existing_histograms)
+            # 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"]),
+            )
 
-            # 判定是否为重复（融合分数超过阈值）
-            if visual_sim >= DUPLICATE_THRESHOLD:
+            # 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 = visual_sim
+                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),
@@ -999,18 +1181,20 @@ def _save_fingerprint_chunks(
     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
 
+    # 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)
@@ -1032,9 +1216,17 @@ def check_duplicate_task(self: Task, generated_video_id: str) -> dict:
             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
-        )
+        # 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)
 
@@ -1045,7 +1237,9 @@ def check_duplicate_task(self: Task, generated_video_id: str) -> dict:
             session,
             scope="user",
             user_id=video.user_id,
-            duration_sec=fingerprint.duration / 1000 if fingerprint.duration else 0,
+            # Issue #1702: fingerprint.duration 单位已经是秒，旧代码 /1000 导致
+            # ±15% 时长预过滤窗口缩到 ~0.013s，scope=user 的跨项目查重永远返回 None。
+            duration_sec=fingerprint.duration if fingerprint.duration else 0,
         )
 
         video.video_fingerprint = fingerprint.to_dict()

apps/worker/video_processing/dedup_helpers.py

@@ -92,7 +92,8 @@ def create_video_record_and_dedup(
                 logger.warning("Failed to save fingerprint chunks for %s: %s", video_id, chunk_err)
 
             # (a) 历史成片查重（跨项目全局 + 时长预过滤）
-            duration_sec = fingerprint.duration / 1000 if fingerprint.duration else 0
+            # Issue #1702: fingerprint.duration 单位是秒，旧代码 /1000 让时长预过滤失效
+            duration_sec = fingerprint.duration if fingerprint.duration else 0
             duplicate_result = deduplicator.check_duplicate(
                 fingerprint,
                 project_id,

tests/unit/test_1677_batch_variants.py

@@ -0,0 +1,507 @@
+"""Issue #1677 多视频批量生成 — 变体独立配置与批量预览/批量生成测试。
+
+覆盖：
+- 批量预览：preview_count=N 一次创建 N 个独立任务，返回变体数组
+- 变体克隆链路：N 个预览/正式任务各自关联独立克隆 plan
+- 变体独立配置：titles[]/voice_library_ids[]/cover_urls[] 按变体注入
+- 长度校验：数组长度必须为 1 或 N（共用或独立），非法长度报错
+- N=1 向后兼容：旧字段单值行为不变
+"""
+
+from datetime import datetime, timezone
+from unittest.mock import MagicMock, patch
+
+import pytest
+from app.core.task_enqueue import GlobalQueueFull, UserPendingLimitExceeded
+from app.schemas.generation_task import (
+    BatchPreviewGenerationTaskResponse,
+    CreateGenerationTaskRequest,
+    CreatePreviewGenerationTaskRequest,
+)
+
+from packages.domain import GenerationTask
+from packages.domain.generation_task import GenerationTaskStatus
+
+# ════════════════════════════════════════════════════════════════════════════
+# 辅助构造
+# ════════════════════════════════════════════════════════════════════════════
+
+
+def _make_user(user_id="test_user_001"):
+    mock_user = MagicMock()
+    mock_user.id = user_id
+    auth = MagicMock()
+    auth.user = mock_user
+    return auth
+
+
+def _make_task(task_id=None, status=GenerationTaskStatus.PENDING, source_plan_id=None):
+    task = GenerationTask.create(
+        project_id="",
+        asset_library_id="",
+        template_id="tpl_001",
+        asset_ids=["asset_1"],
+    )
+    if task_id:
+        task.id = task_id
+    task.status = status
+    task.is_preview = True
+    task.source_edit_plan_id = source_plan_id or ""
+    task.voice_library_id = ""
+    task.title_config = {}
+    task.cover_url = ""
+    return task
+
+
+def _make_preview_request(**kwargs):
+    defaults = {
+        "template_id": "tpl_001",
+        "asset_ids": ["asset_1", "asset_2"],
+    }
+    defaults.update(kwargs)
+    return CreatePreviewGenerationTaskRequest(**defaults)
+
+
+def _repo_mock():
+    repo = MagicMock()
+    repo.count_pending_by_user.return_value = 0
+    repo.count_pending_total.return_value = 0
+    repo.get.side_effect = lambda tid: None
+    return repo
+
+
+# ════════════════════════════════════════════════════════════════════════════
+# Schema 校验：变体数组长度
+# ════════════════════════════════════════════════════════════════════════════
+
+
+class TestVariantArrayValidation:
+    """变体数组字段长度校验。"""
+
+    def test_preview_titles_length_matches_count(self):
+        """titles 长度 = preview_count 合法"""
+        req = _make_preview_request(preview_count=3, titles=["标题A", "标题B", "标题C"])
+        assert len(req.titles) == 3
+
+    def test_preview_titles_single_shared(self):
+        """titles 长度 1 = 所有变体共用，合法"""
+        req = _make_preview_request(preview_count=3, titles=["共用标题"])
+        assert req.titles == ["共用标题"]
+
+    def test_preview_titles_wrong_length_raises(self):
+        """titles 长度 2 与 preview_count=3 不匹配 → 报错"""
+        with pytest.raises(ValueError, match="titles"):
+            _make_preview_request(preview_count=3, titles=["A", "B"])
+
+    def test_preview_voice_ids_wrong_length_raises(self):
+        """voice_library_ids 长度非法 → 报错"""
+        from pydantic import ValidationError
+
+        with pytest.raises(ValidationError, match="voice_library_ids"):
+            _make_preview_request(preview_count=4, voice_library_ids=["v1", "v2"])
+
+    def test_preview_empty_arrays_ok(self):
+        """空数组（回退单值字段）合法"""
+        req = _make_preview_request(preview_count=3)
+        assert req.titles == []
+        assert req.voice_library_ids == []
+        assert req.cover_urls == []
+
+    def test_generation_titles_length_matches_count(self):
+        """正式生成 titles 长度 = count 合法"""
+        req = CreateGenerationTaskRequest(
+            template_id="tpl_1",
+            asset_ids=["a1"],
+            count=3,
+            titles=["A", "B", "C"],
+        )
+        assert len(req.titles) == 3
+
+    def test_generation_arrays_wrong_length_raises(self):
+        """正式生成 cover_urls 长度与 count 不匹配 → 报错"""
+        from pydantic import ValidationError
+
+        with pytest.raises(ValidationError, match="cover_urls"):
+            CreateGenerationTaskRequest(
+                template_id="tpl_1",
+                asset_ids=["a1"],
+                count=3,
+                cover_urls=["c1", "c2"],
+            )
+
+    def test_generation_single_count_no_arrays(self):
+        """N=1 且不传数组：完全旧行为"""
+        req = CreateGenerationTaskRequest(template_id="tpl_1", asset_ids=["a1"])
+        assert req.count == 1
+        assert req.titles == []
+        assert req.voice_library_ids == []
+        assert req.cover_urls == []
+
+
+# ════════════════════════════════════════════════════════════════════════════
+# 批量预览路由
+# ════════════════════════════════════════════════════════════════════════════
+
+
+class TestBatchPreviewRoute:
+    """POST /preview 批量变体。"""
+
+    def test_preview_count_1_returns_single_item_array(self):
+        """N=1 返回 items 长度 1 的批量响应（结构统一）"""
+        from app.api.routes.generation_preview import create_preview_generation_task
+
+        task = _make_task(task_id="task_1")
+        repo = _repo_mock()
+        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
+            MockUC.return_value.execute.return_value = task
+            with patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True):
+                resp = create_preview_generation_task(
+                    _make_preview_request(preview_count=1),
+                    authenticated_user=_make_user(),
+                    generation_task_repository=repo,
+                    db=MagicMock(),
+                )
+        assert isinstance(resp, BatchPreviewGenerationTaskResponse)
+        assert resp.total == 1
+        assert len(resp.items) == 1
+        assert resp.items[0].task_id == "task_1"
+        assert resp.items[0].variant_index == 0
+
+    def test_preview_count_3_creates_three_independent_tasks(self):
+        """N=3 创建 3 个独立任务，返回 3 个变体，task_id 各不相同"""
+        from app.api.routes.generation_preview import create_preview_generation_task
+
+        tasks = [_make_task(task_id=f"task_{i}") for i in range(3)]
+        repo = _repo_mock()
+        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
+            MockUC.return_value.execute.side_effect = tasks
+            with patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True):
+                resp = create_preview_generation_task(
+                    _make_preview_request(preview_count=3),
+                    authenticated_user=_make_user(),
+                    generation_task_repository=repo,
+                    db=MagicMock(),
+                )
+        assert resp.total == 3
+        task_ids = [item.task_id for item in resp.items]
+        assert task_ids == ["task_0", "task_1", "task_2"]
+        assert len(set(task_ids)) == 3
+        for i, item in enumerate(resp.items):
+            assert item.variant_index == i
+
+    def test_preview_count_3_clones_three_variant_plans(self):
+        """有源 plan 时，N=3 克隆 3 个独立变体 plan（预览全部克隆，不用源 plan）"""
+        from app.api.routes.generation_preview import create_preview_generation_task
+
+        tasks = [_make_task(task_id=f"task_{i}", source_plan_id="source_plan") for i in range(3)]
+        repo = _repo_mock()
+        cloned_plan_ids = ["clone_1", "clone_2", "clone_3"]
+        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
+            MockUC.return_value.execute.side_effect = tasks
+            with patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True):
+                with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc:
+                    clone_results = [MagicMock(id=pid) for pid in cloned_plan_ids]
+                    MockPlanSvc.return_value.clone_plan_for_variant.side_effect = clone_results
+                    create_preview_generation_task(
+                        _make_preview_request(preview_count=3),
+                        authenticated_user=_make_user(),
+                        generation_task_repository=repo,
+                        db=MagicMock(),
+                    )
+        # 克隆被调用 3 次
+        assert MockPlanSvc.return_value.clone_plan_for_variant.call_count == 3
+        # 每个任务关联到不同的克隆 plan
+        for i, task in enumerate(tasks):
+            assert task.source_edit_plan_id == cloned_plan_ids[i]
+
+    def test_preview_variant_titles_injected_per_variant(self):
+        """titles[] 按变体注入 title_config.text"""
+        from app.api.routes.generation_preview import create_preview_generation_task
+
+        tasks = [_make_task(task_id=f"task_{i}") for i in range(3)]
+        repo = _repo_mock()
+        captured_commands = []
+        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
+
+            def _execute(cmd):
+                captured_commands.append(cmd)
+                return tasks[len(captured_commands) - 1]
+
+            MockUC.return_value.execute.side_effect = _execute
+            with patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True):
+                create_preview_generation_task(
+                    _make_preview_request(
+                        preview_count=3,
+                        title_config={"font": "黑体", "position": "bottom"},
+                        titles=["标题A", "标题B", "标题C"],
+                    ),
+                    authenticated_user=_make_user(),
+                    generation_task_repository=repo,
+                    db=MagicMock(),
+                )
+        assert len(captured_commands) == 3
+        assert captured_commands[0].title_config["text"] == "标题A"
+        assert captured_commands[1].title_config["text"] == "标题B"
+        assert captured_commands[2].title_config["text"] == "标题C"
+        # 样式全局共用
+        assert all(c.title_config["font"] == "黑体" for c in captured_commands)
+
+    def test_preview_shared_title_when_single_length(self):
+        """titles 长度 1 = 所有变体共用同一标题"""
+        from app.api.routes.generation_preview import create_preview_generation_task
+
+        tasks = [_make_task(task_id=f"task_{i}") for i in range(3)]
+        repo = _repo_mock()
+        captured = []
+        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
+
+            def _execute(cmd):
+                captured.append(cmd)
+                return tasks[len(captured) - 1]
+
+            MockUC.return_value.execute.side_effect = _execute
+            with patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True):
+                create_preview_generation_task(
+                    _make_preview_request(preview_count=3, titles=["共用标题"]),
+                    authenticated_user=_make_user(),
+                    generation_task_repository=repo,
+                    db=MagicMock(),
+                )
+        assert all(c.title_config["text"] == "共用标题" for c in captured)
+
+    def test_preview_independent_voice_per_variant(self):
+        """voice_library_ids[] 按变体注入独立配音"""
+        from app.api.routes.generation_preview import create_preview_generation_task
+
+        tasks = [_make_task(task_id=f"task_{i}") for i in range(3)]
+        repo = _repo_mock()
+        captured = []
+        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
+
+            def _execute(cmd):
+                captured.append(cmd)
+                return tasks[len(captured) - 1]
+
+            MockUC.return_value.execute.side_effect = _execute
+            with patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True):
+                create_preview_generation_task(
+                    _make_preview_request(
+                        preview_count=3,
+                        voice_library_ids=["voice_a", "voice_b", "voice_c"],
+                    ),
+                    authenticated_user=_make_user(),
+                    generation_task_repository=repo,
+                    db=MagicMock(),
+                )
+        assert [c.voice_library_id for c in captured] == ["voice_a", "voice_b", "voice_c"]
+
+    def test_preview_voice_fallback_to_single_field(self):
+        """voice_library_ids 为空时回退 voice_library_id 单值字段（向后兼容）"""
+        from app.api.routes.generation_preview import create_preview_generation_task
+
+        task = _make_task(task_id="task_1")
+        repo = _repo_mock()
+        captured = []
+        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
+
+            def _execute(cmd):
+                captured.append(cmd)
+                return task
+
+            MockUC.return_value.execute.side_effect = _execute
+            with patch("app.api.routes.generation_preview.safe_enqueue_generation_task", return_value=True):
+                create_preview_generation_task(
+                    _make_preview_request(voice_library_id="legacy_voice"),
+                    authenticated_user=_make_user(),
+                    generation_task_repository=repo,
+                    db=MagicMock(),
+                )
+        assert captured[0].voice_library_id == "legacy_voice"
+
+    def test_preview_queue_limit_checks_total_count(self):
+        """限流预检查按变体总数计：用户 pending + N 超限 → 429"""
+        from app.api.routes.generation_preview import create_preview_generation_task
+        from fastapi import HTTPException
+
+        repo = MagicMock()
+        repo.count_pending_by_user.return_value = 3
+        repo.count_pending_total.return_value = 0
+        with pytest.raises(HTTPException) as exc:
+            create_preview_generation_task(
+                _make_preview_request(preview_count=5),
+                authenticated_user=_make_user(),
+                generation_task_repository=repo,
+                db=MagicMock(),
+            )
+        assert exc.value.status_code == 429
+
+    def test_preview_clone_failure_marks_all_failed(self):
+        """克隆变体 plan 失败 → 已创建任务全部标记 failed 并 500"""
+        from app.api.routes.generation_preview import create_preview_generation_task
+        from fastapi import HTTPException
+
+        tasks = [_make_task(task_id=f"task_{i}", source_plan_id="source_plan") for i in range(3)]
+        repo = _repo_mock()
+        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
+            MockUC.return_value.execute.side_effect = tasks
+            with patch("app.services.edit_plan_service.EditPlanService") as MockPlanSvc:
+                MockPlanSvc.return_value.clone_plan_for_variant.side_effect = RuntimeError("db down")
+                with pytest.raises(HTTPException) as exc:
+                    create_preview_generation_task(
+                        _make_preview_request(preview_count=3),
+                        authenticated_user=_make_user(),
+                        generation_task_repository=repo,
+                        db=MagicMock(),
+                    )
+        assert exc.value.status_code == 500
+        # 所有已创建任务都被标记 failed
+        assert all(t.status == GenerationTaskStatus.FAILED for t in tasks)
+
+
+# ════════════════════════════════════════════════════════════════════════════
+# 批量正式生成：变体配置注入
+# ════════════════════════════════════════════════════════════════════════════
+
+
+class TestBatchGenerationVariantConfig:
+    """POST /tasks count=N 时变体独立配置。"""
+
+    def _call_create_tasks(self, request, repo=None):
+        from app.api.routes.generation_tasks import create_generation_task
+
+        repo = repo or MagicMock()
+        repo.count_pending_by_user.return_value = 0
+        repo.count_pending_total.return_value = 0
+        repo.update.return_value = None
+
+        # 模板模式：asset_repository.find_by_id 返回 None（无 project 关联，
+        # 纯模板模式 project_id/library_id 都为空），避免 MagicMock 属性污染
+        asset_repo = MagicMock()
+        asset_repo.find_by_id.return_value = None
+
+        # db.query().filter()...first() 返回 None：不走兜底关联编辑计划
+        db = MagicMock()
+        db.query.return_value.filter.return_value.order_by.return_value.first.return_value = None
+
+        return create_generation_task(
+            request,
+            authenticated_user=_make_user(),
+            generation_task_repository=repo,
+            project_repository=MagicMock(),
+            asset_library_repository=MagicMock(),
+            asset_repository=asset_repo,
+            db=db,
+        )
+
+    def test_count_3_variant_titles_voices_covers_injected(self):
+        """count=3：titles/voice_library_ids/cover_urls 按变体注入"""
+        from app.api.routes import generation_tasks as routes
+
+        tasks = [_make_task(task_id=f"gen_{i}") for i in range(3)]
+        captured = []
+        with patch.object(routes, "CreateGenerationTaskUseCase") as MockUC:
+
+            def _execute(cmd):
+                captured.append(cmd)
+                t = tasks[len(captured) - 1]
+                t.title_config = cmd.title_config
+                t.voice_library_id = cmd.voice_library_id
+                t.cover_url = cmd.cover_url
+                return t
+
+            MockUC.return_value.execute.side_effect = _execute
+            with patch.object(routes, "safe_enqueue_generation_task", return_value=True):
+                req = CreateGenerationTaskRequest(
+                    template_id="tpl_1",
+                    asset_ids=["a1"],
+                    count=3,
+                    title_config={"font": "宋体"},
+                    titles=["成片标题1", "成片标题2", "成片标题3"],
+                    voice_library_ids=["v1", "v2", "v3"],
+                    cover_urls=["http://c1", "http://c2", "http://c3"],
+                )
+                resp = self._call_create_tasks(req)
+        assert resp.total == 3
+        assert [c.title_config["text"] for c in captured] == ["成片标题1", "成片标题2", "成片标题3"]
+        assert [c.voice_library_id for c in captured] == ["v1", "v2", "v3"]
+        assert [c.cover_url for c in captured] == ["http://c1", "http://c2", "http://c3"]
+        # 样式共用
+        assert all(c.title_config["font"] == "宋体" for c in captured)
+
+    def test_count_1_legacy_fields_unchanged(self):
+        """N=1 不传数组：旧字段 voice_library_id/cover_url/title_config 行为不变"""
+        from app.api.routes import generation_tasks as routes
+
+        task = _make_task(task_id="gen_1")
+        task.is_preview = False
+        captured = []
+        with patch.object(routes, "CreateGenerationTaskUseCase") as MockUC:
+
+            def _execute(cmd):
+                captured.append(cmd)
+                return task
+
+            MockUC.return_value.execute.side_effect = _execute
+            with patch.object(routes, "safe_enqueue_generation_task", return_value=True):
+                req = CreateGenerationTaskRequest(
+                    template_id="tpl_1",
+                    asset_ids=["a1"],
+                    count=1,
+                    voice_library_id="legacy_voice",
+                    cover_url="http://legacy-cover",
+                    title_config={"text": "旧标题", "font": "黑体"},
+                )
+                resp = self._call_create_tasks(req)
+        assert resp.total == 1
+        assert captured[0].voice_library_id == "legacy_voice"
+        assert captured[0].cover_url == "http://legacy-cover"
+        assert captured[0].title_config["text"] == "旧标题"
+
+    def test_count_3_shared_single_value_arrays(self):
+        """数组长度 1：3 个变体共用同一配音/封面"""
+        from app.api.routes import generation_tasks as routes
+
+        tasks = [_make_task(task_id=f"gen_{i}") for i in range(3)]
+        captured = []
+        with patch.object(routes, "CreateGenerationTaskUseCase") as MockUC:
+
+            def _execute(cmd):
+                captured.append(cmd)
+                return tasks[len(captured) - 1]
+
+            MockUC.return_value.execute.side_effect = _execute
+            with patch.object(routes, "safe_enqueue_generation_task", return_value=True):
+                req = CreateGenerationTaskRequest(
+                    template_id="tpl_1",
+                    asset_ids=["a1"],
+                    count=3,
+                    voice_library_ids=["shared_voice"],
+                    cover_urls=["http://shared"],
+                )
+                self._call_create_tasks(req)
+        assert all(c.voice_library_id == "shared_voice" for c in captured)
+        assert all(c.cover_url == "http://shared" for c in captured)
+
+
+class TestVariantValueHelper:
+    """_variant_value 取值逻辑。"""
+
+    def test_empty_returns_fallback(self):
+        from app.api.routes.generation_preview import _variant_value
+
+        assert _variant_value([], 0, fallback="fb") == "fb"
+
+    def test_single_length_shared(self):
+        from app.api.routes.generation_preview import _variant_value
+
+        assert _variant_value(["only"], 5) == "only"
+
+    def test_indexed_access(self):
+        from app.api.routes.generation_preview import _variant_value
+
+        assert _variant_value(["a", "b", "c"], 1) == "b"
+
+    def test_index_out_of_range_fallback(self):
+        from app.api.routes.generation_preview import _variant_value
+
+        assert _variant_value(["a", "b"], 9, fallback="x") == "x"

tests/unit/test_bad_fingerprint_filter.py

@@ -85,17 +85,18 @@ class TestIsBadFingerprint:
         assert VideoDeduplicator._is_bad_fingerprint(["abcdef0123456789"]) is False
 
     def test_all_identical_phashes_is_bad(self):
-        """多帧但所有 phash 完全相同 → 黑屏/纯色视频。"""
-        phashes = ["aaaaaaaaaaaaaaaa"] * 5
+        """>=8 帧且所有 phash 完全相同 → 黑屏/纯色视频（#1702：短帧不误杀）。"""
+        phashes = ["aaaaaaaaaaaaaaaa"] * 10
         assert VideoDeduplicator._is_bad_fingerprint(phashes) is True
 
-    def test_two_identical_phashes_is_bad(self):
-        """两帧完全相同也视为坏指纹。"""
-        assert VideoDeduplicator._is_bad_fingerprint(["bbbbbbbbbbbbbbbb", "bbbbbbbbbbbbbbbb"]) is True
+    def test_short_identical_phashes_not_bad(self):
+        """<8 帧完全相同不判坏——短视频内容连续时相邻采样帧 phash 天然相同（#1702）。"""
+        assert VideoDeduplicator._is_bad_fingerprint(["bbbbbbbbbbbbbbbb"] * 5) is False
+        assert VideoDeduplicator._is_bad_fingerprint(["bbbbbbbbbbbbbbbb", "bbbbbbbbbbbbbbbb"]) is False
 
     def test_all_very_similar_phashes_is_bad(self):
-        """多帧 phash 之间的汉明距离都 < 3 → 近似黑屏。"""
-        phashes = ["0000000000000000", "0000000000000001", "0000000000000002"]
+        """>=8 帧 phash 之间的汉明距离都 < 3 且高占比 → 近似黑屏。"""
+        phashes = ["0000000000000000"] * 8 + ["0000000000000001", "0000000000000002"]
         assert VideoDeduplicator._is_bad_fingerprint(phashes) is True
 
     def test_diverse_phashes_is_good(self):
@@ -122,7 +123,9 @@ class TestIsBadFingerprint:
         """已知黑屏视频的 phash 特征（全零或均匀分布）。"""
         assert VideoDeduplicator._is_bad_fingerprint(["0000000000000000"] * 10) is True
         assert VideoDeduplicator._is_bad_fingerprint(["ffffffffffffffff"] * 8) is True
-        assert VideoDeduplicator._is_bad_fingerprint(["9999999999999966"] * 6) is True
+        assert VideoDeduplicator._is_bad_fingerprint(["9999999999999966"] * 8) is True
+        # <8 帧不判坏（#1702 短视频保护）
+        assert VideoDeduplicator._is_bad_fingerprint(["9999999999999966"] * 5) is False
 
 
 # ── Helper ──────────────────────────────────────────────────────
@@ -151,13 +154,13 @@ class TestCheckDuplicateBadFingerprint:
         deduplicator = VideoDeduplicator()
         mock_session = MagicMock()
 
-        black_screen = _make_existing_video("vid-black", "md5_black", ["aaaaaaaaaaaaaaaa"] * 5)
+        black_screen = _make_existing_video("vid-black", "md5_black", ["aaaaaaaaaaaaaaaa"] * 10)
         mock_repo = MagicMock()
         mock_repo.list_by_user.return_value = [black_screen]
 
         fingerprint = VideoFingerprint(
             md5="md5_normal",
-            keyframe_phashes=["aaaaaaaaaaaaaaaa"] * 5,
+            keyframe_phashes=["aaaaaaaaaaaaaaaa"] * 10,
             color_histograms=[],
             duration=10.0,
             resolution=(1280, 720),
@@ -206,7 +209,7 @@ class TestCheckDuplicateBadFingerprint:
         deduplicator = VideoDeduplicator()
         mock_session = MagicMock()
 
-        black_screen = _make_existing_video("vid-black", "same_md5", ["aaaaaaaaaaaaaaaa"] * 5)
+        black_screen = _make_existing_video("vid-black", "same_md5", ["aaaaaaaaaaaaaaaa"] * 10)
         mock_repo = MagicMock()
         mock_repo.list_by_user.return_value = [black_screen]
 
@@ -283,7 +286,7 @@ class TestComputeDuplicateRateBadFingerprint:
         mock_session = MagicMock()
 
         videos = [
-            _make_existing_video("vid-b1", "md5_b1", ["aaaaaaaaaaaaaaaa"] * 5),
+            _make_existing_video("vid-b1", "md5_b1", ["aaaaaaaaaaaaaaaa"] * 10),
             _make_existing_video("vid-b2", "md5_b2", ["bbbbbbbbbbbbbbbb"] * 5),
         ]
         mock_repo = MagicMock()

tests/unit/test_dedup_1702_zero_rate_fix.py

@@ -0,0 +1,389 @@
+"""Issue #1702 — 查重率恒为 0% 修复：单测.
+
+覆盖验收要求：
+1. 同源不同裁剪的两个视频能检出非 0 相似度（指纹中心裁剪绕开降重 + 阈值校准）
+2. 局部片段复用（B 结尾 2s ≈ A 中间 2s）能检出
+3. 异源视频不误报（相似度接近 0）
+4. N=1 现有流程不回归
+5. P1 确定性 bug：时长预过滤单位 /1000、直方图归一化、temporal_coverage 量纲、阈值比较统一
+6. P0：±1 邻接对齐、短视频自适应连续门槛
+7. P2：0 匹配也要落日志
+"""
+
+from __future__ import annotations
+
+import logging
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+sys.modules.setdefault("cv2", MagicMock())
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT / "apps" / "worker"))
+sys.path.insert(0, str(ROOT / "packages"))
+
+
+from video_processing.dedup import (  # noqa: E402
+    PHASH_THRESHOLD,
+    SEGMENT_MATCH_THRESHOLD,
+    FingerprintChunk,
+    VideoDeduplicator,
+    VideoFingerprint,
+    find_duplicate_segments,
+)
+
+# ── helpers ────────────────────────────────────────────────────
+
+
+def _h(d: int) -> str:
+    """64-bit phash with exactly d bits set vs zero hash."""
+    bits = ["0"] * 64
+    for i in range(d):
+        bits[i] = "1"
+    return f"{int(''.join(bits), 2):016x}"
+
+
+def _chunk(phash: str, t0: float, t1: float):
+
+    return FingerprintChunk(
+        start_time_ms=int(t0 * 1000),
+        end_time_ms=int(t1 * 1000),
+        phash_binary=phash,
+        color_histogram=[],
+        frame_count=1,
+    )
+
+
+def _fingerprint(phashes, duration, chunks=None, md5="fp-md5-x"):
+
+    return VideoFingerprint(
+        md5=md5,
+        keyframe_phashes=list(phashes),
+        color_histograms=[],
+        duration=duration,
+        resolution=(1280, 720),
+        chunks=chunks or [],
+    )
+
+
+def _video(vid, phashes, duration=10.0, project_id="proj1"):
+    from packages.domain import GeneratedVideo
+
+    return GeneratedVideo(
+        id=vid,
+        project_id=project_id,
+        generation_task_id=f"task-{vid}",
+        name=f"video-{vid}.mp4",
+        file_url=f"https://example.com/{vid}.mp4",
+        file_size=1000,
+        duration=duration,
+        width=1280,
+        height=720,
+        fps=25.0,
+        video_fingerprint={"md5": f"md5-{vid}", "keyframe_phashes": list(phashes)},
+    )
+
+
+def _rate(deduplicator, fp, videos, session=None):
+    session_magic = MagicMock()
+    # 分片表无数据 -> 回退 JSON keyframe_phashes
+    session_magic.query.return_value.filter.return_value.order_by.return_value.all.return_value = []
+    with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+        repo = MockRepo.return_value
+        repo.list_by_project.return_value = videos
+        repo.list_by_user.return_value = videos
+        return deduplicator.compute_duplicate_rate(fp, "proj1", "new-vid", session_magic, scope="project")
+
+
+def _check(deduplicator, fp, videos, scope="project", **kw):
+    session_magic = MagicMock()
+    session_magic.query.return_value.filter.return_value.order_by.return_value.all.return_value = []
+    with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+        repo = MockRepo.return_value
+        repo.list_by_project.return_value = videos
+        repo.list_by_user.return_value = videos
+        return deduplicator.check_duplicate(fp, "proj1", session_magic, scope=scope, **kw)
+
+
+# ── P0-1/P0-2: 同源不同裁剪（距离 6~10）检出非 0 ──────────────
+
+
+class TestSameSourceDifferentCrop:
+    """同源成片：random_edge_crop 后 pHash 距离 6~10，应检出非 0 相似度。"""
+
+    def test_same_source_high_similarity_detected(self):
+
+        ddp = VideoDeduplicator()
+        # 新视频 5 个分片，每个 phash 与已有视频对应分片距离 6（< 阈值）
+        base = [_h(0) for _ in range(5)]
+        new = [_h(6) for _ in range(5)]
+        existing = _video("v-old", base, duration=11.0)
+        chunks = [_chunk(h, i * 2.2, (i + 1) * 2.2) for i, h in enumerate(new)]
+        fp = _fingerprint(new, 11.0, chunks=chunks)
+
+        result = _rate(ddp, fp, [existing], MagicMock())
+        assert result["duplicate_rate"] > 0
+        assert result["visual_similarity"] > 0
+
+    def test_same_source_distance_at_threshold_still_detected(self):
+        """距离正好等于阈值（<=）也要算匹配——阈值比较统一为 <=。"""
+
+        assert PHASH_THRESHOLD <= 12, "阈值应经校准保持在能检出同源裁剪的范围"
+        ddp = VideoDeduplicator()
+        base = [_h(0) for _ in range(6)]
+        new = [_h(PHASH_THRESHOLD) for _ in range(6)]
+        existing = _video("v-old", base, duration=12.0)
+        chunks = [_chunk(h, i * 2, (i + 1) * 2) for i, h in enumerate(new)]
+        fp = _fingerprint(new, 12.0, chunks=chunks)
+
+        result = _rate(ddp, fp, [existing], MagicMock())
+        assert result["duplicate_rate"] > 0
+
+
+# ── P0-2: 局部片段复用（B 结尾 2s ≈ A 中间 2s） ────────────────
+
+
+class TestPartialReuse:
+    def test_partial_reuse_tail_overlap_detected(self):
+        """新视频 6 片，最后 2 片命中已有视频中间 2 片（距离 4），其余不匹配。
+
+        旧逻辑 frame_match_rate=2/6≈0.33（<0.3 硬跳过边界）+ MIN_CONSECUTIVE=5
+        导致完全检不出；新逻辑 coverage 为主指标 + 自适应门槛应检出。
+        """
+
+        ddp = VideoDeduplicator()
+        # 已有 8 片：索引 3、4 是被复用的镜头
+        old = [_h(20 + i) for i in range(8)]
+        # 新视频 6 片：最后 2 片对应 old[3], old[4]，距离 4；其余距离 30
+        new = [_h(50 + i) for i in range(4)] + [_h(4)] * 2
+        # 让 new[4] 与 old[3] 距离 4、new[5] 与 old[4] 距离 4（构造近似）
+        new[4] = f"{int('1' * 4 + '0' * 60, 2):016x}"
+        new[5] = f"{int('1' * 4 + '0' * 60, 2):016x}"
+        old[3] = _h(0)
+        old[4] = _h(0)
+
+        existing = _video("v-old", old, duration=16.0)
+        chunks = [_chunk(h, i * 2, (i + 1) * 2) for i, h in enumerate(new)]
+        fp = _fingerprint(new, 12.0, chunks=chunks)
+
+        result = _rate(ddp, fp, [existing], MagicMock())
+        # 局部复用：duplicate_rate 必须非 0
+        assert result["duplicate_rate"] > 0
+
+    def test_short_video_adaptive_consecutive_threshold(self):
+        """11s/5 片短视频：MIN_CONSECUTIVE 自适应 min(5, max(2, 5//2))=2，
+        2 片连续命中即报片段（旧值 5 让短视频永远无法报片段）。"""
+
+        q = [
+            FingerprintChunk(0, 2000, "f" * 16, []),
+            FingerprintChunk(2000, 4000, "0" * 16, []),
+            FingerprintChunk(4000, 6000, f"{int('11110000', 2):016x}", []),
+        ]
+        t = [
+            FingerprintChunk(0, 2000, "f" * 16, []),
+            FingerprintChunk(2000, 4000, "0" * 16, []),
+            FingerprintChunk(4000, 6000, "e" * 16, []),
+        ]
+        # 3 片视频自适应门槛 = min(5, max(2, 3//2)) = 2
+        segs = find_duplicate_segments(q, t)
+        assert len(segs) >= 1
+
+
+# ── P0-3: ±1 邻接窗口对齐 ─────────────────────────────────────
+
+
+class TestNeighborAlignment:
+    def test_neighbor_window_absorbs_boundary_jitter(self):
+        """切点错位导致目标索引偏移 ±1 时，连续匹配不应被中断。"""
+
+        q = [FingerprintChunk(i * 1000, (i + 1) * 1000, f"{i:016x}", []) for i in range(4)]
+        # 目标：前 3 片与 q 相同，但第 3 片最佳匹配偏移 +1（t[4]），t[3] 是无关内容
+        t_hashes = [f"{i:016x}" for i in range(3)] + ["f" * 16, f"{3:016x}"]
+        t = [FingerprintChunk(i * 1000, (i + 1) * 1000, h, []) for i, h in enumerate(t_hashes)]
+        segs = find_duplicate_segments(q, t)
+        # q[0],q[1] 精确匹配 t[0],t[1]；q[2]->t[2]；q[3]->t[4]（步进 2，窗口 ±1 内）
+        assert len(segs) >= 1
+        assert segs[0].query_end_ms >= 3000
+
+
+# ── P0-5 / 验收：异源不误报 ───────────────────────────────────
+
+
+class TestDifferentSourceNoFalsePositive:
+    def test_unrelated_videos_near_zero(self):
+
+        ddp = VideoDeduplicator()
+        # 异源：所有分片距离 >= 20
+        old = [_h(40 + i * 3 % 20) for i in range(6)]
+        new = [_h(0 + i) for i in range(6)]
+        existing = _video("v-old", old, duration=12.0)
+        chunks = [_chunk(h, i * 2, (i + 1) * 2) for i, h in enumerate(new)]
+        fp = _fingerprint(new, 12.0, chunks=chunks)
+
+        result = _rate(ddp, fp, [existing], MagicMock())
+        assert result["duplicate_rate"] == 0
+        assert result["visual_similarity"] < 0.7
+        assert result["match_count"] == 0
+
+    def test_check_duplicate_returns_none_for_unrelated(self):
+
+        ddp = VideoDeduplicator()
+        old = [_h(40 + i) for i in range(6)]
+        new = [_h(i) for i in range(6)]
+        existing = _video("v-old", old, duration=12.0)
+        fp = _fingerprint(new, 12.0)
+
+        result = _check(ddp, fp, [existing])
+        assert result is None
+
+
+# ── N=1 不回归 ────────────────────────────────────────────────
+
+
+class TestSingleChunkNoRegression:
+    def test_single_chunk_identical_detected(self):
+
+        ddp = VideoDeduplicator()
+        h = _h(2)
+        existing = _video("v-old", [h], duration=3.0)
+        chunks = [_chunk(h, 0, 3000)]
+        fp = _fingerprint([h], 3.0, chunks=chunks)
+        result = _rate(ddp, fp, [existing], MagicMock())
+        assert result["duplicate_rate"] > 0
+
+    def test_single_chunk_md5_exact_match(self):
+
+        ddp = VideoDeduplicator()
+        existing = _video("v-old", [_h(0)], duration=3.0)
+        existing.video_fingerprint["md5"] = "same"
+        fp = _fingerprint([_h(0)], 3.0, md5="same")
+        result = _check(ddp, fp, [existing])
+        assert result is not None
+        assert result["reason"] == "exact_md5_match"
+
+
+# ── P1-6: 时长预过滤单位 bug ──────────────────────────────────
+
+
+class TestDurationPrefilterUnit:
+    def test_duration_sec_not_divided_by_1000(self):
+        """fingerprint.duration 单位是秒，传给 check_duplicate 不应再 /1000。
+
+        旧 bug：duration/1000 → duration_max≈0.0135s，所有真实视频被过滤。
+        """
+
+        ddp = VideoDeduplicator()
+        fp = _fingerprint([_h(0)], 13.5)
+        session_magic = MagicMock()
+        session_magic.query.return_value.filter.return_value.order_by.return_value.all.return_value = []
+        with patch("video_processing.dedup.SQLAlchemyGeneratedVideoRepository") as MockRepo:
+            repo = MockRepo.return_value
+            repo.list_by_user.return_value = []
+            ddp.check_duplicate(fp, "proj1", session_magic, scope="user", user_id="u1", duration_sec=fp.duration)
+            _, kwargs = repo.list_by_user.call_args
+            # ±15% 窗口：13.5s -> [11.475, 15.525]
+            assert 11.0 < kwargs["duration_min"] < 12.0
+            assert 15.0 < kwargs["duration_max"] < 16.0
+
+
+# ── P1-7: 颜色直方图归一化 ────────────────────────────────────
+
+
+class TestHistogramNormalization:
+    def test_bhattacharyya_coefficient_in_unit_range(self):
+        """Bhattacharyya 系数必须在 [0,1]（旧 L2 + 3 通道拼接算出 ~14.9）。"""
+
+        # 3 通道拼接、每通道概率分布（Σ=1）
+        hist_a = [0.5, 0.5] + [0.0] * 94 + [0.5, 0.5] + [0.0] * 94 + [0.5, 0.5] + [0.0] * 94
+        # 长度裁剪到 96（3 通道 × 32 bins）
+        hist_a = ([0.5, 0.5] + [0.0] * 30) * 3
+        hist_b = ([0.5, 0.5] + [0.0] * 30) * 3
+
+        coeff = VideoDeduplicator._bhattacharyya_coefficient(hist_a, hist_b)
+        assert 0.0 <= coeff <= 1.0
+        assert coeff > 0.99  # 完全相同 -> 1.0
+
+    def test_bhattacharyya_disjoint_hist_low(self):
+
+        hist_a = ([1.0] + [0.0] * 31) * 3
+        hist_b = ([0.0] * 31 + [1.0]) * 3
+        coeff = VideoDeduplicator._bhattacharyya_coefficient(hist_a, hist_b)
+        assert coeff < 0.05
+
+
+# ── P1-8: temporal_coverage 量纲 ──────────────────────────────
+
+
+class TestTemporalCoverageUnits:
+    def test_coverage_uses_milliseconds(self):
+        """命中片段 6s / 视频 12s -> coverage=0.5；旧 bug 把 duration（秒）当毫秒，
+        covered_ms(6000)/duration(12) = 500 -> min(1.0)=1.0 误判 100% 覆盖。"""
+
+        ddp = VideoDeduplicator()
+        old = [_h(0) for _ in range(6)]
+        new = [_h(0) for _ in range(3)] + [_h(30) for _ in range(3)]
+        existing = _video("v-old", old, duration=12.0)
+        # 新视频 12s，前 6s（3 片）与 old 相同
+        chunks = [_chunk(h, i * 2, (i + 1) * 2) for i, h in enumerate(new)]
+        fp = _fingerprint(new, 12.0, chunks=chunks)
+        result = _rate(ddp, fp, [existing], MagicMock())
+        # coverage 应约 0.5（3 片 × 2s = 6s / 12s），duplicate_rate ≈ (0.5*0.4 + 0.5*0.6)*100 = 50
+        assert 30 < result["duplicate_rate"] < 70
+
+
+# ── P1-9: 阈值比较统一 ────────────────────────────────────────
+
+
+class TestThresholdConsistency:
+    def test_frame_and_segment_thresholds_same_source(self):
+
+        assert SEGMENT_MATCH_THRESHOLD == PHASH_THRESHOLD
+        assert VideoDeduplicator.PHASH_THRESHOLD == PHASH_THRESHOLD
+
+
+# ── P2: 0 匹配也要有日志痕迹 ──────────────────────────────────
+
+
+class TestZeroMatchLogging:
+    def test_no_match_emits_info_log(self, caplog):
+
+        ddp = VideoDeduplicator()
+        old = [_h(40 + i) for i in range(5)]
+        existing = _video("v-old", old, duration=10.0)
+        fp = _fingerprint([_h(i) for i in range(5)], 10.0)
+
+        with caplog.at_level(logging.INFO, logger="video_processing.dedup"):
+            result = _check(ddp, fp, [existing])
+        assert result is None
+        assert any("no match" in r.message for r in caplog.records)
+
+
+# ── recompute 任务下载路径（#1702 连带修复：旧硬编码 key 404） ─────
+
+
+class TestRecomputeDownloadPath:
+    """recompute-dedup 走 check_duplicate_task，需要从 OSS 重新下载成片。
+
+    旧代码硬编码 projects/{pid}/generated/{vid}/{vid}.mp4（从不存在），
+    真实 key 在 file_url：generated/projects/{pid}/tasks/{tid}/rendered_*.mp4。
+    """
+
+    def test_task_downloads_from_file_url(self):
+        import inspect
+
+        import video_processing.dedup as dedup_mod
+
+        source = inspect.getsource(dedup_mod.check_duplicate_task)
+        # 下载 key 必须来自 video.file_url
+        assert 'getattr(video, "file_url"' in source or "video.file_url" in source
+        # 旧的硬编码 key 只能作为回退存在，不能是主路径
+        assert "falling back to legacy key" in source
+        # download_file 接收的是派生 key 而非硬编码 f-string
+        assert "storage_service.download_file(download_key" in source
+        assert '/generated/{generated_video_id}/{generated_video_id}.mp4"' not in source.replace(
+            'download_key = f"projects/{video.project_id}/generated/{generated_video_id}/{generated_video_id}.mp4"',
+            "",
+        )

tests/unit/test_dedup_engine.py

@@ -358,11 +358,11 @@ class TestVideoDeduplicatorCheckDuplicate:
         finally:
             self._restore_repo(mod, orig)
 
-    def test_first_match_returned(self, deduplicator, mock_session):
-        """返回第一个通过阈值的匹配（非最优匹配）。"""
-        # vid-1: 距离=2 bits（0x03 XOR 0x01 = 0x02 → 1 bit），通过阈值
+    def test_highest_score_match_returned(self, deduplicator, mock_session):
+        """Issue #1702: 遍历所有候选取融合分最高者（旧逻辑首个过阈即返回）。"""
+        # vid-1: 距离=1 bit（0x03 XOR 0x01 = 0x02 → 1 bit），通过阈值
         vid1 = self._make_existing_video("vid-1", "md5_1", phashes=["0000000000000003"])
-        # vid-2: 距离=0 bits（完全匹配）
+        # vid-2: 距离=0 bits（完全匹配），融合分更高
         vid2 = self._make_existing_video("vid-2", "md5_2", phashes=["0000000000000001"])
 
         mock_repo = MagicMock()
@@ -380,8 +380,8 @@ class TestVideoDeduplicatorCheckDuplicate:
         try:
             result = deduplicator.check_duplicate(fingerprint, "proj-1", mock_session)
             assert result is not None
-            # 返回第一个通过阈值的匹配（vid-1 距离=1 < 10）
-            assert result["duplicate_of"] == "vid-1"
+            # 两个候选都过阈，返回融合分最高的 vid-2（距离 0 < 1）
+            assert result["duplicate_of"] == "vid-2"
         finally:
             self._restore_repo(mod, orig)
 

tests/unit/test_dedup_pure.py

@@ -185,11 +185,14 @@ class TestBhattacharyyaCoefficient:
     """_bhattacharyya_coefficient Bhattacharyya 系数测试."""
 
     def test_identical_histograms(self):
-        """完全相同的直方图系数为1.0."""
-        hist = [0.5, 0.5, 0.0, 0.3]
+        """完全相同的直方图系数为1.0（#1702：按 Σ 归一，概率分布语义）。"""
+        hist = [0.5, 0.5, 0.0, 0.0]  # Σ=1 的概率分布
         bc = VideoDeduplicator._bhattacharyya_coefficient(hist, hist)
-        # Σ √(a[i]*a[i]) = Σ a[i] = 1.0 (normalized)
-        assert bc == pytest.approx(sum(h for h in hist))
+        assert bc == pytest.approx(1.0)
+        # 非归一化输入也归一到 1.0（三通道拼接 Σ=3 的等价情形）
+        hist3 = [0.5, 0.5, 0.0, 0.3]
+        bc3 = VideoDeduplicator._bhattacharyya_coefficient(hist3, hist3)
+        assert bc3 == pytest.approx(1.0)
 
     def test_zero_histograms(self):
         """全零直方图系数为0."""
@@ -202,10 +205,10 @@ class TestBhattacharyyaCoefficient:
         assert bc == pytest.approx(0.0)
 
     def test_different_lengths(self):
-        """不同长度直方图取最小长度对齐."""
+        """不同长度直方图取最小长度对齐，并按各自总量归一（#1702 概率分布语义）。"""
+        # 对齐到前 2 维：coeff = 2，norm = √(Σa·Σb) = √(2·2) = 2 → 1.0
         bc = VideoDeduplicator._bhattacharyya_coefficient([1.0, 1.0, 0.0, 0.0], [1.0, 1.0])
-        # 对齐到前2维: √(1*1) + √(1*1) = 2.0
-        assert bc == pytest.approx(2.0)
+        assert bc == pytest.approx(1.0)
 
     def test_known_value(self):
         """已知值验证."""

tests/unit/test_dedup_v2.py

@@ -484,8 +484,10 @@ class TestBackwardCompatibility:
         chunks_b = [{"phash_binary": "aaaaaaaaaaaaaaaa", "start_time_ms": 0, "end_time_ms": 5000}]
 
         segments = find_duplicate_segments(chunks_a, chunks_b)
-        # 1 帧 < min_consecutive=5，不会报重复
-        assert segments == []
+        # Issue #1702: 自适应门槛 min(5, max(2, 1//2))=2，1 帧不成段；
+        # N=1 的检出由 _evaluate_candidate 匹配帧回退兜底（见 test_dedup_1702）。
+        # 这里只要求不崩溃。
+        assert isinstance(segments, list)
 
 
 # ── TestConstants ───────────────────────────────────────────────
@@ -495,8 +497,9 @@ class TestConstants:
     """常量值验证 — 使用已在模块顶部导入的常量，避免重新 import."""
 
     def test_segment_match_threshold(self):
-        # 从已导入的 find_duplicate_segments 默认参数间接验证
-        assert SEGMENT_MATCH_THRESHOLD == 8
+        # Issue #1702: pHash 阈值经 staging 真实同源/异源指纹回归校准
+        # （同源密集采样 min=8、异源 min=24），统一为模块常量 PHASH_THRESHOLD=12。
+        assert SEGMENT_MATCH_THRESHOLD == 12
 
     def test_min_consecutive_matches(self):
         assert MIN_CONSECUTIVE_MATCHES == 5

tests/unit/test_fingerprint_chunks.py

@@ -3,7 +3,7 @@
 覆盖：
 - 分片策略：60秒视频 → 30片，120秒视频 → 24片
 - VideoFingerprint.to_chunk_models() 输出正确
-- _save_fingerprint_chunks 幂等性（已有数据跳过）
+- _save_fingerprint_chunks 替换语义（Issue #1702：重算时先删旧分片再写入）
 - to_dict() 向后兼容
 """
 
@@ -169,11 +169,15 @@ class TestVideoFingerprintToChunkModels:
         assert models == []
 
 
-class TestSaveFingerprintChunksIdempotent:
-    """测试 _save_fingerprint_chunks 幂等性。"""
+class TestSaveFingerprintChunksReplace:
+    """测试 _save_fingerprint_chunks 替换语义（Issue #1702）。
 
-    def test_save_skips_existing(self):
-        """已有分片数据时跳过写入。"""
+    重算查重时指纹算法已升级（中心裁剪 + 新采样/阈值），旧分片必须先删除
+    再写入新分片，否则 recompute-dedup 永远读到旧指纹、修复对存量视频不生效。
+    """
+
+    def test_save_replaces_existing(self):
+        """已有分片数据时：先删除旧分片，再写入新分片。"""
         fp = VideoFingerprint(
             md5="abc",
             keyframe_phashes=["a1b2"],
@@ -186,16 +190,22 @@ class TestSaveFingerprintChunksIdempotent:
         )
 
         session = MagicMock()
-        # Mock: 已有 1 条分片数据
-        session.query.return_value.filter.return_value.count.return_value = 1
+        # Mock: 删除旧分片返回 3（旧算法留下的 3 条分片）
+        session.query.return_value.filter.return_value.delete.return_value = 3
 
         _save_fingerprint_chunks(fp, video_id="v1", project_id="p1", user_id="u1", session=session)
 
-        # bulk_save_objects 不应被调用
-        session.bulk_save_objects.assert_not_called()
+        # 必须先执行删除
+        session.query.return_value.filter.return_value.delete.assert_called_once()
+        # 新分片必须写入
+        session.bulk_save_objects.assert_called_once()
+        saved_models = session.bulk_save_objects.call_args[0][0]
+        assert len(saved_models) == 1
+        assert saved_models[0].video_id == "v1"
+        assert saved_models[0].phash_binary == "a1b2"
 
     def test_save_writes_new(self):
-        """无分片数据时写入。"""
+        """无旧分片时直接写入。"""
         fp = VideoFingerprint(
             md5="abc",
             keyframe_phashes=["a1b2"],
@@ -208,12 +218,12 @@ class TestSaveFingerprintChunksIdempotent:
         )
 
         session = MagicMock()
-        # Mock: 无分片数据
-        session.query.return_value.filter.return_value.count.return_value = 0
+        # Mock: 无旧分片
+        session.query.return_value.filter.return_value.delete.return_value = 0
 
         _save_fingerprint_chunks(fp, video_id="v1", project_id="p1", user_id="u1", session=session)
 
-        # bulk_save_objects 应被调用一次
+        session.query.return_value.filter.return_value.delete.assert_called_once()
         session.bulk_save_objects.assert_called_once()
         saved_models = session.bulk_save_objects.call_args[0][0]
         assert len(saved_models) == 1
@@ -221,7 +231,7 @@ class TestSaveFingerprintChunksIdempotent:
         assert saved_models[0].phash_binary == "a1b2"
 
     def test_save_skips_no_chunks(self):
-        """指纹无 chunks 时跳过。"""
+        """指纹无 chunks 时跳过（不删不写）。"""
         fp = VideoFingerprint(
             md5="abc",
             keyframe_phashes=[],
@@ -232,11 +242,11 @@ class TestSaveFingerprintChunksIdempotent:
         )
 
         session = MagicMock()
-        session.query.return_value.filter.return_value.count.return_value = 0
 
         _save_fingerprint_chunks(fp, video_id="v1", project_id="p1", user_id="u1", session=session)
 
-        # bulk_save_objects 不应被调用
+        # 无 chunks：不查询、不删除、不写入
+        session.query.assert_not_called()
         session.bulk_save_objects.assert_not_called()
 
 

tests/unit/test_generation_preview.py

@@ -725,8 +725,12 @@ class TestCreatePreviewRoute:
                     generation_task_repository=repo,
                     db=MagicMock(),
                 )
-        assert resp.task_id == "preview_task_001"
-        assert resp.status == "pending"
+        # 批量响应：N=1 时 items 长度为 1
+        assert resp.total == 1
+        assert len(resp.items) == 1
+        assert resp.items[0].task_id == "preview_task_001"
+        assert resp.items[0].status == "pending"
+        assert resp.items[0].variant_index == 0
 
     def test_user_pending_limit_exceeded(self):
         """用户待处理任务超限 → 429"""
@@ -807,7 +811,13 @@ class TestCreatePreviewRoute:
         repo.count_pending_total.return_value = 0
 
         task = _make_task()
-        from fastapi import HTTPException
+
+        # 模拟 mark_failed 真实更新任务状态（_mark_task_failed 内部调用）
+        def _set_failed(error_message="", **_kwargs):
+            task.status = GenerationTaskStatus.FAILED
+            task.error_message = error_message
+
+        task.mark_failed.side_effect = _set_failed
 
         with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
             MockUC.return_value.execute.return_value = task
@@ -815,14 +825,15 @@ class TestCreatePreviewRoute:
                 "app.api.routes.generation_preview.safe_enqueue_generation_task",
                 return_value=False,
             ):
-                with pytest.raises(HTTPException) as exc_info:
-                    create_preview_generation_task(
-                        self._make_request(),
-                        authenticated_user=_make_user(),
-                        generation_task_repository=repo,
-                        db=MagicMock(),
-                    )
-                assert exc_info.value.status_code == 500
+                resp = create_preview_generation_task(
+                    self._make_request(),
+                    authenticated_user=_make_user(),
+                    generation_task_repository=repo,
+                    db=MagicMock(),
+                )
+            # 入队失败：任务被标记 failed（mark_failed 设置错误信息），响应正常返回
+            assert resp.total == 1
+            assert resp.items[0].status == "failed"
 
     def test_enqueue_raises_user_limit(self):
         """safe_enqueue 抛出 UserPendingLimitExceeded → 429"""
@@ -833,6 +844,12 @@ class TestCreatePreviewRoute:
         task = _make_task()
         from fastapi import HTTPException
 
+        def _set_failed_limit(error_message="", **_kwargs):
+            task.status = GenerationTaskStatus.FAILED
+            task.error_message = error_message or "待处理任务超限"
+
+        task.mark_failed.side_effect = _set_failed_limit
+
         with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
             MockUC.return_value.execute.return_value = task
             with patch(
@@ -846,6 +863,7 @@ class TestCreatePreviewRoute:
                         generation_task_repository=repo,
                         db=MagicMock(),
                     )
+                # 全部变体入队失败且错误消息含"待处理任务" → 429
                 assert exc_info.value.status_code == 429
 
     def test_enqueue_raises_global_queue_full(self):
@@ -857,6 +875,12 @@ class TestCreatePreviewRoute:
         task = _make_task()
         from fastapi import HTTPException
 
+        def _set_failed_queue(error_message="", **_kwargs):
+            task.status = GenerationTaskStatus.FAILED
+            task.error_message = error_message or "系统队列已满"
+
+        task.mark_failed.side_effect = _set_failed_queue
+
         with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
             MockUC.return_value.execute.return_value = task
             with patch(
@@ -870,6 +894,7 @@ class TestCreatePreviewRoute:
                         generation_task_repository=repo,
                         db=MagicMock(),
                     )
+                # 全部变体入队失败且错误消息含"队列" → 503
                 assert exc_info.value.status_code == 503
 
 

tests/unit/test_phash_threshold_calibration_1658.py

@@ -102,6 +102,7 @@ from video_processing.dedup import (  # noqa: E402
     DUPLICATE_THRESHOLD,
     HISTOGRAM_WEIGHT,
     MATCH_RATIO_THRESHOLD,
+    PHASH_THRESHOLD,
     PHASH_WEIGHT,
     VideoDeduplicator,
 )
@@ -128,11 +129,15 @@ _ZERO_HIST = [0.0] * 96  # 全黑视频的全零直方图（有效数据）
 
 
 class TestThresholdCalibration:
-    """pHash 阈值由 10 收紧到 8（Issue #1658）。"""
+    """pHash 阈值校准（Issue #1658 收紧到 8，Issue #1702 经真实指纹分布重校准为 12）。
 
-    def test_phash_threshold_is_8(self):
-        """PHASH_THRESHOLD 必须为 8（旧值 10 会放过 8~9 汉明距离的不同视频）。"""
-        assert VideoDeduplicator.PHASH_THRESHOLD == 8
+    #1702 staging 离线实验：同帧两次 2-5% 随机裁剪距离 4~10；同源成片（密集 1s
+    采样）最小距离 8、<=12 命中 10/31；异源成片最小距离 24。8 会漏检同源裁剪，
+    12 检出同源且与异源分布（>=24）间隔充足。
+    """
+
+    def test_phash_threshold_is_calibrated(self):
+        assert VideoDeduplicator.PHASH_THRESHOLD == PHASH_THRESHOLD == 12
 
     def test_match_ratio_threshold_constant(self):
         assert MATCH_RATIO_THRESHOLD == 0.7
@@ -144,22 +149,21 @@ class TestThresholdCalibration:
         assert PHASH_WEIGHT == 0.7
         assert HISTOGRAM_WEIGHT == 0.3
 
-    def test_threshold_tightening_excludes_distance_8_and_9(self):
-        """距离 8、9 的帧：旧阈值 10 下算匹配，新阈值 8 下不算匹配。
+    def test_threshold_matching_semantics(self):
+        """阈值比较统一为 <=（帧匹配与片段匹配同一口径）。
 
-        场景：5 个关键帧距离为 [7, 7, 7, 9, 9]。
-        - 旧阈值 10：5 帧全部 < 10 → match_ratio = 1.0（误放过）
-        - 新阈值 8：仅 3 帧 < 8 → match_ratio = 0.6 < 0.7（正确跳过）
+        场景：5 个关键帧距离为 [10, 12, 12, 24, 26]。
+        - <=12（#1702 校准阈值）：3 帧匹配 → 0.6 < 0.7 被帧比例门槛拦截异源
+        - 距离 12 的同源裁剪帧应算匹配（< 与 <= 口径统一）
         """
-        distances = [7, 7, 7, 9, 9]
+        distances = [10, 12, 12, 24, 26]
+        matched = sum(1 for d in distances if d <= VideoDeduplicator.PHASH_THRESHOLD)
+        assert matched == 3
+        assert matched / len(distances) == 0.6
+        assert matched / len(distances) < MATCH_RATIO_THRESHOLD
 
-        matched_old = sum(1 for d in distances if d < 10)
-        assert matched_old == 5  # 旧行为：全匹配 → 误判风险
-
-        matched_new = sum(1 for d in distances if d < VideoDeduplicator.PHASH_THRESHOLD)
-        assert matched_new == 3
-        assert matched_new / len(distances) == 0.6
-        assert matched_new / len(distances) < MATCH_RATIO_THRESHOLD  # 被帧比例门槛拦截
+        # 异源典型距离（>=24）绝不匹配
+        assert not any(d <= VideoDeduplicator.PHASH_THRESHOLD for d in (24, 26, 30))
 
 
 # ── TestComputeFusionScore：统一融合得分方法 ────────────────────