feat(#1970): 智能降重 PR2 - dedup_enabled 开关 + 6 维片段级微变换 (#1975)

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

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

@@ -167,11 +167,15 @@ def _writeback_edit_plan_config(
     task_id: str,
     title_config: dict | None,
     db: Session,
+    dedup_enabled: bool | None = None,
+    video_index: int | None = None,
 ) -> None:
     """[已下沉] 路由层兼容别名 → app.services.generation_common.writeback_edit_plan_config。"""
     from app.services.generation_common import writeback_edit_plan_config
 
-    return writeback_edit_plan_config(plan_id, task_id, title_config, db)
+    return writeback_edit_plan_config(
+        plan_id, task_id, title_config, db, dedup_enabled=dedup_enabled, video_index=video_index
+    )
 
 
 def _resolve_project_and_library(
@@ -332,6 +336,7 @@ def create_generation_task(
                             task_id=preview_task.id,
                             title_config=fallback_title_config,
                             db=db,
+                            dedup_enabled=request.dedup_enabled,
                         )
 
                     logger.info(
@@ -481,9 +486,7 @@ def create_generation_task(
 
             _batch_segments = _collect_segments(_plan0.id, _plan_svc._clip_repo)
             # #1970：批次内原子片段硬避让集合
-            _batch_atom_ids: list[str] = _collect_atom_ids(
-                _plan0.id, _plan_svc._clip_repo
-            )
+            _batch_atom_ids: list[str] = _collect_atom_ids(_plan0.id, _plan_svc._clip_repo)
 
             # 变体 1..N-1 独立选片（传入累积batch_segments做素材区间避让）
             for task_index in range(1, count):
@@ -536,9 +539,7 @@ def create_generation_task(
                     for _aid, _ivs in _new_segs.items():
                         _batch_segments.setdefault(_aid, []).extend(_ivs)
                     # #1970：同步累积原子片段ID
-                    _batch_atom_ids.extend(
-                        _collect_atom_ids(variant.id, _plan_svc._clip_repo)
-                    )
+                    _batch_atom_ids.extend(_collect_atom_ids(variant.id, _plan_svc._clip_repo))
                 except Exception:
                     logger.exception("[生成任务] 变体%d 区间收集失败（不阻断）", task_index)
 
@@ -682,6 +683,8 @@ def create_generation_task(
                         task_id=task.id,
                         title_config=variant_title_config,
                         db=db,
+                        dedup_enabled=request.dedup_enabled,
+                        video_index=task_index,
                     )
 
                 if safe_enqueue_generation_task(
@@ -772,6 +775,7 @@ def confirm_generation(
             generation_task_repository.update(source_task)
 
             # 同步标题到 EditPlan.config
+            # #1970：确认生成复用预览计划，dedup_enabled 沿用计划已有值，不在此覆盖
             if confirmed_title_config and source_task.source_edit_plan_id:
                 _writeback_edit_plan_config(
                     plan_id=source_task.source_edit_plan_id,

apps/api/app/schemas/generation_task.py

@@ -98,6 +98,11 @@ class CreateGenerationTaskRequest(BaseModel):
         description="各变体独立标题文字数组：长度1=共用，长度=count=独立。为空时使用 title_config.text",
     )
 
+    # ── 智能降重开关（#1970）──
+    # True（默认）：edge_crop + 片段级微变换（hflip/变速/亮度/对比度/饱和度/BGM偏移）全部生效；
+    # False：跳过 edge_crop、不注入微变换，渲染确定性（固定种子）。
+    dedup_enabled: bool = Field(default=True, description="智能降重开关，默认开启；关闭后跳过边缘裁切与微变换")
+
     @model_validator(mode="after")
     def _check_variant_arrays(self) -> "CreateGenerationTaskRequest":
         """变体数组字段长度校验 + #1749 配音严格守卫。

apps/api/app/services/generation_common.py

@@ -61,10 +61,13 @@ def writeback_edit_plan_config(
     task_id: str,
     title_config: dict | None,
     db: Session,
+    dedup_enabled: bool | None = None,
+    video_index: int | None = None,
 ) -> None:
     """任务入队成功后，回写 EditPlan.config：generation_task_id + title_config。
 
     用 merge 方式更新，不整体覆盖 config，避免丢失其他字段。
+    #1970：dedup_enabled 非 None 时一并写入，worker 据此决定 edge_crop/微变换。
     失败只记日志，不影响任务创建。
     """
     if not plan_id:
@@ -80,6 +83,10 @@ def writeback_edit_plan_config(
         current_config = plan_model.config if isinstance(plan_model.config, dict) else {}
         merged = dict(current_config)
         merged["generation_task_id"] = task_id
+        if dedup_enabled is not None:
+            merged["dedup_enabled"] = bool(dedup_enabled)
+        if video_index is not None:
+            merged["video_index"] = int(video_index)
 
         if title_config:
             # #1901 统一字段名为 "title"（worker sync_configs_to_plan 写的是 "title"）

apps/worker/video_processing/micro_transform_pure.py

@@ -0,0 +1,176 @@
+"""智能降重微变换纯逻辑模块 — #1970 PR2.
+
+所有函数均为纯函数：不调用 FFmpeg、不读写文件，只负责按可复现种子
+生成每个片段 / 整片的微变换参数与 filter_complex 片段。
+
+6 个维度：
+1. hflip 水平翻转（每片段 50%，有字幕/文字的片段不翻转）
+2. 播放速度 0.97~1.03x（视频 setpts + 音频 atempo）
+3. 亮度 ±2%（eq=brightness）
+4. 对比度 ±2%（eq=contrast）
+5. 饱和度 ±2%（eq=saturation）
+6. BGM 起始偏移 2~8 秒（音频 atrim 起点）
+
+随机种子 = hash(task_id + video_index) % 10000，保证同一任务同一视频
+可复现；dedup_enabled=False 时不生成本模块任何输出。
+"""
+
+from __future__ import annotations
+
+import random
+from dataclasses import dataclass, field
+
+# ── 常量（与需求文档 §2 对齐）──────────────────────────────────────────────────
+
+SPEED_MIN = 0.97
+SPEED_MAX = 1.03
+COLOR_DELTA = 0.02
+HFLIP_PROBABILITY = 0.5
+BGM_OFFSET_MIN = 2.0
+BGM_OFFSET_MAX = 8.0
+SEED_MODULO = 10000
+
+
+def make_video_seed(task_id: str, video_index: int) -> int:
+    """生成视频级可复现种子：hash(task_id+video_index) % 10000。
+
+    用 sha256 而非内置 hash()：内置 hash 对字符串带进程级随机盐（PYTHONHASHSEED），
+    跨进程不可复现。结果映射到 0~9999。
+    """
+    import hashlib
+
+    raw = f"{task_id or ''}:{int(video_index)}"
+    digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()
+    return int(digest[:8], 16) % SEED_MODULO
+
+
+@dataclass(slots=True)
+class ClipMicroTransform:
+    """单个片段的微变换参数。"""
+
+    clip_index: int
+    hflip: bool = False
+    speed: float = 1.0
+    brightness: float = 0.0
+    contrast: float = 1.0
+    saturation: float = 1.0
+    has_text: bool = False
+
+    def video_filter_suffix(self) -> str:
+        """返回追加在片段视频处理链上的 filter 后缀（无末尾标签）。
+
+        顺序：trim/setpts（已有）→ 调速 setpts → hflip → eq → format。
+        调速的 setpts 必须位于 trim 之后；hflip/eq 在缩放之后即可，
+        concat_engine 按「调速 → hflip → eq」顺序拼接到 scale/fps 之前的
+        trim 之后、scale 之后均可，这里只产出独立步骤、由引擎决定插入点。
+        """
+        parts: list[str] = []
+        # 速度：setpts=PTS/speed（speed>1 时画面加速，时间戳变小）
+        if abs(self.speed - 1.0) > 1e-4:
+            parts.append(f"setpts=PTS/{self.speed:.5f}")
+        # 水平翻转：有文字/字幕片段不翻转
+        if self.hflip and not self.has_text:
+            parts.append("hflip")
+        # 色彩微调：brightness 取值 -1~1（±0.02），contrast/saturation 围绕 1.0
+        if abs(self.brightness) > 1e-4 or abs(self.contrast - 1.0) > 1e-4 or abs(self.saturation - 1.0) > 1e-4:
+            parts.append(
+                f"eq=brightness={self.brightness:+.4f}:"
+                f"contrast={self.contrast:.4f}:saturation={self.saturation:.4f}"
+            )
+        return ",".join(parts)
+
+    def audio_filter_suffix(self) -> str:
+        """返回片段音频链上的调速 filter（atempo），无调速时返回空串。"""
+        if abs(self.speed - 1.0) <= 1e-4:
+            return ""
+        return f"atempo={self.speed:.5f}"
+
+
+@dataclass(slots=True)
+class VideoMicroTransformPlan:
+    """一个成片视频的全部微变换参数。"""
+
+    task_id: str
+    video_index: int
+    seed: int
+    clips: list[ClipMicroTransform] = field(default_factory=list)
+    bgm_start_offset: float = 0.0
+
+    def clip(self, index: int) -> ClipMicroTransform | None:
+        for c in self.clips:
+            if c.clip_index == index:
+                return c
+        return None
+
+
+def _draw_speed(rng: random.Random) -> float:
+    return round(rng.uniform(SPEED_MIN, SPEED_MAX), 5)
+
+
+def _draw_signed_delta(rng: random.Random) -> float:
+    return round(rng.uniform(-COLOR_DELTA, COLOR_DELTA), 4)
+
+
+def build_micro_transform_plan(
+    task_id: str,
+    video_index: int,
+    clip_count: int,
+    *,
+    clip_has_text: list[bool] | None = None,
+    enable_bgm_offset: bool = True,
+) -> VideoMicroTransformPlan:
+    """按可复现种子生成整片的微变换计划。
+
+    Args:
+        task_id: 生成任务 ID（种子输入）
+        video_index: 视频在批次中的序号（0 起）
+        clip_count: 片段数量
+        clip_has_text: 每个片段是否有字幕/文字轨道（True 的片段不翻转）；
+                       None 时按 P1 约定视为无可靠文字检测——保守起见 hflip 一律关闭
+        enable_bgm_offset: 是否生成 BGM 起始偏移（无 BGM 时调用方可忽略该值）
+
+    Returns:
+        VideoMicroTransformPlan
+    """
+    seed = make_video_seed(task_id, video_index)
+    rng = random.Random(seed)
+
+    # P1 字幕检测约定：无法判断片段是否有文字时，一律不翻转（宁可少一个维度也不误翻字幕）
+    safe_has_text = clip_has_text if clip_has_text is not None else [True] * max(clip_count, 0)
+
+    clips: list[ClipMicroTransform] = []
+    for i in range(max(clip_count, 0)):
+        has_text = bool(safe_has_text[i]) if i < len(safe_has_text) else True
+        do_hflip = (not has_text) and rng.random() < HFLIP_PROBABILITY
+        clips.append(
+            ClipMicroTransform(
+                clip_index=i,
+                hflip=do_hflip,
+                speed=_draw_speed(rng),
+                brightness=_draw_signed_delta(rng),
+                contrast=round(1.0 + _draw_signed_delta(rng), 4),
+                saturation=round(1.0 + _draw_signed_delta(rng), 4),
+                has_text=has_text,
+            )
+        )
+
+    bgm_offset = rng.uniform(BGM_OFFSET_MIN, BGM_OFFSET_MAX) if enable_bgm_offset else 0.0
+    return VideoMicroTransformPlan(
+        task_id=task_id,
+        video_index=video_index,
+        seed=seed,
+        clips=clips,
+        bgm_start_offset=round(bgm_offset, 3),
+    )
+
+
+def build_bgm_offset_trim(start_offset: float, bgm_duration: float) -> str:
+    """生成 BGM 起始偏移的 atrim 片段。
+
+    偏移超出 BGM 长度时回退为 0（从头播放），避免空输入。
+    返回的字符串形如 "atrim=start=3.200,"，可拼到 BGM filter chain 最前面；
+    无需偏移时返回空串。
+    """
+    if start_offset <= 0 or bgm_duration <= 0 or start_offset >= bgm_duration - 0.5:
+        return ""
+    return f"atrim=start={start_offset:.3f},"

apps/worker/video_processing/render_audio.py

@@ -98,6 +98,7 @@ def mix_audio(
     bgm_path: str | None = None,
     bgm_config: dict | None = None,
     audio_tracks_config: dict | None = None,
+    bgm_start_offset: float = 0.0,
 ) -> Path | None:
     """音频后处理混音.
 
@@ -157,7 +158,10 @@ def mix_audio(
         if bgm_path and bgm_config and isinstance(bgm_config, dict) and bgm_config.get("enabled", False):
             from video_processing.bgm_mixer import BGMConfig, build_bgm_only
 
-            bgm_cfg = BGMConfig.from_config_dict(bgm_path, bgm_config)
+            _bgm_cfg_dict = dict(bgm_config or {})
+            if bgm_start_offset and not _bgm_cfg_dict.get("audio_offset"):
+                _bgm_cfg_dict["audio_offset"] = round(float(bgm_start_offset), 3)
+            bgm_cfg = BGMConfig.from_config_dict(bgm_path, _bgm_cfg_dict)
             try:
                 return build_bgm_only(ctx, bgm_cfg, video_duration)
             except Exception:
@@ -187,7 +191,10 @@ def mix_audio(
     if bgm_path and bgm_config and isinstance(bgm_config, dict) and bgm_config.get("enabled", False):
         from video_processing.bgm_mixer import BGMConfig, mix_bgm_with_main
 
-        bgm_cfg = BGMConfig.from_config_dict(bgm_path, bgm_config)
+        _bgm_cfg_dict = dict(bgm_config or {})
+        if bgm_start_offset and not _bgm_cfg_dict.get("audio_offset"):
+            _bgm_cfg_dict["audio_offset"] = round(float(bgm_start_offset), 3)
+        bgm_cfg = BGMConfig.from_config_dict(bgm_path, _bgm_cfg_dict)
 
         try:
             # 这里 main_audio 就是 output_path，先有主音频再混 BGM

apps/worker/video_processing/unified_render_service.py

@@ -171,6 +171,87 @@ class UnifiedRenderService:
         self._speed_engine = SpeedEngine()
         self._asr_timeline_cache: Any = None  # ASR 字幕结果缓存，避免重复调用
         self._asr_timeline_cached = False
+        # #1970 PR2：片段级微变换计划缓存（懒构建，dedup_enabled=False 时为 None）
+        self._micro_plan_cache: Any = None
+        self._micro_plan_loaded = False
+
+    # ── #1970 PR2 智能降重：片段级微变换 ───────────────────────────────────
+    def _dedup_enabled(self) -> bool:
+        """读取 plan.config.dedup_enabled，缺省视为 True（向后兼容）。"""
+        cfg = self.plan.config or {}
+        return bool(cfg.get("dedup_enabled", True))
+
+    def _get_micro_transform_plan(self, clip_count: int) -> Any:
+        """按 task_id+视频序号构建可复现的片段级微变换计划。
+
+        种子 hash(generation_task_id + video_index)%10000，同一任务重渲结果一致。
+        dedup_enabled=False 时返回 None，调用方不注入任何微变换。
+        P1 字幕检测：无可靠的片段文字轨道信息，hflip 一律关闭（宁可不翻转）。
+        """
+        if self._micro_plan_loaded:
+            return self._micro_plan_cache
+        self._micro_plan_loaded = True
+        if not self._dedup_enabled() or clip_count <= 0:
+            self._micro_plan_cache = None
+            return None
+        try:
+            from video_processing.micro_transform_pure import build_micro_transform_plan
+
+            cfg = self.plan.config or {}
+            task_id = str(cfg.get("generation_task_id", "") or "")
+            video_index = int(cfg.get("video_index", 0) or 0)
+            self._micro_plan_cache = build_micro_transform_plan(
+                task_id,
+                video_index,
+                clip_count,
+                clip_has_text=None,  # P1 保守策略：全部按有文字处理，不翻转
+                enable_bgm_offset=bool(cfg.get("bgm")),
+            )
+        except Exception as e:
+            logger.warning("[unified-render] 微变换计划构建失败，本次不注入: %s", e)
+            self._micro_plan_cache = None
+        return self._micro_plan_cache
+
+    @staticmethod
+    def _apply_micro_transform_video(filters: list[str], mt: Any) -> None:
+        """把片段视频微变换就地追加到 filter 链（post-scale 阶段调用）。
+
+        顺序：hflip 在 pre-scale 阶段由 _apply_micro_hflip 处理，这里只加
+        eq 亮度/对比度/饱和度。速度 setpts 与既有 clip speed 相乘（见调用点），
+        避免出现两条 setpts 互相覆盖。
+        """
+        if mt is None:
+            return
+        if abs(mt.brightness) > 1e-4 or abs(mt.contrast - 1.0) > 1e-4 or abs(mt.saturation - 1.0) > 1e-4:
+            filters.append(
+                f"eq=brightness={mt.brightness:+.4f}:" f"contrast={mt.contrast:.4f}:saturation={mt.saturation:.4f}"
+            )
+
+    @staticmethod
+    def _apply_micro_hflip(filters: list[str], mt: Any) -> None:
+        """片段级水平翻转（pre-scale 阶段）。P1 有文字/无法判定时 mt.hflip=False。"""
+        if mt is not None and mt.hflip and not mt.has_text:
+            filters.append("hflip")
+
+    @staticmethod
+    def _micro_speed_factor(mt: Any) -> float:
+        """片段微变换速度因子（0.97~1.03），无计划返回 1.0。"""
+        if mt is None:
+            return 1.0
+        return float(getattr(mt, "speed", 1.0) or 1.0)
+
+    def _get_micro_bgm_offset(self) -> float:
+        """#1970 PR2：读取本视频 BGM 起始偏移（秒），无 BGM/禁用时为 0。"""
+        if not self.plan.config:
+            return 0.0
+        try:
+            count = len([c for c in (self.plan.clips or []) if getattr(c, "clip_type", "main") != "audio"])
+            plan = self._get_micro_transform_plan(count)
+            if plan:
+                return round(float(plan.bgm_start_offset or 0.0), 3)
+        except Exception:
+            logger.debug("微变换 BGM 偏移读取失败，按 0 处理: plan_id=%s", getattr(self.plan, "id", "?"))
+        return 0.0
 
     def render(self) -> RenderResult:
         """执行渲染，返回 RenderResult.
@@ -316,6 +397,9 @@ class UnifiedRenderService:
                     ctx = RenderContext(work_dir=self.work_dir, plan_id=self.plan.id)
                     from video_processing.bgm_mixer import BGMConfig, mix_bgm_with_main
 
+                    _bgm_off = self._get_micro_bgm_offset()
+                    if _bgm_off and not (bgm_config or {}).get("audio_offset"):
+                        bgm_config = {**bgm_config, "audio_offset": _bgm_off}
                     bgm_cfg = BGMConfig.from_config_dict(self.bgm_path, bgm_config)
                     # 从直通输出中提取音频
                     main_audio_path = self.work_dir / f"pass_through_audio_{self.plan.id}.aac"
@@ -365,6 +449,7 @@ class UnifiedRenderService:
                 bgm_path=self.bgm_path,
                 bgm_config=bgm_config,
                 audio_tracks_config=audio_tracks_config,
+                bgm_start_offset=self._get_micro_bgm_offset(),
             )
             t_audio_end = time.time()
             audio_mix_ms = int((t_audio_end - t_audio_start) * 1000)
@@ -1112,6 +1197,28 @@ class UnifiedRenderService:
         if ass_path is not None:
             return False, "有字幕叠加"
 
+        # #1970 PR2：片段级微变换（变速/hflip/亮度/对比度/饱和度）需要重编码
+        try:
+            _video_sources = [c for c in (self.clips or []) if getattr(c, "clip_type", "main") != "audio"]
+            _ordinal = -1
+            for _i, _c in enumerate(_video_sources):
+                if getattr(_c, "id", None) == getattr(clip, "clip_id", None):
+                    _ordinal = _i
+                    break
+            _mt_plan = self._get_micro_transform_plan(len(_video_sources))
+            if _mt_plan and 0 <= _ordinal < len(_mt_plan.clips):
+                _mt = _mt_plan.clips[_ordinal]
+                if (
+                    abs(UnifiedRenderService._micro_speed_factor(_mt) - 1.0) >= 1e-6
+                    or (_mt.hflip and not _mt.has_text)
+                    or abs(_mt.brightness) > 1e-4
+                    or abs(_mt.contrast - 1.0) > 1e-4
+                    or abs(_mt.saturation - 1.0) > 1e-4
+                ):
+                    return False, "启用了片段级微变换"
+        except Exception:
+            logger.debug("stream copy 微变换门控检查异常，按可 copy 处理", exc_info=True)
+
         # 有调速 → 需要重编码 → 不能 copy
         speed = UnifiedRenderService._clip_speed(clip)
         if abs(speed - 1.0) >= 1e-6:
@@ -1318,11 +1425,16 @@ class UnifiedRenderService:
 
         # 视觉扰动（plan 级别，直通模式同样适用）
         vp = self._get_visual_perturbation()
+        # #1970 PR2：单片段直通；计划按源视频片段数构建，序号取 config._micro_index
+        _src_video_count = len([c for c in (self.clips or []) if getattr(c, "clip_type", "main") != "audio"])
+        mt_plan = self._get_micro_transform_plan(max(1, _src_video_count))
+        _mi = int(clip.config.get("_micro_index", 0)) if isinstance(clip.config, dict) else 0
+        mt = mt_plan.clips[_mi] if mt_plan and 0 <= _mi < len(mt_plan.clips) else None
 
-        # 调速 — 与 filter_complex 路径一致（叠加视觉扰动 speed_factor）
+        # 调速 — 与 filter_complex 路径一致（叠加视觉扰动 speed_factor 与 #1970 微变换速度）
         speed = UnifiedRenderService._clip_speed(clip)
         vp_speed = vp.get("speed_factor", 1.0) if vp else 1.0
-        effective_speed = speed * vp_speed
+        effective_speed = speed * vp_speed  # 微变换速度已烘焙进 playback_speed
         if abs(effective_speed - 1.0) >= 1e-6:
             filters.append(f"setpts=PTS/{effective_speed:.4f}")
 
@@ -1336,6 +1448,8 @@ class UnifiedRenderService:
         # 视觉扰动：hflip（在 scale 之前）
         if vp:
             self._apply_visual_perturbation_pre_scale(filters, vp)
+        # #1970 PR2：片段级 hflip（P1 保守：有文字/无法判定时不翻转）
+        UnifiedRenderService._apply_micro_hflip(filters, mt)
 
         # scale + pad（等比缩放+留黑边）
         if role in ("overlay", "corner_voice"):
@@ -1354,6 +1468,8 @@ class UnifiedRenderService:
         # 视觉扰动：zoom + brightness（在 scale+pad 之后、调色之前）
         if vp:
             self._apply_visual_perturbation_post_scale(filters, vp)
+        # #1970 PR2：片段级亮度/对比度/饱和度微调
+        UnifiedRenderService._apply_micro_transform_video(filters, mt)
 
         # 调色滤镜
         color_grade = ColorGradeConfig.from_dict(clip.config.get("color_grade"))
@@ -1450,7 +1566,8 @@ class UnifiedRenderService:
             # 音频调速（在降噪之后、音量之前，与 render_audio.py concat 路径保持一致）
             # SpeedEngine.build_audio_filter 内部已实现多级 atempo 串联，
             # 自动处理超出 [0.5, 2.0] 范围的速度（如 0.25x → atempo=0.5,atempo=0.5）。
-            speed = UnifiedRenderService._clip_speed(clip)
+            # #1970 PR2：叠加片段微变换速度因子，保持音画同步。
+            speed = UnifiedRenderService._clip_speed(clip)  # 微变换速度已烘焙进 playback_speed
             if abs(speed - 1.0) >= 1e-6:
                 try:
                     from video_processing.speed_engine import SpeedConfig, SpeedEngine
@@ -1522,11 +1639,20 @@ class UnifiedRenderService:
         支持多段裁剪：一个 clip 配置了 trim_segments 时会展开为多个 ResolvedClip。
         """
         resolved: list[ResolvedClip] = []
+        # #1970 PR2：预建片段级微变换计划，按源视频片段序号取速度因子，
+        # 烘焙进 playback_speed，保证视频 setpts 与音频 atempo 一致。
+        video_source_clips = [c for c in self.clips if getattr(c, "clip_type", "main") != "audio"]
+        mt_plan = self._get_micro_transform_plan(len(video_source_clips))
+        _video_ordinal = {id(c): i for i, c in enumerate(video_source_clips)}
+
         for clip in self.clips:
             asset_id = clip.asset_id
             if not asset_id:
                 logger.warning("片段无素材: clip_id=%s", clip.id)
                 continue
+            _mt_idx = _video_ordinal.get(id(clip), -1)
+            _mt = mt_plan.clips[_mt_idx] if mt_plan and 0 <= _mt_idx < len(mt_plan.clips) else None
+            _micro_speed = UnifiedRenderService._micro_speed_factor(_mt)
 
             local_path = self.asset_path_map.get(asset_id)
             if local_path is None or not local_path.exists():
@@ -1555,7 +1681,7 @@ class UnifiedRenderService:
                     seg_duration = seg.trim.duration
 
                     # 多段裁剪：如果段的时长超过素材实际时长，减速补偿
-                    seg_speed = configured_speed
+                    seg_speed = configured_speed * _micro_speed
                     if actual_duration > 0 and seg_duration > actual_duration + 0.05:
                         seg_speed = max(0.25, round(configured_speed * actual_duration / seg_duration, 4))
                         logger.info(
@@ -1578,7 +1704,7 @@ class UnifiedRenderService:
                         transition_effect=clip.transition_effect or "cut",
                         transition_duration=getattr(clip, "transition_duration", 0.0) or 0.0,
                         playback_speed=seg_speed,
-                        config={**clip_config, "_segment_id": seg.segment_id},
+                        config={**clip_config, "_segment_id": seg.segment_id, "_micro_index": _mt_idx},
                         actual_duration=actual_duration,
                         trim_config=seg.trim,
                     )
@@ -1633,12 +1759,13 @@ class UnifiedRenderService:
                     avail_in_asset,
                     freeze_seconds,
                 )
-            final_speed = configured_speed
+            final_speed = configured_speed * _micro_speed
 
             # freeze 标记写入 config，供视频 tpad / 音频 apad 读取
             resolved_config = dict(clip_config)
             if freeze_seconds > 0:
                 resolved_config["_freeze_seconds"] = freeze_seconds
+            resolved_config["_micro_index"] = _mt_idx
 
             rc = ResolvedClip(
                 clip_id=clip.id,
@@ -1754,9 +1881,15 @@ class UnifiedRenderService:
         preprocessed_labels: list[str] = []
         # 视觉扰动（plan 级别，所有 clip 共享同一套扰动参数）
         vp = self._get_visual_perturbation()
+        # #1970 PR2：片段级微变换（每片段独立参数，dedup_enabled=False 时为 None）
+        # 计划按源视频片段数构建，trim 多段展开时各段通过 config._micro_index 找参数
+        _src_video_count = len([c for c in (self.clips or []) if getattr(c, "clip_type", "main") != "audio"])
+        mt_plan = self._get_micro_transform_plan(_src_video_count)
         for i, clip in enumerate(all_clips):
             label = f"v{i}"
             role = _resolve_layer_role(clip.clip_type, clip.config)
+            _mi = int(clip.config.get("_micro_index", i)) if isinstance(clip.config, dict) else i
+            mt = mt_plan.clips[_mi] if mt_plan and 0 <= _mi < len(mt_plan.clips) else None
 
             filters: list[str] = []
 
@@ -1774,10 +1907,10 @@ class UnifiedRenderService:
                     filters.append(f"trim=duration={trim_dur:.3f}")
                 filters.append("setpts=PTS-STARTPTS")
 
-            # 调速 — 基于 setpts 改变播放速度（叠加视觉扰动 speed_factor）
+            # 调速 — 基于 setpts 改变播放速度（叠加视觉扰动 speed_factor 与 #1970 微变换速度）
             speed = UnifiedRenderService._clip_speed(clip)
             vp_speed = vp.get("speed_factor", 1.0) if vp else 1.0
-            effective_speed = speed * vp_speed
+            effective_speed = speed * vp_speed  # 微变换速度已烘焙进 playback_speed
             if abs(effective_speed - 1.0) >= 1e-6:
                 filters.append(f"setpts=PTS/{effective_speed:.4f}")
 
@@ -1791,6 +1924,8 @@ class UnifiedRenderService:
             # 视觉扰动：hflip（在 scale 之前，翻转原始画面）
             if vp:
                 self._apply_visual_perturbation_pre_scale(filters, vp)
+            # #1970 PR2：片段级 hflip（P1 保守：有文字/无法判定时不翻转）
+            UnifiedRenderService._apply_micro_hflip(filters, mt)
 
             # scale
             if role in ("overlay", "corner_voice"):
@@ -1809,6 +1944,8 @@ class UnifiedRenderService:
             # 视觉扰动：zoom + brightness（在 scale+pad 之后、调色之前）
             if vp:
                 self._apply_visual_perturbation_post_scale(filters, vp)
+            # #1970 PR2：片段级亮度/对比度/饱和度微调
+            UnifiedRenderService._apply_micro_transform_video(filters, mt)
 
             # 调色滤镜（每个 clip 独立的 color grade 配置）
             color_grade = ColorGradeConfig.from_dict(clip.config.get("color_grade"))

apps/worker/worker_app/tasks/generation.py

@@ -890,25 +890,51 @@ def generate_video(self, task_id: str) -> dict:
                         _flush_logs(task_id, gen_task)
                     _update_task_progress(task_id, 80, "渲染完成")
 
-                    # ── 3.5 随机边缘裁剪降重（#1664） ──────────────────────────
-                    from video_processing.ffmpeg_utils import random_edge_crop
-
+                    # ── 3.5 随机边缘裁剪降重（#1664；#1970 dedup_enabled=False 时跳过） ──
+                    _dedup_enabled = True
                     try:
-                        cropped_path = random_edge_crop(output_path)
-                        if cropped_path != output_path:
-                            output_path = cropped_path
-                        if gen_task and render_attempt == 0:
-                            gen_task.append_log("边缘裁剪", "已应用随机 2-5% 边缘裁剪降重")
-                            _flush_logs(task_id, gen_task)
-                        logger.info("[task_id=%s] 随机边缘裁剪完成: %s", task_id, output_path)
-                    except Exception as crop_err:
+                        from packages.adapters.sqlalchemy_impl.models import EditPlanModel
+
+                        with SessionLocal() as _dedup_db:
+                            _plan_row = (
+                                _dedup_db.query(EditPlanModel.config)
+                                .filter(EditPlanModel.id == current_plan_id)
+                                .first()
+                            )
+                        if _plan_row is not None:
+                            _cfg = _plan_row[0] if isinstance(_plan_row[0], dict) else {}
+                            _dedup_enabled = bool(_cfg.get("dedup_enabled", True))
+                    except Exception:
                         logger.warning(
-                            "[task_id=%s] 随机边缘裁剪失败，使用原始视频继续: %s",
+                            "[task_id=%s] 读取 plan dedup_enabled 失败，按开启处理",
                             task_id,
-                            crop_err,
                             exc_info=True,
                         )
 
+                    if not _dedup_enabled:
+                        logger.info("[task_id=%s] dedup_enabled=False，跳过边缘裁剪与微变换", task_id)
+                        if gen_task and render_attempt == 0:
+                            gen_task.append_log("降重", "已关闭边缘裁剪与微变换（确定性渲染）")
+                            _flush_logs(task_id, gen_task)
+                    else:
+                        from video_processing.ffmpeg_utils import random_edge_crop
+
+                        try:
+                            cropped_path = random_edge_crop(output_path)
+                            if cropped_path != output_path:
+                                output_path = cropped_path
+                            if gen_task and render_attempt == 0:
+                                gen_task.append_log("边缘裁剪", "已应用随机 2-5% 边缘裁剪降重")
+                                _flush_logs(task_id, gen_task)
+                            logger.info("[task_id=%s] 随机边缘裁剪完成: %s", task_id, output_path)
+                        except Exception as crop_err:
+                            logger.warning(
+                                "[task_id=%s] 随机边缘裁剪失败，使用原始视频继续: %s",
+                                task_id,
+                                crop_err,
+                                exc_info=True,
+                            )
+
                     # ── 4. 上传 OSS（不落库） ───────────────────────────────
                     _update_task_progress(task_id, 85, "开始上传")
                     file_url, _storage_key = _upload_rendered_video(

tests/unit/test_1970_micro_transform_pure.py

@@ -0,0 +1,178 @@
+"""#1970 PR2 微变换纯逻辑单元测试。
+
+覆盖：
+- 种子可复现（同 task_id+video_index 跨调用一致；不同 video_index 不同）
+- 6 维参数取值范围（speed 0.97~1.03、色彩 ±0.02、hflip 概率与字幕门控）
+- BGM 偏移 2~8s 与 atrim 片段边界
+- filter 片段格式
+"""
+
+from __future__ import annotations
+
+import random
+
+import pytest
+from video_processing.micro_transform_pure import (
+    BGM_OFFSET_MAX,
+    BGM_OFFSET_MIN,
+    COLOR_DELTA,
+    HFLIP_PROBABILITY,
+    SPEED_MAX,
+    SPEED_MIN,
+    build_bgm_offset_trim,
+    build_micro_transform_plan,
+    make_video_seed,
+)
+
+
+class TestSeed:
+    def test_seed_in_range(self):
+        for i in range(50):
+            s = make_video_seed("task-xyz", i)
+            assert 0 <= s < 10000
+
+    def test_seed_deterministic_across_calls(self):
+        a = make_video_seed("task-1", 2)
+        b = make_video_seed("task-1", 2)
+        assert a == b
+
+    def test_seed_differs_by_task_or_index(self):
+        base = make_video_seed("task-1", 0)
+        assert make_video_seed("task-2", 0) != base or make_video_seed("task-1", 1) != base
+        # 至少 video_index 不同时种子不同（概率上必然，用多组确认）
+        seeds = {make_video_seed("task-fixed", i) for i in range(8)}
+        assert len(seeds) > 1
+
+    def test_empty_task_id_does_not_raise(self):
+        assert 0 <= make_video_seed("", 0) < 10000
+
+
+class TestBuildPlan:
+    def test_zero_clips_plan_has_bgm_offset(self):
+        plan = build_micro_transform_plan("t1", 0, 0)
+        assert plan.clips == []
+        assert BGM_OFFSET_MIN <= plan.bgm_start_offset <= BGM_OFFSET_MAX
+
+    def test_clip_param_ranges(self):
+        plan = build_micro_transform_plan("t-range", 0, 30)
+        assert len(plan.clips) == 30
+        for c in plan.clips:
+            assert SPEED_MIN <= c.speed <= SPEED_MAX
+            assert -COLOR_DELTA - 1e-9 <= c.brightness <= COLOR_DELTA + 1e-9
+            assert 1.0 - COLOR_DELTA - 1e-9 <= c.contrast <= 1.0 + COLOR_DELTA + 1e-9
+            assert 1.0 - COLOR_DELTA - 1e-9 <= c.saturation <= 1.0 + COLOR_DELTA + 1e-9
+
+    def test_plan_reproducible(self):
+        p1 = build_micro_transform_plan("repro", 1, 10)
+        p2 = build_micro_transform_plan("repro", 1, 10)
+        assert [c.speed for c in p1.clips] == [c.speed for c in p2.clips]
+        assert [c.brightness for c in p1.clips] == [c.brightness for c in p2.clips]
+        assert p1.bgm_start_offset == p2.bgm_start_offset
+
+    def test_hflip_disabled_when_no_text_info(self):
+        # clip_has_text=None（P1 保守）：全部按有文字处理，一律不翻转
+        plan = build_micro_transform_plan("t1", 0, 40, clip_has_text=None)
+        assert all(not c.hflip for c in plan.clips)
+        assert all(c.has_text for c in plan.clips)
+
+    def test_hflip_never_on_text_clips(self):
+        # 全部标记有文字：无论如何都不翻转
+        plan = build_micro_transform_plan("t-text", 0, 40, clip_has_text=[True] * 40)
+        assert all(not c.hflip for c in plan.clips)
+
+    def test_hflip_roughly_half_on_clean_clips(self):
+        # 全部无文字：翻转比例应接近 50%（给宽松区间防 flaky）
+        plan = build_micro_transform_plan("t-clean", 0, 2000, clip_has_text=[False] * 2000)
+        flipped = sum(1 for c in plan.clips if c.hflip)
+        ratio = flipped / 2000
+        assert HFLIP_PROBABILITY == 0.5
+        assert 0.40 < ratio < 0.60
+
+    def test_hflip_mixed_text_mask(self):
+        mask = [i % 2 == 0 for i in range(100)]  # 偶数位有文字
+        plan = build_micro_transform_plan("t-mask", 0, 100, clip_has_text=mask)
+        for c in plan.clips:
+            if mask[c.clip_index]:
+                assert not c.hflip
+
+    def test_bgm_offset_disabled(self):
+        plan = build_micro_transform_plan("t1", 0, 5, enable_bgm_offset=False)
+        assert plan.bgm_start_offset == 0.0
+
+    def test_clip_lookup(self):
+        plan = build_micro_transform_plan("t1", 0, 3)
+        assert plan.clip(0) is plan.clips[0]
+        assert plan.clip(2) is plan.clips[2]
+        assert plan.clip(99) is None
+
+
+class TestFilterSuffix:
+    def test_identity_transform_empty_suffix(self):
+        plan = build_micro_transform_plan("t", 0, 1, clip_has_text=[True])
+        c = plan.clips[0]
+        # 强制为恒等参数验证格式
+        object.__setattr__(c, "speed", 1.0)
+        object.__setattr__(c, "brightness", 0.0)
+        object.__setattr__(c, "contrast", 1.0)
+        object.__setattr__(c, "saturation", 1.0)
+        object.__setattr__(c, "hflip", False)
+        assert c.video_filter_suffix() == ""
+        assert c.audio_filter_suffix() == ""
+
+    def test_video_filter_order_speed_hflip_eq(self):
+        plan = build_micro_transform_plan("t", 0, 1, clip_has_text=[False])
+        c = plan.clips[0]
+        object.__setattr__(c, "speed", 1.02)
+        object.__setattr__(c, "hflip", True)
+        object.__setattr__(c, "has_text", False)
+        object.__setattr__(c, "brightness", 0.01)
+        suffix = c.video_filter_suffix()
+        steps = suffix.split(",")
+        assert steps[0].startswith("setpts=")
+        assert steps[1] == "hflip"
+        assert steps[2].startswith("eq=brightness=")
+
+    def test_hflip_blocked_by_text_in_suffix(self):
+        plan = build_micro_transform_plan("t", 0, 1)
+        c = plan.clips[0]
+        object.__setattr__(c, "hflip", True)
+        object.__setattr__(c, "has_text", True)
+        assert "hflip" not in c.video_filter_suffix()
+
+    def test_audio_suffix_only_for_speed(self):
+        plan = build_micro_transform_plan("t", 0, 1)
+        c = plan.clips[0]
+        object.__setattr__(c, "speed", 0.98)
+        assert c.audio_filter_suffix() == "atempo=0.98000"
+        object.__setattr__(c, "speed", 1.0)
+        assert c.audio_filter_suffix() == ""
+
+
+class TestBgmTrim:
+    def test_normal_offset(self):
+        assert build_bgm_offset_trim(3.0, 30.0) == "atrim=start=3.000,"
+
+    def test_zero_or_negative(self):
+        assert build_bgm_offset_trim(0.0, 30.0) == ""
+        assert build_bgm_offset_trim(-1.0, 30.0) == ""
+
+    def test_offset_near_end_falls_back(self):
+        # 距尾部不足 0.5s → 空串
+        assert build_bgm_offset_trim(29.7, 30.0) == ""
+
+    def test_invalid_duration(self):
+        assert build_bgm_offset_trim(3.0, 0.0) == ""
+
+
+class TestDistributionSanity:
+    def test_speed_distribution_spans_range(self):
+        # 多片段采样确认速度在全区间有分布（非常量）
+        plan = build_micro_transform_plan("t-dist", 0, 500)
+        speeds = [c.speed for c in plan.clips]
+        assert min(speeds) < 0.99
+        assert max(speeds) > 1.01
+
+    def test_bgm_offset_range_many_seeds(self):
+        for i in range(100):
+            plan = build_micro_transform_plan("t", i, 1)
+            assert BGM_OFFSET_MIN <= plan.bgm_start_offset <= BGM_OFFSET_MAX

tests/unit/test_1970_micro_transform_render.py

@@ -0,0 +1,196 @@
+"""#1970 PR2 渲染服务微变换注入测试。
+
+不做真实渲染，只验证 UnifiedRenderService 上微变换计划的开关、缓存、
+滤镜注入与速度因子；纯参数生成在 test_1970_micro_transform_pure 覆盖。
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+
+def _make_service(plan_config: dict | None = None, clips=None):
+    from video_processing.unified_render_service import UnifiedRenderService
+
+    svc = object.__new__(UnifiedRenderService)
+    svc.plan = MagicMock()
+    svc.plan.config = plan_config or {}
+    svc.plan.id = "plan-1"
+    svc.plan.clips = clips or []
+    svc._micro_plan_cache = None
+    svc._micro_plan_loaded = False
+    return svc
+
+
+class TestDedupGate:
+    def test_default_enabled_when_config_missing(self):
+        svc = _make_service({})
+        assert svc._dedup_enabled() is True
+
+    def test_explicit_true(self):
+        svc = _make_service({"dedup_enabled": True})
+        assert svc._dedup_enabled() is True
+
+    def test_explicit_false(self):
+        svc = _make_service({"dedup_enabled": False})
+        assert svc._dedup_enabled() is False
+
+    def test_plan_none_config_treated_enabled(self):
+        svc = _make_service(None)
+        svc.plan.config = None
+        assert svc._dedup_enabled() is True
+
+
+class TestPlanBuild:
+    def test_disabled_returns_none_and_cached(self):
+        svc = _make_service({"dedup_enabled": False, "generation_task_id": "t1"})
+        assert svc._get_micro_transform_plan(5) is None
+        # 第二次走缓存
+        svc._dedup_enabled = MagicMock(side_effect=AssertionError("不应再次计算"))
+        assert svc._get_micro_transform_plan(5) is None
+
+    def test_zero_clips_returns_none(self):
+        svc = _make_service({"generation_task_id": "t1"})
+        assert svc._get_micro_transform_plan(0) is None
+
+    def test_enabled_builds_reproducible_plan(self):
+        cfg = {"generation_task_id": "task-abc", "video_index": 2, "bgm": {"enabled": True}}
+        svc1 = _make_service(cfg)
+        svc2 = _make_service(dict(cfg))
+        p1 = svc1._get_micro_transform_plan(6)
+        p2 = svc2._get_micro_transform_plan(6)
+        assert p1 is not None and p2 is not None
+        assert [c.speed for c in p1.clips] == [c.speed for c in p2.clips]
+        assert p1.seed == p2.seed
+        assert len(p1.clips) == 6
+
+    def test_p1_conservative_no_hflip(self):
+        svc = _make_service({"generation_task_id": "t1"})
+        plan = svc._get_micro_transform_plan(30)
+        assert all(not c.hflip for c in plan.clips)
+
+    def test_no_bgm_config_zero_offset(self):
+        svc = _make_service({"generation_task_id": "t1"})
+        plan = svc._get_micro_transform_plan(3)
+        assert plan.bgm_start_offset == 0.0
+
+    def test_bgm_enabled_offset_in_range(self):
+        svc = _make_service({"generation_task_id": "t1", "bgm": {"enabled": True}})
+        plan = svc._get_micro_transform_plan(3)
+        assert 2.0 <= plan.bgm_start_offset <= 8.0
+
+
+class TestFilterInjection:
+    def test_none_mt_noop(self):
+        svc = _make_service({})
+        from video_processing.unified_render_service import UnifiedRenderService
+
+        filters = ["scale=100:100"]
+        UnifiedRenderService._apply_micro_transform_video(filters, None)
+        UnifiedRenderService._apply_micro_hflip(filters, None)
+        assert filters == ["scale=100:100"]
+
+    def test_eq_injection(self):
+        svc = _make_service({"generation_task_id": "t1"})
+        plan = svc._get_micro_transform_plan(1)
+        mt = plan.clips[0]
+        from video_processing.unified_render_service import UnifiedRenderService
+
+        filters: list[str] = []
+        UnifiedRenderService._apply_micro_transform_video(filters, mt)
+        assert filters and filters[0].startswith("eq=brightness=")
+        assert "contrast=" in filters[0] and "saturation=" in filters[0]
+
+    def test_hflip_skipped_p1(self):
+        svc = _make_service({"generation_task_id": "t1"})
+        plan = svc._get_micro_transform_plan(10)
+        from video_processing.unified_render_service import UnifiedRenderService
+
+        for mt in plan.clips:
+            filters: list[str] = []
+            UnifiedRenderService._apply_micro_hflip(filters, mt)
+            assert filters == []
+
+    def test_speed_factor(self):
+        from video_processing.micro_transform_pure import ClipMicroTransform
+        from video_processing.unified_render_service import UnifiedRenderService
+
+        assert UnifiedRenderService._micro_speed_factor(None) == 1.0
+        assert UnifiedRenderService._micro_speed_factor(ClipMicroTransform(0, speed=1.025)) == pytest.approx(1.025)
+        assert UnifiedRenderService._micro_speed_factor(MagicMock(speed=0.97)) == pytest.approx(0.97)
+
+    def test_bgm_offset_reader_respects_flag(self):
+        svc_off = _make_service({"dedup_enabled": False})
+        assert svc_off._get_micro_bgm_offset() == 0.0
+
+        svc_on = _make_service({"generation_task_id": "t1", "bgm": {"enabled": True}}, clips=[MagicMock()])
+        off = svc_on._get_micro_bgm_offset()
+        assert 2.0 <= off <= 8.0
+
+    def test_bgm_offset_zero_without_bgm(self):
+        svc = _make_service({"generation_task_id": "t1"}, clips=[MagicMock()])
+        assert svc._get_micro_bgm_offset() == 0.0
+
+
+class TestStreamCopyGate:
+    """dedup 开启时微变换需要重编码，stream copy 必须被拒绝。"""
+
+    def _build(self, config):
+        from types import SimpleNamespace
+        from unittest.mock import patch
+
+        from video_processing.unified_render_service import ResolvedClip, UnifiedRenderService
+
+        source = SimpleNamespace(
+            id="c1",
+            clip_type="main",
+        )
+        svc = object.__new__(UnifiedRenderService)
+        svc.output_width = 1280
+        svc.output_height = 720
+        svc.output_fps = 25
+        svc.plan = MagicMock()
+        svc.plan.id = "plan-1"
+        svc.plan.config = config
+        svc.plan.clips = [source]
+        svc.clips = [source]
+        svc._micro_plan_cache = None
+        svc._micro_plan_loaded = False
+        resolved = ResolvedClip(
+            clip_id="c1",
+            asset_id="a1",
+            local_path=Path("/tmp/a1.mp4"),
+            clip_type="main",
+            order=0,
+        )
+        info = {
+            "width": 1280,
+            "height": 720,
+            "fps": 25.0,
+            "video_codec": "h264",
+            "pix_fmt": "yuv420p",
+            "duration": 5.0,
+            "has_audio": True,
+            "audio_codec": "aac",
+        }
+        return svc, resolved, info
+
+    def test_dedup_enabled_blocks_stream_copy(self):
+        from unittest.mock import patch
+
+        svc, resolved, info = self._build({"dedup_enabled": True, "generation_task_id": "t1"})
+        with patch("video_processing.unified_render_service.probe_video_info", return_value=info):
+            can_copy, reason = svc._can_use_stream_copy(resolved)
+        assert can_copy is False
+        assert "微变换" in reason
+
+    def test_dedup_disabled_allows_stream_copy(self):
+        from unittest.mock import patch
+
+        svc, resolved, info = self._build({"dedup_enabled": False})
+        with patch("video_processing.unified_render_service.probe_video_info", return_value=info):
+            can_copy, _ = svc._can_use_stream_copy(resolved)
+        assert can_copy is True

tests/unit/test_generation_common.py

@@ -340,3 +340,31 @@ class TestCollectPlanAtomClipIds:
         repo = MagicMock()
         repo.list_by_plan.side_effect = [[legacy, _make_atom_clip("atom-9")], []]
         assert collect_plan_atom_clip_ids("p1", repo) == ["atom-9"]
+
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# writeback_edit_plan_config：#1970 dedup_enabled / video_index
+# ═══════════════════════════════════════════════════════════════════════════════
+
+
+class TestWritebackDedupAndVideoIndex:
+    def test_writes_dedup_enabled_and_video_index(self):
+        from app.services.generation_common import writeback_edit_plan_config
+
+        plan = _make_plan_model({})
+        db = MagicMock()
+        db.query.return_value.filter.return_value.first.return_value = plan
+        writeback_edit_plan_config("p1", "t1", None, db, dedup_enabled=False, video_index=3)
+        assert plan.config["dedup_enabled"] is False
+        assert plan.config["video_index"] == 3
+        assert plan.config["generation_task_id"] == "t1"
+
+    def test_none_dedup_does_not_touch_flag(self):
+        from app.services.generation_common import writeback_edit_plan_config
+
+        plan = _make_plan_model({"dedup_enabled": True})
+        db = MagicMock()
+        db.query.return_value.filter.return_value.first.return_value = plan
+        writeback_edit_plan_config("p1", "t1", None, db)
+        assert plan.config["dedup_enabled"] is True
+        assert "video_index" not in plan.config

tests/unit/test_unified_render_service.py

@@ -53,11 +53,16 @@ class FakeClip:
 
 @dataclass
 class FakePlan:
-    """模拟 EditPlan。"""
+    """模拟 EditPlan。
+
+    注意：config 默认 dedup_enabled=False，关闭 #1970 片段级微变换，
+    让本文件既有的确定性渲染/stream copy 断言不受随机微变换影响；
+    微变换本身的行为在 test_1970_micro_transform_render.py 覆盖。
+    """
 
     id: str = "plan_001"
     name: str = "测试计划"
-    config: dict[str, Any] = field(default_factory=dict)
+    config: dict[str, Any] = field(default_factory=lambda: {"dedup_enabled": False})
 
 
 @pytest.fixture(autouse=True)