release v0.1.137: 抖音多源轮询重构+ASR降级+文案库切换+积分系统+多个修复

fix(generate): #1970 Step1 智能降重开关紧贴标题文字 (#1977 )
Co-authored-by: xiaoxia <dev@xiaoxiajianji.com> Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>
2026-09-18 19:41:47 +08:00 · 2026-09-18 10:39:59 +08:00 · 2026-09-18 07:30:43 +08:00 · 2026-09-18 05:44:59 +08:00 · 2026-09-18 03:57:07 +08:00 · 2026-09-18 00:23:19 +08:00
44 changed files with 4487 additions and 116 deletions
@@ -0,0 +1,58 @@
+"""add asset_atom_clips table
+
+Revision ID: 079_asset_atom_clips
+Revises: 078_drop_script_title_fields
+Create Date: 2026-09-17
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "079_asset_atom_clips"
+down_revision = "078_drop_script_title_fields"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "asset_atom_clips",
+        sa.Column("id", sa.String(36), primary_key=True),
+        sa.Column(
+            "asset_id",
+            sa.String(36),
+            sa.ForeignKey("assets.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("start_time", sa.Float(), nullable=False),
+        sa.Column("end_time", sa.Float(), nullable=False),
+        sa.Column("duration", sa.Float(), nullable=False),
+        sa.Column("clip_index", sa.Integer(), nullable=False),
+        sa.Column("tags", sa.JSON(), nullable=False, server_default=sa.text("'[]'")),
+        sa.Column("scene_change_at", sa.Float(), nullable=True),
+        sa.Column(
+            "is_fallback",
+            sa.Boolean(),
+            nullable=False,
+            server_default=sa.text("false"),
+        ),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("NOW()"),
+        ),
+    )
+    # 按素材查片段并按索引排序（复合索引前缀可独立用于 asset_id 过滤）
+    op.create_index(
+        "ix_asset_atom_clips_asset_index",
+        "asset_atom_clips",
+        ["asset_id", "clip_index"],
+        unique=True,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_asset_atom_clips_asset_index", table_name="asset_atom_clips")
+    op.drop_table("asset_atom_clips")
@@ -0,0 +1,37 @@
+"""add edit_plan_clips.atom_clip_id for #1970
+
+Revision ID: 080_edit_plan_clips_atom_clip_id
+Revises: 079_asset_atom_clips
+Create Date: 2026-09-17
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "080_edit_plan_clips_atom_clip_id"
+down_revision = "079_asset_atom_clips"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "edit_plan_clips",
+        sa.Column(
+            "atom_clip_id",
+            sa.String(36),
+            nullable=False,
+            server_default=sa.text("''"),
+        ),
+    )
+    op.create_index(
+        "ix_edit_plan_clips_atom_clip_id",
+        "edit_plan_clips",
+        ["atom_clip_id"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_edit_plan_clips_atom_clip_id", table_name="edit_plan_clips")
+    op.drop_column("edit_plan_clips", "atom_clip_id")
@@ -16,10 +16,12 @@ from app.core.task_enqueue import (
 from app.dependencies import (
    get_asset_library_repository,
    get_asset_repository,
+    get_cosyvoice_service,
    get_db_session,
    get_generated_video_repository,
    get_generation_task_repository,
    get_project_repository,
+    get_voice_clone_profile_repository,
 )
 from app.schemas.generated_video import (
    GeneratedVideoResponse,
@@ -132,6 +134,8 @@ def _select_assets_from_library(
    mode: str,
    count: int,
    rng=None,
+    script_tags: list | None = None,
+    tag_names_by_id: dict | None = None,
 ) -> list[str]:
    """根据选取模式从素材库中选取 ready 状态的视频素材 ID。

@@ -141,6 +145,8 @@ def _select_assets_from_library(
        count: 选取数量，0 表示全部（仅 smart 模式有效）
        rng: 可选随机源（smart 模式排序噪声用），生产环境不传则内部随机；
            测试可注入固定种子或零噪声随机源获得确定性结果。
+        script_tags: #1970 叙事模式文案标签；非空时标签命中素材优先，不足再用其余素材兜底。
+        tag_names_by_id: asset_id → 素材标签名列表（素材只存 tag_ids 时由调用方查名称注入）。

    Returns:
        选中的素材 ID 列表
@@ -150,6 +156,20 @@ def _select_assets_from_library(
    if not ready_video_assets:
        return []

+    # 叙事模式（#1970 PR3）：文案标签命中池优先；无任何命中时完全降级为现有随机逻辑。
+    if script_tags:
+        from packages.domain.narrative_match import pick_narrative_assets
+
+        limit = count if count > 0 else None
+        picked = pick_narrative_assets(
+            ready_video_assets,
+            script_tags=script_tags,
+            tag_names_by_id=tag_names_by_id,
+            limit=limit,
+            rng=rng,
+        )
+        return [a.id for a in picked]
+
    if mode == "smart":
        # 智能匹配：统一使用 packages/domain/smart_match.py 的多维评分+多样性选取
        # 评分维度：质量分(40%) + 时长适配(30%) + 新鲜度(20%) + 未使用加分(10%)
@@ -162,16 +182,78 @@ def _select_assets_from_library(
    return [a.id for a in ready_video_assets]


+# #1970 PR3：video_ratio → 默认输出分辨率（显式 output_width/output_height 优先）
+_VIDEO_RATIO_DIMENSIONS = {
+    "9:16": (1080, 1920),
+    "16:9": (1920, 1080),
+    "1:1": (1080, 1080),
+    "3:4": (1080, 1440),
+    "4:3": (1440, 1080),
+}
+
+
+def _resolve_output_dimensions(request: CreateGenerationTaskRequest) -> tuple[int, int]:
+    """解析输出分辨率：显式 output_width/output_height 非旧默认值时优先，否则按 video_ratio。
+
+    前端 #1973 总是同时传 video_ratio 与具体分辨率，两者一致；此函数主要服务
+    只传比例的调用方，并保证旧调用（不传比例）维持 1280x720 行为。
+    """
+    width, height = request.output_width, request.output_height
+    ratio = (request.video_ratio or "").strip()
+    if ratio in _VIDEO_RATIO_DIMENSIONS and (width, height) == (1280, 720):
+        return _VIDEO_RATIO_DIMENSIONS[ratio]
+    return width, height
+
+
+def _load_asset_tag_names(db: Session, assets: list, user_id: str) -> dict[str, list[str]]:
+    """叙事模式：查 TagModel 名称，构造 asset_id → 标签名列表（失败返回空 dict 降级随机）。"""
+    try:
+        from packages.adapters.sqlalchemy_impl.models import AssetTagModel, TagModel
+
+        tag_ids = {tid for a in assets for tid in (getattr(a, "tag_ids", None) or [])}
+        if not tag_ids:
+            return {}
+        name_rows = (
+            db.query(TagModel.id, TagModel.name).filter(TagModel.id.in_(tag_ids), TagModel.user_id == user_id).all()
+        )
+        name_by_id = {row.id: row.name for row in name_rows}
+        links = db.query(AssetTagModel.asset_id, AssetTagModel.tag_id).filter(AssetTagModel.tag_id.in_(tag_ids)).all()
+        index: dict[str, list[str]] = {}
+        for asset_id, tag_id in links:
+            name = name_by_id.get(tag_id)
+            if name:
+                index.setdefault(asset_id, []).append(name)
+        return index
+    except Exception:  # noqa: BLE001 - 标签匹配是加分项，查询失败不阻断生成
+        logger.warning("[叙事模式] 素材标签查询失败，降级随机选片", exc_info=True)
+        return {}
+
+
 def _writeback_edit_plan_config(
    plan_id: str,
    task_id: str,
    title_config: dict | None,
    db: Session,
+    dedup_enabled: bool | None = None,
+    video_index: int | None = None,
+    assembly_mode: str | None = None,
+    script_id: str | None = None,
+    video_ratio: str | 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,
+        assembly_mode=assembly_mode,
+        script_id=script_id,
+        video_ratio=video_ratio,
+    )


 def _resolve_project_and_library(
@@ -221,16 +303,63 @@ def create_generation_task(
    asset_library_repository: Any = Depends(get_asset_library_repository),
    asset_repository: Any = Depends(get_asset_repository),
    db: Session = Depends(get_db_session),
+    cosyvoice_service: Any = Depends(get_cosyvoice_service),
+    voice_clone_repository: Any = Depends(get_voice_clone_profile_repository),
 ) -> BatchGenerationTaskResponse:
    logger.info(
-        "[生成任务] 接收请求: user_id=%s, template_id=%s, asset_count=%d, mode=%s, count=%d",
+        "[生成任务] 接收请求: user_id=%s, template_id=%s, asset_count=%d, mode=%s, assembly=%s, count=%d",
        authenticated_user.user.id,
        request.template_id,
        len(request.asset_ids),
        request.asset_select_mode,
+        request.assembly_mode,
        request.count,
    )

+    # video_ratio → 默认分辨率（显式分辨率优先）
+    request.output_width, request.output_height = _resolve_output_dimensions(request)
+
+    # ── #1970 PR3 叙事模式：入队前同步合成配音并落为 audio asset ──
+    # 合成结果覆盖 voice_library_id（下游按 audio asset id 消费），失败直接 4xx 不入队。
+    narrative_script_tags: list = []
+    if request.assembly_mode == "narrative":
+        from app.config import settings as _settings
+        from app.services.narrative_service import NarrativeError, prepare_narrative_voice
+
+        from packages.adapters.sqlalchemy_impl.tts_job_repository import SQLAlchemyTTSJobRepository
+
+        try:
+            narrative_ctx = prepare_narrative_voice(
+                db=db,
+                user_id=authenticated_user.user.id,
+                script_id=request.script_id,
+                tts_voice_id=request.tts_voice_id,
+                tts_voice_source=request.tts_voice_source,
+                tts_repository=SQLAlchemyTTSJobRepository(db),
+                cosyvoice_service=cosyvoice_service,
+                voice_clone_repository=voice_clone_repository,
+                asset_repository=asset_repository,
+                asset_library_repository=asset_library_repository,
+                project_repository=project_repository,
+                storage_service=get_storage_service(),
+                points_enabled=bool(getattr(_settings, "points_enabled", False)),
+                is_member=bool(getattr(authenticated_user.user, "is_member", False)),
+                member_type=getattr(authenticated_user.user, "member_type", None),
+            )
+        except NarrativeError as e:
+            logger.warning("[叙事模式] 配音前置处理失败: %s", e.message)
+            raise HTTPException(status_code=e.status_code, detail=e.message) from e
+
+        request.voice_library_id = narrative_ctx.voice_asset_id
+        narrative_script_tags = list(getattr(narrative_ctx.script, "tags", None) or [])
+        logger.info(
+            "[叙事模式] 配音已就绪: script_id=%s, tts_job=%s, voice_asset=%s, duration=%.2f",
+            request.script_id,
+            narrative_ctx.tts_job_id,
+            narrative_ctx.voice_asset_id,
+            narrative_ctx.audio_duration,
+        )
+
    try:
        project_id, asset_library_id = _resolve_project_and_library(
            request, project_repository, asset_library_repository, asset_repository, authenticated_user
@@ -256,19 +385,29 @@ def create_generation_task(

        # 素材库自动匹配：当未显式指定 asset_ids 时，按模式自动选取
        if not resolved_asset_ids:
+            _tag_index = (
+                _load_asset_tag_names(db, assets, authenticated_user.user.id) if narrative_script_tags else None
+            )
            resolved_asset_ids = _select_assets_from_library(
                assets,
                mode=request.asset_select_mode,
                count=request.asset_select_count,
+                script_tags=narrative_script_tags or None,
+                tag_names_by_id=_tag_index,
            )
-    elif project_id and not resolved_asset_ids and request.asset_select_mode in ("smart",):
-        # 项目级模式：未指定 asset_ids 且选择了 smart 模式时，也自动选取
+    elif project_id and not resolved_asset_ids and (request.asset_select_mode in ("smart",) or narrative_script_tags):
+        # 项目级模式：未指定 asset_ids 且选择了 smart 模式（或叙事模式按标签匹配）时自动选取
        assets = asset_repository.find_by_project(project_id)
        if assets:
+            _tag_index = (
+                _load_asset_tag_names(db, assets, authenticated_user.user.id) if narrative_script_tags else None
+            )
            resolved_asset_ids = _select_assets_from_library(
                assets,
                mode=request.asset_select_mode,
                count=request.asset_select_count,
+                script_tags=narrative_script_tags or None,
+                tag_names_by_id=_tag_index,
            )
            if not resolved_asset_ids:
                raise HTTPException(
@@ -332,6 +471,10 @@ def create_generation_task(
                            task_id=preview_task.id,
                            title_config=fallback_title_config,
                            db=db,
+                            dedup_enabled=request.dedup_enabled,
+                            assembly_mode=request.assembly_mode,
+                            script_id=request.script_id or None,
+                            video_ratio=request.video_ratio or None,
                        )

                    logger.info(
@@ -476,9 +619,12 @@ def create_generation_task(
            variant_plan_ids.append(_plan0.id)

            # #1855 P0：批次区间避让表，从变体0实际clips构建初始值（公共函数）
+            from app.services.generation_common import collect_plan_atom_clip_ids as _collect_atom_ids
            from app.services.generation_common import collect_plan_segments as _collect_segments

            _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)

            # 变体 1..N-1 独立选片（传入累积batch_segments做素材区间避让）
            for task_index in range(1, count):
@@ -493,6 +639,7 @@ def create_generation_task(
                            name_suffix=f"批量{task_index + 1}",
                            voice_duration=voice_durations[task_index] if task_index < len(voice_durations) else 0.0,
                            batch_segments=_batch_segments,
+                            batch_used_atom_ids=_batch_atom_ids,
                        )
                        break
                    except ValueError as ve:
@@ -529,6 +676,8 @@ def create_generation_task(
                    _new_segs = _collect_segments(variant.id, _plan_svc._clip_repo)
                    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))
                except Exception:
                    logger.exception("[生成任务] 变体%d 区间收集失败（不阻断）", task_index)

@@ -672,6 +821,11 @@ def create_generation_task(
                        task_id=task.id,
                        title_config=variant_title_config,
                        db=db,
+                        dedup_enabled=request.dedup_enabled,
+                        video_index=task_index,
+                        assembly_mode=request.assembly_mode,
+                        script_id=request.script_id or None,
+                        video_ratio=request.video_ratio or None,
                    )

                if safe_enqueue_generation_task(
@@ -762,6 +916,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,
@@ -55,7 +55,9 @@ _DOUYIN_DEBUG_ERRORS = os.environ.get("DOUYIN_DEBUG_ERRORS", "").lower() in (
    "1",
    "true",
    "yes",
-) or os.environ.get("APP_ENV", "").lower() in ("staging", "dev", "development", "test")
+) or os.environ.get(
+    "APP_ENV", ""
+).lower() in ("staging", "dev", "development", "test")

 _TAIL_PUNCT = ".,;:!?，。；：！？)）]》" + chr(34) + chr(39) + "<>"
 _URL_EXTRACT_RE = re.compile(r"https?://\S+", re.IGNORECASE)
@@ -140,6 +142,7 @@ def _extract_and_validate_douyin_url(raw_input):

 def _mk_post_json(self, path, payload):
    import httpx
+
    if not self.is_available:
        raise MediaKitError("MediaKit API Key 未配置", code="NotConfigured")
    url = self._base_url + path
@@ -168,6 +171,7 @@ def _mk_post_json(self, path, payload):

 def _mk_get_json(self, path):
    import httpx
+
    if not self.is_available:
        raise MediaKitError("MediaKit API Key 未配置", code="NotConfigured")
    url = self._base_url + path
@@ -283,7 +287,7 @@ def _direct_url_download_and_local_asr(direct_url, page_url, temp_dir):
        raise
    except httpx.TimeoutException:
        logger.warning("直链下载超时: %s", page_url)
-        raise HTTPException(status_code=status.HTTP_504_GATEWAY_TIMEOUT, detail="视频下载超时，请稍后重试")
+        raise HTTPException(status_code=status.HTTP_504_GATEWAY_TIMEOUT, detail="视频下载超时，请稍后重试") from None
    except Exception as exc:  # noqa: BLE001
        logger.exception("直链下载失败: url=%s err=%s", page_url, exc)
        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="视频下载失败: " + str(exc)[:200]) from exc
@@ -420,7 +424,10 @@ def extract_from_douyin(
            if text:
                logger.info(
                    "抖音 MediaKit ASR 成功: source=%s text_len=%d duration=%.1f total_time=%.1fs",
-                    result.source, len(text), duration, time.time() - t0,
+                    result.source,
+                    len(text),
+                    duration,
+                    time.time() - t0,
                )
            else:
                logger.info("抖音 MediaKit ASR 返回空文本（无旁白/BGM视频）")
@@ -440,13 +447,25 @@ def extract_from_douyin(
                if text:
                    logger.info(
                        "抖音本地 ASR 成功: source=%s text_len=%d total_time=%.1fs",
-                        result.source, len(text), time.time() - t0,
+                        result.source,
+                        len(text),
+                        time.time() - t0,
                    )
                    last_err_stage = "asr"
-        except HTTPException:
-            raise
+        except HTTPException as exc:
+            # 下载超时（504）是明确的网络错误，直接抛出
+            if exc.status_code == status.HTTP_504_GATEWAY_TIMEOUT:
+                raise
+            # 本地 ASR 不可用/失败（502/503）时记录后继续走 desc 兜底，
+            # 不直接抛 502，避免 API 镜像缺 worker 模块时整条链路挂掉
+            logger.warning("本地 ASR 链路失败(status=%d): %s", exc.status_code, exc.detail)
+            text = ""
+            # 如果是下载失败（非ASR错误），保持stage为download
+            if "语音识别" in str(exc.detail) or "ASR" in str(exc.detail):
+                last_err_stage = "asr"
        except Exception as exc:  # noqa: BLE001
            logger.warning("本地 ASR 链路异常: %s", exc)
+            text = ""

    # ── Phase C：结果判定 & 兜底 ──

@@ -529,6 +548,7 @@ def ai_generate_titles(
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="文案内容不能为空")
    count = max(1, min(5, request.count))
    from app.services.ai_service import generate_smart_titles
+
    result = generate_smart_titles(description=content, style="viral", count=count)
    titles = result.get("titles", [])[:count]
    return AiGenerateTitlesResponse(titles=titles)
@@ -98,6 +98,24 @@ 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="智能降重开关，默认开启；关闭后跳过边缘裁切与微变换")
+
+    # ── 剪辑组装模式（#1970 PR3）──
+    # random（默认，完全兼容现有随机混剪）/ narrative（叙事剪辑：文案→TTS 配音→标签匹配画面）
+    assembly_mode: str = Field(default="random", description="组装模式：random=随机混剪（默认），narrative=叙事剪辑")
+    # 叙事模式必填：文案库 scripts.id（后端据此读取 content 合成 TTS）
+    script_id: str = Field(default="", description="叙事模式必填：文案库 ID")
+    # 叙事模式必填：TTS 音色 ID（preset 为 CosyVoice 音色 id；clone 为克隆档案 id）
+    tts_voice_id: str = Field(default="", description="叙事模式必填：TTS 音色 ID（系统音色或克隆档案 ID）")
+    tts_voice_source: str = Field(default="preset", description="TTS 音色来源：preset=系统预设（默认），clone=克隆音色")
+    # 视频比例：当前前端 9:16/16:9；与 output_width/output_height 并存，传了具体分辨率时以分辨率为准
+    video_ratio: str = Field(
+        default="", description="视频比例，如 9:16（默认竖屏）/16:9；与显式分辨率冲突时以分辨率为准"
+    )
+
    @model_validator(mode="after")
    def _check_variant_arrays(self) -> "CreateGenerationTaskRequest":
        """变体数组字段长度校验 + #1749 配音严格守卫。
@@ -127,6 +145,26 @@ class CreateGenerationTaskRequest(BaseModel):
            raise ValueError(f"variant_plan_ids 长度({len(self.variant_plan_ids)})必须与 count({self.count})一致")
        return self

+    @model_validator(mode="after")
+    def _check_assembly_mode(self) -> "CreateGenerationTaskRequest":
+        """#1970 组装模式与叙事模式入参校验。"""
+        if self.assembly_mode not in ("random", "narrative"):
+            raise ValueError("assembly_mode 仅支持 'random'（默认）或 'narrative'")
+        if self.tts_voice_source not in ("preset", "clone"):
+            raise ValueError("tts_voice_source 仅支持 'preset' 或 'clone'")
+        if self.video_ratio:
+            parts = self.video_ratio.split(":")
+            if len(parts) != 2 or not all(p.isdigit() and int(p) > 0 for p in parts):
+                raise ValueError("video_ratio 格式必须为 '宽:高'，如 9:16 或 16:9")
+            if self.video_ratio not in ("9:16", "16:9", "1:1", "3:4", "4:3"):
+                raise ValueError("video_ratio 仅支持 9:16 / 16:9 / 1:1 / 3:4 / 4:3")
+        if self.assembly_mode == "narrative":
+            if not self.script_id.strip():
+                raise ValueError("叙事模式（narrative）必须提供 script_id（文案库 ID）")
+            if not self.tts_voice_id.strip():
+                raise ValueError("叙事模式（narrative）必须提供 tts_voice_id（TTS 音色 ID）")
+        return self
+
    @model_validator(mode="after")
    def _check_at_least_one_mode(self) -> "CreateGenerationTaskRequest":
        has_project = bool(self.project_id.strip())
@@ -52,7 +52,7 @@ def _extract_url_from_text(text: str) -> str:
    if not text:
        return ""
    m = re.search(r"https?://\S+", text)
-    return m.group(0).rstrip("。，！？!?,，；;\"'）)】") if m else ""
+    return m.group(0).rstrip("。，！？!?,，；;\"'）)】") if m else ""  # noqa: B005


 def _canonicalize_url(url: str, timeout: int = 8) -> str:
@@ -423,6 +423,7 @@ class EditPlanService:
                    clip_type=clip.clip_type,
                    order=clip.order,
                    asset_id=clip.asset_id,
+                    atom_clip_id=clip_item.get("atom_clip_id", ""),
                    text_content=clip.text_content,
                    start_time=clip.start_time,
                    duration=clip.duration,
@@ -474,6 +475,7 @@ class EditPlanService:
        voice_duration: float = 0.0,
        rng=None,
        batch_segments: dict[str, list[tuple[float, float]]] | None = None,
+        batch_used_atom_ids: set[str] | list[str] | None = None,
    ) -> EditPlan:
        """为批量变体生成独立 plan：完整重跑单视频选片流程（#1743）。

@@ -608,18 +610,69 @@ class EditPlanService:
                    st = float(c.start_time or 0.0)
                    batch_segments_resolved.setdefault(c.asset_id, []).append((st, st + float(c.duration)))

-        clips_data = reselect_clips_for_variant(
-            source_clips_data,
-            pool_ids,
-            asset_durations=durations,
-            asset_scene_points=scene_points,
-            historical_used_segments=historical,
-            batch_segments=batch_segments_resolved,
-            target_durations=target_durations,
-            rng=rng,
-        )
+        clips_data = None
+        # #1970 原子片段级变体重选：候选素材已切片时优先按原子片段选片
+        try:
+            from packages.adapters.sqlalchemy_impl.asset_atom_clip_repository import (
+                SQLAlchemyAssetAtomClipRepository,
+            )
+            from packages.domain.atom_clip_resolver import flatten_candidates, load_atom_clips_for_assets
+            from packages.domain.atom_clip_selector import reselect_clips_from_atoms

-        # 片段区间写回素材 metadata（与落库同事务；replace_all_clips_transactional 内 commit）
+            atom_repo = SQLAlchemyAssetAtomClipRepository(db)
+
+            # 兜底切片只需要时长；本方法已查出 durations，封装一个只读假素材仓储
+            class _DurationOnlyAssetRepo:
+                def __init__(self, durations_map: dict[str, float]) -> None:
+                    self._durations = durations_map
+
+                def get(self, asset_id: str):
+                    if asset_id not in self._durations:
+                        return None
+
+                    class _A:
+                        pass
+
+                    a = _A()
+                    a.duration = self._durations[asset_id]
+                    return a
+
+            clips_by_asset = load_atom_clips_for_assets(
+                pool_ids,
+                atom_clip_repo=atom_repo,
+                asset_repo=_DurationOnlyAssetRepo(durations),
+            )
+            atom_candidates = flatten_candidates(clips_by_asset)
+            if atom_candidates:
+                # 历史成片已用原子片段（降权）；批次内前序变体已用（硬避让）
+                historical_atom_ids = set(
+                    self._clip_repo.list_recent_atom_clip_ids_by_user(
+                        created_by_user_id or source.created_by_user_id or "",
+                        limit=200,
+                    )
+                )
+                clips_data = reselect_clips_from_atoms(
+                    source_clips_data,
+                    atom_candidates,
+                    historical_atom_ids=historical_atom_ids,
+                    batch_used_atom_ids=(set(batch_used_atom_ids) if batch_used_atom_ids else None),
+                    rng=rng,
+                )
+        except Exception:
+            logger.warning("原子片段变体重选失败，回退整条素材选片", exc_info=True)
+            clips_data = None
+
+        if clips_data is None:
+            clips_data = reselect_clips_for_variant(
+                source_clips_data,
+                pool_ids,
+                asset_durations=durations,
+                asset_scene_points=scene_points,
+                historical_used_segments=historical,
+                batch_segments=batch_segments_resolved,
+                target_durations=target_durations,
+                rng=rng,
+            )  # 片段区间写回素材 metadata（与落库同事务；replace_all_clips_transactional 内 commit）
        for item in clips_data:
            aid = item.get("asset_id", "")
            if aid:
@@ -61,10 +61,17 @@ def writeback_edit_plan_config(
    task_id: str,
    title_config: dict | None,
    db: Session,
+    dedup_enabled: bool | None = None,
+    video_index: int | None = None,
+    assembly_mode: str | None = None,
+    script_id: str | None = None,
+    video_ratio: str | None = None,
 ) -> None:
    """任务入队成功后，回写 EditPlan.config：generation_task_id + title_config。

    用 merge 方式更新，不整体覆盖 config，避免丢失其他字段。
+    #1970：dedup_enabled 非 None 时一并写入，worker 据此决定 edge_crop/微变换；
+    PR3 叙事模式再写 assembly_mode/script_id/video_ratio（可追溯，不影响渲染）。
    失败只记日志，不影响任务创建。
    """
    if not plan_id:
@@ -80,6 +87,16 @@ 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 assembly_mode:
+            merged["assembly_mode"] = assembly_mode
+        if script_id:
+            merged["script_id"] = script_id
+        if video_ratio:
+            merged["video_ratio"] = video_ratio

        if title_config:
            # #1901 统一字段名为 "title"（worker sync_configs_to_plan 写的是 "title"）
@@ -157,6 +174,33 @@ def collect_plan_segments(
    return segs


+def collect_plan_atom_clip_ids(
+    plan_id: str,
+    clip_repo: Any,
+    *,
+    page_size: int = 500,
+) -> list[str]:
+    """分页读取 plan 所有 clips，收集已选用的原子片段 ID（#1970）。
+
+    用于批量变体间原子片段级硬避让：同一原子片段在同批次内只用一次。
+    旧路径 clips 的 atom_clip_id 为空串，自动忽略。
+    """
+    ids: list[str] = []
+    sk, pg = 0, page_size
+    while True:
+        batch = clip_repo.list_by_plan(plan_id, skip=sk, limit=pg)
+        if not batch:
+            break
+        for c in batch:
+            acid = getattr(c, "atom_clip_id", "") or ""
+            if acid:
+                ids.append(acid)
+        if len(batch) < pg:
+            break
+        sk += pg
+    return ids
+
+
 def resolve_latest_plan_by_template(
    db: Session,
    *,
@@ -0,0 +1,344 @@
+"""叙事剪辑前置服务 — #1970 PR3.
+
+叙事模式（assembly_mode='narrative'）在生成任务入队前同步完成：
+
+1. 按 script_id 读取文案（归属校验）；
+2. 按 tts_voice_source 解析音色（preset=CosyVoice 音色 id；clone=克隆档案 id，
+   解析档案归属并取其 CosyVoice voice_id）；
+3. 同步 TTS 合成（复用 tts_job 现有 workflow：提交即同步返回，未完成则轮询兜底），
+   失败直接抛 NarrativeError（HTTP 层转 4xx，任务不入队）；
+4. 把合成音频转存为配音库 audio asset（与 /tts/jobs/{id}/save-to-library 同一套
+   存储路径与元信息约定），返回 asset_id —— 下游仍以 voice_library_id（实为
+   audio asset id）消费，渲染链路零改动。
+
+积分扣点与 /tts 合成端点保持一致（ai_voice 场景），失败退费。
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import math
+import subprocess
+import tempfile
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from sqlalchemy.orm import Session
+
+from packages.adapters.sqlalchemy_impl.models import ScriptModel
+from packages.application.cosyvoice_service import CosyVoiceService
+from packages.application.tts_job.use_cases import CreateTTSJobUseCase
+from packages.application.tts_job.workflow import TTSWorkflowService
+from packages.domain import Asset, AssetLibrary, AssetLibraryKind, AssetStatus, ClassificationStatus
+from packages.domain.points_rules import calculate_points_cost
+from packages.domain.points_service import PointsService
+from packages.shared.storage import SharedStorageService
+
+logger = logging.getLogger(__name__)
+
+_POINTS_SCENE = "ai_voice"
+_SYNTH_TIMEOUT = 180.0  # 叙事配音在 HTTP 请求内同步等待，长文案分段合成时留出余量
+_CONTENT_TYPE_MAP = {"mp3": "audio/mpeg", "wav": "audio/wav", "pcm": "audio/pcm", "opus": "audio/opus"}
+
+
+class NarrativeError(Exception):
+    """叙事模式前置处理失败（文案/音色/TTS/落库）。"""
+
+    def __init__(self, message: str, *, status_code: int = 400) -> None:
+        super().__init__(message)
+        self.message = message
+        self.status_code = status_code
+
+
+@dataclass(slots=True)
+class NarrativeContext:
+    """叙事模式前置处理结果。"""
+
+    script: ScriptModel
+    voice_asset_id: str
+    tts_job_id: str
+    audio_duration: float
+
+
+def _find_or_create_voice_library(
+    *,
+    user_id: str,
+    project_repository: Any,
+    asset_library_repository: Any,
+) -> AssetLibrary:
+    """找到（或自动创建）用户 voice 素材库；与 tts.py 保存配音库逻辑一致。"""
+    projects = project_repository.find_accessible_projects(user_id)
+    if not projects:
+        raise NarrativeError("没有可用的项目，无法保存叙事配音", status_code=400)
+
+    for project in projects:
+        for lib in asset_library_repository.find_by_project(project.id):
+            kind = lib.kind.value if hasattr(lib.kind, "value") else lib.kind
+            if kind == AssetLibraryKind.VOICE.value:
+                return lib
+
+    project = projects[0]
+    library = AssetLibrary.create(project_id=project.id, name="配音素材库", kind=AssetLibraryKind.VOICE)
+    from sqlalchemy.exc import IntegrityError
+
+    try:
+        return asset_library_repository.create(library)
+    except IntegrityError:
+        session = getattr(asset_library_repository, "session", None)
+        if session is not None:
+            try:
+                session.rollback()
+            except Exception:  # noqa: BLE001 - 回滚失败不影响重查
+                logger.warning("IntegrityError 后回滚 session 失败", exc_info=True)
+        for lib in asset_library_repository.find_by_project(project.id):
+            kind = lib.kind.value if hasattr(lib.kind, "value") else lib.kind
+            if kind == AssetLibraryKind.VOICE.value:
+                return lib
+        raise NarrativeError("配音素材库创建失败，请重试", status_code=500) from None
+
+
+def _resolve_voice(
+    *,
+    user_id: str,
+    tts_voice_id: str,
+    tts_voice_source: str,
+    voice_clone_repository: Any,
+) -> tuple[str, str]:
+    """解析音色 → (CosyVoice voice_id, voice_clone_profile_id)。"""
+    if tts_voice_source == "clone":
+        profile = voice_clone_repository.get(tts_voice_id)
+        if profile is None:
+            raise NarrativeError("克隆音色不存在", status_code=404)
+        if profile.user_id != user_id:
+            raise NarrativeError("无权使用该克隆音色", status_code=403)
+        if not profile.voice_id:
+            raise NarrativeError("音色克隆尚未完成，请稍后再试", status_code=400)
+        return profile.voice_id, profile.id
+    # preset：tts_voice_id 即 CosyVoice 音色 id；与 /tts 端点一致，
+    # 若前端误传克隆档案 UUID，同样兼容解析。
+    profile = voice_clone_repository.get(tts_voice_id)
+    if profile is not None:
+        if profile.user_id != user_id:
+            raise NarrativeError("无权使用该音色", status_code=403)
+        if not profile.voice_id:
+            raise NarrativeError("音色克隆尚未完成，请稍后再试", status_code=400)
+        return profile.voice_id, profile.id
+    return tts_voice_id, ""
+
+
+def _save_tts_job_as_voice_asset(
+    *,
+    job: Any,
+    user_id: str,
+    name: str,
+    project_repository: Any,
+    asset_library_repository: Any,
+    asset_repository: Any,
+    storage_service: SharedStorageService,
+) -> Asset:
+    """把已完成 TTS job 的音频转存为配音库 audio asset（同 save-to-library 约定）。"""
+    if not job.output_audio_url and not job.output_audio_key:
+        raise NarrativeError("TTS 合成缺少输出音频", status_code=502)
+
+    library = _find_or_create_voice_library(
+        user_id=user_id,
+        project_repository=project_repository,
+        asset_library_repository=asset_library_repository,
+    )
+
+    audio_format = (job.format or "mp3").strip() or "mp3"
+    content_type = _CONTENT_TYPE_MAP.get(audio_format, "audio/mpeg")
+    storage_key = f"uploads/voice/tts/{job.id}.{audio_format}"
+
+    tmp_path: Path | None = None
+    audio_duration: float | None = None
+    file_size = 0
+    try:
+        with tempfile.NamedTemporaryFile(suffix=f".{audio_format}", delete=False) as tmp:
+            tmp_path = Path(tmp.name)
+        download_source = job.output_audio_key or job.output_audio_url
+        downloaded = storage_service.download_asset(download_source, tmp_path)
+        if not downloaded or not tmp_path.exists() or tmp_path.stat().st_size == 0:
+            raise NarrativeError("叙事配音音频转存失败", status_code=502)
+        file_size = tmp_path.stat().st_size
+        storage_service.upload_file(tmp_path, storage_key, content_type=content_type)
+
+        try:
+            proc = subprocess.run(
+                [
+                    "ffprobe",
+                    "-v",
+                    "quiet",
+                    "-print_format",
+                    "json",
+                    "-show_format",
+                    str(tmp_path),
+                ],
+                capture_output=True,
+                text=True,
+                timeout=10,
+            )
+            if proc.returncode == 0:
+                dur = float(json.loads(proc.stdout).get("format", {}).get("duration", 0))
+                if dur > 0:
+                    audio_duration = dur
+        except Exception:  # noqa: BLE001 - ffprobe 仅用于时长兜底
+            logger.warning("叙事配音 ffprobe 时长提取失败: job_id=%s", job.id, exc_info=True)
+    except NarrativeError:
+        raise
+    except Exception as e:  # noqa: BLE001
+        logger.error("叙事配音转存失败: job_id=%s, error=%s", job.id, e, exc_info=True)
+        raise NarrativeError("叙事配音音频转存失败", status_code=502) from e
+    finally:
+        if tmp_path and tmp_path.exists():
+            try:
+                tmp_path.unlink()
+            except OSError:
+                pass
+
+    metadata_: dict[str, object] = {
+        "source": "tts_job",
+        "tts_job_id": job.id,
+        "narrative": True,
+        "format": job.format,
+        "sample_rate": job.sample_rate,
+        "voice_id": job.voice_id,
+        "voice_name": job.voice_model or "",
+    }
+    if job.metadata:
+        for key in ("speed", "language"):
+            if key in job.metadata:
+                metadata_[key] = job.metadata[key]
+
+    asset = Asset.create(
+        project_id=library.project_id,
+        library_id=library.id,
+        name=name or f"叙事配音-{job.id[:8]}",
+        storage_key=storage_key,
+        mime_type=content_type,
+        metadata=metadata_,
+        file_size=file_size,
+        duration=job.duration or audio_duration or None,
+        status=AssetStatus.READY,
+        classification_status=ClassificationStatus.PENDING,
+        uploaded_by_user_id=user_id,
+    )
+    try:
+        return asset_repository.create(asset)
+    except Exception as e:  # noqa: BLE001
+        logger.error("叙事配音 asset 落库失败，清理 OSS: %s, error=%s", storage_key, e, exc_info=True)
+        try:
+            storage_service.delete_file(storage_key)
+        except Exception:  # noqa: BLE001
+            logger.warning("清理孤儿 OSS 文件失败: %s", storage_key, exc_info=True)
+        raise NarrativeError("叙事配音保存失败，请重试", status_code=502) from e
+
+
+def prepare_narrative_voice(
+    *,
+    db: Session,
+    user_id: str,
+    script_id: str,
+    tts_voice_id: str,
+    tts_voice_source: str,
+    tts_repository: Any,
+    cosyvoice_service: CosyVoiceService,
+    voice_clone_repository: Any,
+    asset_repository: Any,
+    asset_library_repository: Any,
+    project_repository: Any,
+    storage_service: SharedStorageService,
+    points_enabled: bool = False,
+    is_member: bool = False,
+    member_type: str | None = None,
+) -> NarrativeContext:
+    """叙事模式入队前同步合成配音并落为 audio asset。
+
+    Raises:
+        NarrativeError: 文案缺失/归属不符、音色不可用、TTS 失败、转存失败。
+    """
+    script = db.query(ScriptModel).filter(ScriptModel.id == script_id, ScriptModel.user_id == user_id).first()
+    if script is None:
+        raise NarrativeError("文案不存在或无权使用", status_code=404)
+    content = (script.content or "").strip()
+    if not content:
+        raise NarrativeError("文案内容为空，无法合成配音", status_code=400)
+
+    actual_voice_id, clone_profile_id = _resolve_voice(
+        user_id=user_id,
+        tts_voice_id=tts_voice_id,
+        tts_voice_source=tts_voice_source,
+        voice_clone_repository=voice_clone_repository,
+    )
+
+    # 积分扣点（与 /tts 合成端点同口径），失败时在合成失败分支退费
+    points_svc = PointsService() if points_enabled else None
+    points_deducted = 0
+    if points_svc is not None:
+        est_minutes = max(1.0, math.ceil(len(content) / 240))
+        points_deducted = calculate_points_cost(
+            _POINTS_SCENE,
+            is_member=is_member,
+            duration_minutes=est_minutes,
+            member_type=member_type,
+        )
+        deduct_res = points_svc.deduct_points(user_id, points_deducted, _POINTS_SCENE, db)
+        if not deduct_res["success"]:
+            raise NarrativeError(
+                f"积分不足，需要 {points_deducted} 积分，当前余额 {deduct_res['balance']}",
+                status_code=402,
+            )
+
+    use_case = CreateTTSJobUseCase(tts_repository)
+    job = use_case.execute(
+        user_id=user_id,
+        input_text=content,
+        voice_id=actual_voice_id,
+        voice_clone_profile_id=clone_profile_id,
+        metadata={"speed": 1.0, "emotion": "", "language": "zh-CN", "narrative": True, "script_id": script_id},
+    )
+
+    workflow = TTSWorkflowService(repository=tts_repository, cosyvoice_service=cosyvoice_service)
+    try:
+        job = workflow.start_synthesis(job.id)
+        if not job.is_completed:
+            job = workflow.poll_and_process_synthesis(job.id, timeout=_SYNTH_TIMEOUT)
+    except Exception as e:  # noqa: BLE001 - 同步合成异常统一转 NarrativeError
+        logger.error("叙事配音 TTS 合成失败: job_id=%s, error=%s", job.id, e, exc_info=True)
+        try:
+            workflow.process_synthesis_failure(job.id, str(e))
+        except Exception:  # noqa: BLE001
+            logger.warning("标记叙事 TTS job 失败出错: job_id=%s", job.id, exc_info=True)
+        if points_deducted and points_svc is not None:
+            try:
+                points_svc.refund_points(user_id, points_deducted, _POINTS_SCENE, db, ref_id=job.id)
+            except Exception:  # noqa: BLE001
+                logger.warning("叙事 TTS 失败退积分异常: job_id=%s", job.id, exc_info=True)
+        raise NarrativeError(f"配音合成失败：{e}", status_code=502) from e
+
+    if not job.is_completed:
+        if points_deducted and points_svc is not None:
+            try:
+                points_svc.refund_points(user_id, points_deducted, _POINTS_SCENE, db, ref_id=job.id)
+            except Exception:  # noqa: BLE001
+                logger.warning("叙事 TTS 未完成退积分异常: job_id=%s", job.id, exc_info=True)
+        raise NarrativeError("配音合成未完成，请稍后重试", status_code=504)
+
+    asset = _save_tts_job_as_voice_asset(
+        job=job,
+        user_id=user_id,
+        name=(script.title or "叙事配音")[:60],
+        project_repository=project_repository,
+        asset_library_repository=asset_library_repository,
+        asset_repository=asset_repository,
+        storage_service=storage_service,
+    )
+
+    return NarrativeContext(
+        script=script,
+        voice_asset_id=asset.id,
+        tts_job_id=job.id,
+        audio_duration=float(job.duration or asset.duration or 0.0),
+    )
@@ -22,6 +22,11 @@ from packages.adapters.sqlalchemy_impl import (
    SQLAlchemyEditPlanClipRepository,
    SQLAlchemyEditPlanRepository,
 )
+from packages.domain.atom_clip_resolver import load_atom_clips_for_assets
+from packages.domain.atom_clip_selector import (
+    estimate_required_clip_count,
+    select_atom_clips,
+)
 from packages.domain.config_schemas import normalize_plan_config
 from packages.domain.edit_plan import EditPlan
 from packages.domain.edit_plan_clip import EditPlanClip
@@ -52,10 +57,12 @@ class PlanGeneratorService:
    基于模板 + 素材，自动生成 EditPlan 及 EditPlanClip 列表。
    """

-    def __init__(self, db: Session, asset_repo=None) -> None:
+    def __init__(self, db: Session, asset_repo=None, atom_clip_repo=None) -> None:
        self._plan_repo = SQLAlchemyEditPlanRepository(db)
        self._clip_repo = SQLAlchemyEditPlanClipRepository(db)
        self._asset_repo = asset_repo
+        # #1970 原子化切片：可选注入；未注入时走旧的整条素材选片路径（向后兼容）
+        self._atom_clip_repo = atom_clip_repo

    # ── 公开接口 ─────────────────────────────────────────────────────────────

@@ -121,18 +128,34 @@ class PlanGeneratorService:

        # 4. 按 editing_mode 分配素材
        if asset_ids:
-            # 获取素材时长信息，用于随机起始时间
-            asset_durations = None
-            if self._asset_repo:
-                asset_durations = self._fetch_asset_durations(asset_ids)
-            self._distribute_assets(
-                clips,
-                asset_ids,
-                editing_mode,
-                random_selection=random_preview,
-                asset_durations=asset_durations,
-                user_id=created_by_user_id,
-            )
+            # #1970 原子化切片：素材 clip 从 atom_clips 表选取（未就绪自动内存兜底）。
+            # 预览随机模式保持旧路径（整条素材 + 随机起点），与现有预览契约一致。
+            atom_applied = False
+            if not random_preview and self._atom_clip_repo is not None:
+                try:
+                    atom_applied = self._distribute_atom_clips(
+                        clips,
+                        asset_ids,
+                        editing_mode,
+                        user_id=created_by_user_id,
+                    )
+                except Exception:
+                    logger.warning("原子片段选片失败，回退整条素材选片", exc_info=True)
+                    atom_applied = False
+
+            if not atom_applied:
+                # 获取素材时长信息，用于随机起始时间
+                asset_durations = None
+                if self._asset_repo:
+                    asset_durations = self._fetch_asset_durations(asset_ids)
+                self._distribute_assets(
+                    clips,
+                    asset_ids,
+                    editing_mode,
+                    random_selection=random_preview,
+                    asset_durations=asset_durations,
+                    user_id=created_by_user_id,
+                )

        # 5. 持久化所有 clips 并计算总时长
        created_clips: list[EditPlanClip] = []
@@ -259,6 +282,103 @@ class PlanGeneratorService:
            external_used_segments=external_used_segments,
        )

+    def _distribute_atom_clips(
+        self,
+        clips: list[EditPlanClip],
+        asset_ids: list[str],
+        editing_mode: str,
+        *,
+        user_id: str = "",
+    ) -> bool:
+        """#1970 原子化切片选片（就地修改 clips，未持久化）.
+
+        从 ``asset_atom_clips`` 表按原子片段选取；老素材/切片未就绪的素材
+        内存兜底切片。同一原子片段在一次方案中只用一次；跨视频避让走
+        edit_plan_clips.atom_clip_id 最近使用记录。
+
+        Returns:
+            True 表示原子片段选片成功；False 表示无可用片段，调用方应回退
+            到旧的整条素材 distribute_assets。
+        """
+        # 1. 加载候选原子片段（DB + 兜底）
+        clips_by_asset = load_atom_clips_for_assets(
+            asset_ids,
+            atom_clip_repo=self._atom_clip_repo,
+            asset_repo=self._asset_repo,
+        )
+        if not clips_by_asset:
+            return False
+
+        # 2. 最近使用片段（跨视频原子片段级避让）
+        recently_used: set[str] = set()
+        if user_id and hasattr(self._clip_repo, "list_recent_atom_clip_ids_by_user"):
+            try:
+                recently_used = set(self._clip_repo.list_recent_atom_clip_ids_by_user(user_id, limit=200))
+            except Exception:
+                logger.warning("跨视频原子片段避让查询失败", exc_info=True)
+
+        # 3. 片段需求估算：无配音时按 clips 数量；voice_over 的配音总时长存于
+        #    clip.config["voice_duration"]，按 平均片段时长≈需要片段数 估算
+        voice_total = 0.0
+        for c in clips:
+            cfg_vd = c.config.get("voice_duration") if c.config else None
+            if cfg_vd:
+                voice_total += float(cfg_vd)
+        avg_clip_target = sum(float(c.duration or 0.0) for c in clips) / max(len(clips), 1)
+        required_count = estimate_required_clip_count(
+            voice_total or sum(float(c.duration or 0.0) for c in clips),
+            avg_clip_target or 3.5,
+        )
+        required_count = max(required_count, len(clips))
+
+        rng = random.Random()
+
+        # 4. 正式生成：先按素材 smart_score 对素材池排序，再展开为片段池
+        #    （同素材的片段保持连续，高分素材的片段排在前面优先入选）
+        if self._asset_repo:
+            asset_order = self._sort_assets_by_smart_score(list(clips_by_asset.keys()))
+            ordered: dict[str, list] = {}
+            for aid in asset_order:
+                if aid in clips_by_asset:
+                    ordered[aid] = clips_by_asset[aid]
+            clips_by_asset = ordered
+
+        candidates: list = []
+        for asset_clips in clips_by_asset.values():
+            candidates.extend(asset_clips)
+
+        # 5. 逐虚拟片段选片：评分排序，同片段不重复使用
+        used_atom_ids: set[str] = set()
+        asset_usage: dict[str, int] = {}
+        assigned = 0
+        for clip in clips:
+            # 对每个虚拟片段重新评分（usage_count 随选择动态变化）
+            scored = select_atom_clips(
+                candidates,
+                target_duration=float(clip.duration or 0.0),
+                used_atom_clip_ids=used_atom_ids,
+                asset_usage_counts=asset_usage,
+                recently_used_atom_ids=recently_used,
+                required_count=required_count,
+                limit=1,
+                rng=rng,
+            )
+            if not scored:
+                # 候选耗尽（同片段不可重复），交由调用方回退或留白
+                continue
+            picked = scored[0]
+            clip.asset_id = picked.asset_id
+            clip.atom_clip_id = picked.atom_clip_id
+            clip.start_time = round(picked.start_time, 3)
+            clip.duration = round(picked.duration, 3)
+            used_atom_ids.add(picked.atom_clip_id)
+            asset_usage[picked.asset_id] = asset_usage.get(picked.asset_id, 0) + 1
+            assigned += 1
+
+        if assigned == 0:
+            return False
+        return True
+
    def _fetch_asset_scene_points(self, asset_ids: list[str]) -> dict[str, list[float]]:
        """从素材 metadata 读取场景切换点缓存（无缓存的素材不包含在结果中）。"""
        points_map: dict[str, list[float]] = {}
@@ -38,7 +38,12 @@ def transcribe_to_text(media_path: str | Path) -> str:
        ASRTranscriptionError: ASR 调用失败
    """
    # 延迟导入，避免循环依赖和启动时副作用
-    from apps.worker.services.asr_service_factory import get_asr_service
+    try:
+        from apps.worker.services.asr_service_factory import get_asr_service
+    except ImportError as exc:
+        # API 镜像未打包 worker 代码（本地 ASR 依赖 worker 的 asr_service_factory）
+        logger.warning("本地 ASR 不可用（apps.worker 未安装）: %s", exc)
+        raise ASRNotConfiguredError("本地 ASR 服务不可用（worker 模块未安装）") from exc

    asr = get_asr_service()
    if asr is None:
@@ -212,52 +212,50 @@ const Step1EditMode: React.FC<Step1EditModeProps> = ({
        className="xx-form-field"
        style={{
          marginTop: 16,
-          display: "flex",
-          alignItems: "center",
-          justifyContent: "space-between",
          padding: "12px 16px",
          background: "#f9fafb",
          borderRadius: 8,
        }}
      >
-        <div>
-          <div style={{ fontSize: 14, fontWeight: 500, color: "#111" }}>
+        <div style={{ display: "flex", alignItems: "center", gap: 8 }}>
+          <span style={{ fontSize: 14, fontWeight: 500, color: "#111" }}>
            🎯 智能降重 {dedupEnabled ? "已开启" : "已关闭"}
-          </div>
-          <div style={{ fontSize: 12, color: "#6b7280", marginTop: 2 }}>
-            自动对画面做微调，避免查重不过
-          </div>
-        </div>
-        <button
-          type="button"
-          onClick={() => onDedupEnabledChange(!dedupEnabled)}
-          style={{
-            width: 44,
-            height: 24,
-            borderRadius: 12,
-            border: "none",
-            background: dedupEnabled ? PURPLE : "#d1d5db",
-            position: "relative",
-            cursor: "pointer",
-            transition: "background 0.2s",
-            padding: 0,
-          }}
-          aria-label="toggle dedup"
-        >
-          <span
+          </span>
+          <button
+            type="button"
+            onClick={() => onDedupEnabledChange(!dedupEnabled)}
            style={{
-              position: "absolute",
-              top: 2,
-              left: dedupEnabled ? 22 : 2,
-              width: 20,
-              height: 20,
-              borderRadius: "50%",
-              background: "#fff",
-              transition: "left 0.2s",
-              boxShadow: "0 1px 3px rgba(0,0,0,0.2)",
+              width: 44,
+              height: 24,
+              borderRadius: 12,
+              border: "none",
+              background: dedupEnabled ? PURPLE : "#d1d5db",
+              position: "relative",
+              cursor: "pointer",
+              transition: "background 0.2s",
+              padding: 0,
+              flexShrink: 0,
            }}
-          />
-        </button>
+            aria-label="toggle dedup"
+          >
+            <span
+              style={{
+                position: "absolute",
+                top: 2,
+                left: dedupEnabled ? 22 : 2,
+                width: 20,
+                height: 20,
+                borderRadius: "50%",
+                background: "#fff",
+                transition: "left 0.2s",
+                boxShadow: "0 1px 3px rgba(0,0,0,0.2)",
+              }}
+            />
+          </button>
+        </div>
+        <div style={{ fontSize: 12, color: "#6b7280", marginTop: 4 }}>
+          自动对画面做微调，避免查重不过
+        </div>
      </div>
    </div>
  )
@@ -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},"
@@ -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
@@ -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"))
@@ -27,6 +27,7 @@ celery_app.conf.broker_transport_options = {"visibility_timeout": 4 * 60 * 60}
 celery_app.conf.imports = (
    "worker_app.tasks.health",
    "worker_app.tasks.ingest",
+    "worker_app.tasks.atom_clips",
    "worker_app.tasks.classification",
    "worker_app.tasks.generation",
    "worker_app.tasks.voice_extraction",
@@ -53,12 +53,17 @@ def __getattr__(name: str):
        from .batch_thumbnail import batch_generate_thumbnails

        return batch_generate_thumbnails
+    elif name == "generate_atom_clips":
+        from .atom_clips import generate_atom_clips
+
+        return generate_atom_clips
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")


 __all__ = [
    "batch_generate_thumbnails",
    "classify_asset",
+    "generate_atom_clips",
    "generate_video",
    "healthcheck",
    "ingest_asset",
@@ -0,0 +1,81 @@
+"""素材原子切片 Celery 任务 — #1970 智能剪辑流程重构 P1.
+
+素材入库预处理完成（ingest 置 READY）后异步触发：
+根据素材时长和已缓存的 scdet 切换点计算原子片段并落库。
+失败不阻断素材入库主流程（atom_clips 未就绪时选片有内存兜底）。
+"""
+
+from __future__ import annotations
+
+from celery.utils.log import get_task_logger
+from worker_app.celery_app import celery_app
+from worker_app.db import SessionLocal
+
+from packages.adapters.sqlalchemy_impl.asset_atom_clip_repository import (
+    SQLAlchemyAssetAtomClipRepository,
+)
+from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
+from packages.domain.atom_clip_service import compute_atom_clips
+from packages.domain.plan_generator_utils import extract_scene_points_from_metadata
+
+logger = get_task_logger(__name__)
+
+
+@celery_app.task(name="worker.generate_atom_clips")
+def generate_atom_clips(asset_id: str) -> dict:
+    """为单条视频素材生成原子片段。
+
+    Returns:
+        任务结果 dict：status / asset_id / clips_count。
+    """
+    db = SessionLocal()
+    try:
+        asset_repo = SQLAlchemyAssetRepository(db)
+        atom_repo = SQLAlchemyAssetAtomClipRepository(db)
+
+        asset = asset_repo.find_by_id(asset_id)
+        if asset is None:
+            return {"status": "skipped", "reason": "asset not found", "asset_id": asset_id}
+
+        # 仅视频素材切片
+        if asset.mime_type and not asset.mime_type.startswith("video/"):
+            return {"status": "skipped", "reason": "not a video", "asset_id": asset_id}
+        if not asset.duration or asset.duration <= 0:
+            return {"status": "skipped", "reason": "invalid duration", "asset_id": asset_id}
+
+        # 已生成过则幂等跳过（重新切片需先显式删除）
+        existing = atom_repo.count_by_asset(asset_id)
+        if existing > 0:
+            return {
+                "status": "skipped",
+                "reason": "already generated",
+                "asset_id": asset_id,
+                "clips_count": existing,
+            }
+
+        scene_points = extract_scene_points_from_metadata(asset.metadata)
+        # P1 阶段继承素材的标签 ID；片段级语义标签是 P2 功能
+        tags = list(getattr(asset, "tag_ids", []) or [])
+
+        clips = compute_atom_clips(
+            asset_id=asset_id,
+            duration=float(asset.duration),
+            scene_change_points=scene_points,
+            tags=tags,
+        )
+        if not clips:
+            return {"status": "skipped", "reason": "no clips computed", "asset_id": asset_id}
+
+        atom_repo.batch_create(clips)
+        logger.info(
+            "[atom_clips] asset_id=%s 生成 %d 个原子片段",
+            asset_id,
+            len(clips),
+        )
+        return {"status": "completed", "asset_id": asset_id, "clips_count": len(clips)}
+    except Exception as exc:  # noqa: BLE001 - 后台任务兜底，失败不阻断主流程
+        db.rollback()
+        logger.exception("[atom_clips] asset_id=%s 生成失败: %s", asset_id, exc)
+        return {"status": "failed", "asset_id": asset_id, "error": str(exc)}
+    finally:
+        db.close()
@@ -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(
@@ -808,6 +808,21 @@ def ingest_asset(job_id: str) -> dict:

        db.commit()

+        # ── #1970 素材原子切片：视频 READY 后异步触发，失败不阻断入库 ──
+        # atom_clips 未就绪时选片逻辑有内存兜底（compute_fallback_clips）。
+        try:
+            if media_type == "video" and float(asset.duration or 0) > 0:
+                celery_app.send_task(
+                    "worker.generate_atom_clips",
+                    args=[asset.id],
+                )
+        except Exception as atom_err:  # noqa: BLE001
+            logger.warning(
+                "触发原子切片任务失败（不影响入库）: asset_id=%s err=%s",
+                asset.id,
+                atom_err,
+            )
+
        return {
            "status": "completed",
            "job_id": job.id,
@@ -0,0 +1,112 @@
+"""素材原子片段仓储 SQLAlchemy 实现。"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from sqlalchemy.orm import Session
+
+from packages.adapters.sqlalchemy_impl.models import AssetAtomClipModel
+from packages.domain.asset_atom_clip import AssetAtomClip
+
+
+class SQLAlchemyAssetAtomClipRepository:
+    def __init__(self, session: Session):
+        self.session = session
+
+    def create(self, clip: AssetAtomClip) -> AssetAtomClip:
+        model = self._to_model(clip)
+        self.session.add(model)
+        self.session.flush()
+        self.session.commit()
+        return clip
+
+    def batch_create(self, clips: list[AssetAtomClip]) -> list[AssetAtomClip]:
+        if not clips:
+            return []
+        models = [self._to_model(c) for c in clips]
+        self.session.add_all(models)
+        self.session.flush()
+        self.session.commit()
+        return clips
+
+    def find_by_asset(self, asset_id: str) -> list[AssetAtomClip]:
+        models = (
+            self.session.query(AssetAtomClipModel)
+            .filter(AssetAtomClipModel.asset_id == asset_id)
+            .order_by(AssetAtomClipModel.clip_index.asc())
+            .all()
+        )
+        return [self._to_domain(m) for m in models]
+
+    def find_by_id(self, clip_id: str) -> AssetAtomClip | None:
+        model = self.session.query(AssetAtomClipModel).filter(AssetAtomClipModel.id == clip_id).first()
+        if model is None:
+            return None
+        return self._to_domain(model)
+
+    def find_by_ids(self, clip_ids: list[str]) -> list[AssetAtomClip]:
+        if not clip_ids:
+            return []
+        models = self.session.query(AssetAtomClipModel).filter(AssetAtomClipModel.id.in_(clip_ids)).all()
+        return [self._to_domain(m) for m in models]
+
+    def delete_by_asset(self, asset_id: str) -> int:
+        count = (
+            self.session.query(AssetAtomClipModel)
+            .filter(AssetAtomClipModel.asset_id == asset_id)
+            .delete(synchronize_session=False)
+        )
+        self.session.commit()
+        return count
+
+    def count_by_asset(self, asset_id: str) -> int:
+        return self.session.query(AssetAtomClipModel).filter(AssetAtomClipModel.asset_id == asset_id).count()
+
+    def find_candidates_for_selection(
+        self,
+        asset_ids: list[str],
+        *,
+        min_duration: float | None = None,
+        max_duration: float | None = None,
+        limit: int = 100,
+    ) -> list[AssetAtomClip]:
+        """按筛选条件查找候选原子片段，按时长排序。用于选片逻辑。"""
+        query = self.session.query(AssetAtomClipModel).filter(AssetAtomClipModel.asset_id.in_(asset_ids))
+        if min_duration is not None:
+            query = query.filter(AssetAtomClipModel.duration >= min_duration)
+        if max_duration is not None:
+            query = query.filter(AssetAtomClipModel.duration <= max_duration)
+        query = query.order_by(AssetAtomClipModel.clip_index.asc())
+        if limit > 0:
+            query = query.limit(limit)
+        models = query.all()
+        return [self._to_domain(m) for m in models]
+
+    def _to_model(self, clip: AssetAtomClip) -> AssetAtomClipModel:
+        return AssetAtomClipModel(
+            id=clip.id,
+            asset_id=clip.asset_id,
+            start_time=clip.start_time,
+            end_time=clip.end_time,
+            duration=clip.duration,
+            clip_index=clip.clip_index,
+            tags=clip.tags,
+            scene_change_at=clip.scene_change_at,
+            is_fallback=clip.is_fallback,
+            created_at=clip.created_at or datetime.now(UTC),
+        )
+
+    def _to_domain(self, model: AssetAtomClipModel) -> AssetAtomClip:
+        return AssetAtomClip(
+            id=model.id,
+            asset_id=model.asset_id,
+            start_time=model.start_time,
+            end_time=model.end_time,
+            duration=model.duration,
+            clip_index=model.clip_index,
+            tags=model.tags or [],
+            scene_change_at=model.scene_change_at,
+            is_fallback=model.is_fallback,
+            created_at=model.created_at,
+        )
@@ -50,6 +50,7 @@ class SQLAlchemyEditPlanClipRepository:
            order=clip.order,
            template_clip_config_id=clip.template_clip_config_id,
            asset_id=clip.asset_id,
+            atom_clip_id=getattr(clip, "atom_clip_id", "") or "",
            text_content=clip.text_content,
            start_time=clip.start_time,
            duration=clip.duration,
@@ -74,6 +75,7 @@ class SQLAlchemyEditPlanClipRepository:
        model.order = clip.order
        model.template_clip_config_id = clip.template_clip_config_id
        model.asset_id = clip.asset_id
+        model.atom_clip_id = getattr(clip, "atom_clip_id", "") or ""
        model.text_content = clip.text_content
        model.start_time = clip.start_time
        model.duration = clip.duration
@@ -120,6 +122,7 @@ class SQLAlchemyEditPlanClipRepository:
            order=model.order,
            template_clip_config_id=model.template_clip_config_id or "",
            asset_id=model.asset_id or "",
+            atom_clip_id=getattr(model, "atom_clip_id", "") or "",
            text_content=model.text_content or "",
            start_time=model.start_time or 0.0,
            duration=model.duration or 0.0,
@@ -193,3 +196,53 @@ class SQLAlchemyEditPlanClipRepository:
            result[asset_id].append((start_time or 0.0, (start_time or 0.0) + (duration or 0.0)))

        return result
+
+    def list_recent_atom_clip_ids_by_user(
+        self,
+        user_id: str,
+        *,
+        limit: int = 200,
+    ) -> list[str]:
+        """#1970 跨视频原子片段级避让：查询用户最近成片用过的 atom_clip_id.
+
+        只统计已完成 plan 下已渲染且 atom_clip_id 非空的 clips，按 plan
+        创建时间倒序，返回去重后的 ID 列表。
+        """
+        from packages.adapters.sqlalchemy_impl.models import EditPlanModel
+
+        if not user_id:
+            return []
+
+        recent_plan_ids = [
+            row[0]
+            for row in self.session.query(EditPlanModel.id)
+            .filter(
+                EditPlanModel.created_by_user_id == user_id,
+                EditPlanModel.status == "completed",
+            )
+            .order_by(EditPlanModel.created_at.desc())
+            .limit(50)
+            .all()
+        ]
+        if not recent_plan_ids:
+            return []
+
+        rows = (
+            self.session.query(EditPlanClipModel.atom_clip_id)
+            .filter(
+                EditPlanClipModel.plan_id.in_(recent_plan_ids),
+                EditPlanClipModel.status == "rendered",
+                EditPlanClipModel.atom_clip_id.isnot(None),
+                EditPlanClipModel.atom_clip_id != "",
+            )
+            .all()
+        )
+        seen: set[str] = set()
+        ordered: list[str] = []
+        for (atom_clip_id,) in rows:
+            if atom_clip_id and atom_clip_id not in seen:
+                seen.add(atom_clip_id)
+                ordered.append(atom_clip_id)
+            if len(ordered) >= limit:
+                break
+        return ordered
@@ -1,7 +1,20 @@
 from datetime import UTC, datetime
 from typing import Any

-from sqlalchemy import JSON, Boolean, Column, DateTime, Float, Index, Integer, String, Text, UniqueConstraint, text
+from sqlalchemy import (
+    JSON,
+    Boolean,
+    Column,
+    DateTime,
+    Float,
+    ForeignKey,
+    Index,
+    Integer,
+    String,
+    Text,
+    UniqueConstraint,
+    text,
+)
 from sqlalchemy.orm import declarative_base

 Base: Any = declarative_base()
@@ -234,6 +247,8 @@ class EditPlanClipModel(Base):
    order = Column(Integer, nullable=False)
    template_clip_config_id = Column(String(36), nullable=False, default="", index=True)
    asset_id = Column(String(36), nullable=False, default="", index=True)
+    # #1970 原子化切片：片段选中的原子片段 ID（空串表示旧的整条素材选取路径）
+    atom_clip_id = Column(String(36), nullable=False, default="", index=True)
    text_content = Column(Text, nullable=False, default="")
    start_time = Column(Float, nullable=False, default=0.0)
    duration = Column(Float, nullable=False, default=0.0)
@@ -801,6 +816,32 @@ class PointsOrderModel(Base):
    created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(UTC))


+class AssetAtomClipModel(Base):
+    """素材原子片段 ORM 模型 (#1970 智能剪辑流程重构)。
+
+    逻辑切分单元，不物理切割视频文件。
+    """
+
+    __tablename__ = "asset_atom_clips"
+    __table_args__ = (UniqueConstraint("asset_id", "clip_index", name="uq_asset_atom_clips_asset_index"),)
+
+    id = Column(String(36), primary_key=True)
+    asset_id = Column(
+        String(36),
+        ForeignKey("assets.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    start_time = Column(Float, nullable=False)
+    end_time = Column(Float, nullable=False)
+    duration = Column(Float, nullable=False)
+    clip_index = Column(Integer, nullable=False)
+    tags = Column(JSON, nullable=False, default=list)
+    scene_change_at = Column(Float, nullable=True)
+    is_fallback = Column(Boolean, nullable=False, default=False)
+    created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(UTC))
+
+
 class DailyUsageRecordModel(Base):
    """每日使用记录 ORM 模型 (#1895)"""

@@ -1,5 +1,18 @@
 """Domain package for core business entities and rules."""

+from . import atom_clip_resolver
+from .asset_atom_clip import AssetAtomClip
+from .atom_clip_selector import (
+    ScoredAtomClip,
+    clips_to_segments,
+    estimate_required_clip_count,
+    score_atom_clip,
+    select_atom_clips,
+)
+from .atom_clip_service import (
+    compute_atom_clips,
+    compute_fallback_clips,
+)
 from .classification import (
    AssetClassification,
    ClassificationJob,
@@ -37,6 +50,15 @@ from .voice_library import VoiceLibraryItem

 __all__ = [
    "Asset",
+    "AssetAtomClip",
+    "ScoredAtomClip",
+    "clips_to_segments",
+    "compute_atom_clips",
+    "compute_fallback_clips",
+    "estimate_required_clip_count",
+    "score_atom_clip",
+    "select_atom_clips",
+    "atom_clip_resolver",
    "AssetClassification",
    "DailyUsageRecord",
    "PointsAccount",
@@ -0,0 +1,81 @@
+"""素材原子片段（Atom Clip）领域实体 — #1970 智能剪辑流程重构。
+
+原子片段是素材的逻辑切分单元，不物理切割视频文件。
+每条记录指向某条素材的一段 [start_time, end_time] 区间。
+"""
+
+from __future__ import annotations
+
+import uuid
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+
+
+@dataclass
+class AssetAtomClip:
+    """素材原子片段。
+
+    Attributes:
+        id: 唯一标识。
+        asset_id: 所属素材 ID。
+        start_time: 片段起始时间（秒，浮点）。
+        end_time: 片段结束时间（秒，浮点）。
+        duration: 片段时长 = end_time - start_time（秒）。
+        clip_index: 在同一素材内的顺序编号（从 0 开始）。
+        tags: 继承自素材的标签，JSONB 存储，可为空列表。
+        scene_change_at: 片段尾部是否对齐了 scdet 镜头切换点（存储该切点的精确时间），
+                         未对齐时为 None。
+        is_fallback: 是否为兜底逻辑在内存中生成的临时片段（不入库）。
+        created_at: 创建时间。
+    """
+
+    id: str
+    asset_id: str
+    start_time: float
+    end_time: float
+    duration: float
+    clip_index: int
+    tags: list[str] = field(default_factory=list)
+    scene_change_at: float | None = None
+    is_fallback: bool = False
+    created_at: datetime | None = None
+
+    def __post_init__(self):
+        if not self.id:
+            self.id = str(uuid.uuid4())
+        if self.duration <= 0:
+            self.duration = round(self.end_time - self.start_time, 3)
+        if self.duration < 0:
+            raise ValueError(f"duration must be >= 0, got start={self.start_time}, end={self.end_time}")
+        if self.start_time < 0:
+            raise ValueError(f"start_time must be >= 0, got {self.start_time}")
+        if self.end_time <= self.start_time:
+            raise ValueError(f"end_time must be > start_time, got start={self.start_time}, end={self.end_time}")
+        if self.clip_index < 0:
+            raise ValueError(f"clip_index must be >= 0, got {self.clip_index}")
+        if self.created_at is None:
+            self.created_at = datetime.now(UTC)
+
+    @classmethod
+    def create(
+        cls,
+        asset_id: str,
+        start_time: float,
+        end_time: float,
+        clip_index: int,
+        tags: list[str] | None = None,
+        scene_change_at: float | None = None,
+        is_fallback: bool = False,
+    ) -> AssetAtomClip:
+        """工厂方法：创建一个新的原子片段。"""
+        return cls(
+            id="",  # __post_init__ 会自动生成
+            asset_id=asset_id,
+            start_time=round(start_time, 3),
+            end_time=round(end_time, 3),
+            duration=round(end_time - start_time, 3),
+            clip_index=clip_index,
+            tags=tags or [],
+            scene_change_at=scene_change_at,
+            is_fallback=is_fallback,
+        )
@@ -0,0 +1,104 @@
+"""原子片段加载与兜底 — #1970 智能剪辑流程重构 P1.
+
+选片前从 ``asset_atom_clips`` 表加载素材池的原子片段；老素材/切片任务尚未
+完成/切片失败导致某些素材没有片段时，按需求兜底：内存中按 3-6 秒临时均匀
+切片（不存库，片段标记 is_fallback=True）。
+
+本模块对 repository 做鸭子类型约束（只需 find_by_asset / find_candidates_for_selection
+和 asset_repo.get），方便 API 侧（SQLAlchemy）与 worker 侧复用，也便于单测注入内存假实现。
+"""
+
+from __future__ import annotations
+
+import logging
+
+from packages.domain.asset_atom_clip import AssetAtomClip
+from packages.domain.atom_clip_service import compute_fallback_clips
+
+logger = logging.getLogger(__name__)
+
+# 兜底均匀切片步长（秒），落在 3~6s 区间中段
+FALLBACK_CLIP_SECONDS = 4.5
+
+
+def load_atom_clips_for_assets(
+    asset_ids: list[str],
+    *,
+    atom_clip_repo,
+    asset_repo=None,
+) -> dict[str, list[AssetAtomClip]]:
+    """加载素材池的原子片段（缺失素材走内存兜底）.
+
+    Args:
+        asset_ids: 候选素材 ID（去重保序）。
+        atom_clip_repo: AssetAtomClipRepository 实现（需有
+            ``find_candidates_for_selection`` 或 ``find_by_asset``）。
+        asset_repo: 可选，素材仓储（需有 ``get``），用于读取时长兜底切片。
+            为 None 时，没有原子片段的素材直接跳过（不兜底）。
+
+    Returns:
+        {asset_id: [AssetAtomClip, ...]}，仅包含至少有一个片段的素材，
+        片段按 clip_index 排序。
+    """
+    result: dict[str, list[AssetAtomClip]] = {}
+    unique_ids = list(dict.fromkeys(asset_ids))
+    if not unique_ids:
+        return result
+
+    # 1. 批量查询已生成的原子片段
+    persisted: dict[str, list[AssetAtomClip]] = {}
+    try:
+        if hasattr(atom_clip_repo, "find_candidates_for_selection"):
+            clips = atom_clip_repo.find_candidates_for_selection(unique_ids, limit=0)
+        else:
+            clips = []
+            for asset_id in unique_ids:
+                clips.extend(atom_clip_repo.find_by_asset(asset_id))
+        for clip in clips:
+            persisted.setdefault(clip.asset_id, []).append(clip)
+    except Exception:
+        logger.warning("加载 atom_clips 失败，全部走内存兜底", exc_info=True)
+        persisted = {}
+
+    for asset_id in unique_ids:
+        clips = persisted.get(asset_id)
+        if clips:
+            clips.sort(key=lambda c: c.clip_index)
+            result[asset_id] = clips
+            continue
+
+        # 2. 兜底：内存均匀切片（不存库）
+        if asset_repo is None:
+            continue
+        duration = _safe_asset_duration(asset_repo, asset_id)
+        if duration <= 0:
+            continue
+        result[asset_id] = compute_fallback_clips(
+            asset_id,
+            duration,
+            clip_seconds=FALLBACK_CLIP_SECONDS,
+        )
+
+    return result
+
+
+def flatten_candidates(
+    clips_by_asset: dict[str, list[AssetAtomClip]],
+) -> list[AssetAtomClip]:
+    """把 {asset_id: [clips]} 摊平为候选片段列表（素材顺序内片段有序）。"""
+    flat: list[AssetAtomClip] = []
+    for clips in clips_by_asset.values():
+        flat.extend(clips)
+    return flat
+
+
+def _safe_asset_duration(asset_repo, asset_id: str) -> float:
+    """安全读取素材时长，任何异常返回 0。"""
+    try:
+        asset = asset_repo.get(asset_id)
+        if asset is None:
+            return 0.0
+        return float(getattr(asset, "duration", 0.0) or 0.0)
+    except Exception:
+        logger.warning("读取素材时长失败: asset_id=%s", asset_id, exc_info=True)
+        return 0.0
@@ -0,0 +1,264 @@
+"""原子片段级选片核心 — #1970 智能剪辑流程重构 P1.
+
+选片单元从"整条素材 + 随机起点"升级为"原子片段（atom clip）"：
+
+- 每个 EditPlanClip 指向一个 atom_clip_id（含 asset_id + start/end）；
+- 同一素材的不同原子片段可被同一视频多次选用；
+- 同一原子片段在一个视频内只用一次；
+- 跨变体/跨任务的避让升级为原子片段级（同 asset 的不同片段天然不重叠）；
+- atom_clips 未就绪（老素材/切片失败）时由调用方走内存兜底切片，
+  再不行回退到现有的整条素材随机起点逻辑。
+
+本模块是纯函数：原子片段数据由调用方从 repository 读取后注入，不直接碰 DB，
+便于单元测试。评分维度与 smart_match 保持一致（质量分、时长适配、新鲜度、
+未使用加分），只是评分对象从素材变为原子片段。
+"""
+
+from __future__ import annotations
+
+import random
+from dataclasses import dataclass
+from typing import Any
+
+from packages.domain.asset_atom_clip import AssetAtomClip
+
+
+@dataclass(slots=True)
+class ScoredAtomClip:
+    """带评分的候选原子片段。"""
+
+    clip: AssetAtomClip
+    score: float
+
+    @property
+    def atom_clip_id(self) -> str:
+        return self.clip.id
+
+    @property
+    def asset_id(self) -> str:
+        return self.clip.asset_id
+
+    @property
+    def start_time(self) -> float:
+        return self.clip.start_time
+
+    @property
+    def end_time(self) -> float:
+        return self.clip.end_time
+
+    @property
+    def duration(self) -> float:
+        return self.clip.duration
+
+
+# 评分权重（与 smart_match.score_asset 的维度对齐）
+W_QUALITY = 0.35
+W_DURATION_FIT = 0.30
+W_FRESHNESS = 0.15
+W_UNUSED_BONUS = 0.10
+W_ASSET_BALANCE = 0.10
+
+# 评分随机噪声上限（与 SCORE_RANDOM_NOISE_MAX 同量级，避免反复选同一组合）
+SCORE_NOISE_MAX = 0.05
+
+
+def score_atom_clip(
+    clip: AssetAtomClip,
+    *,
+    target_duration: float,
+    asset_quality: dict[str, float] | None = None,
+    asset_freshness: dict[str, float] | None = None,
+    used_in_video: set[str] | None = None,
+    asset_usage_counts: dict[str, int] | None = None,
+    recently_used: set[str] | None = None,
+    required_count: int = 1,
+    total_candidates: int = 1,
+) -> float:
+    """评估单个原子片段对某个目标槽位的适配分（越高越优先）.
+
+    评分维度：
+    - 质量分（继承素材质量，缺省中性 0.6）；
+    - 时长适配（片段时长越接近目标越好，覆盖不满显著扣分）；
+    - 新鲜度（缺省中性 0.5）；
+    - 未使用加分（本视频内未用过 +1，已用 0）；
+    - 素材均衡（同一素材在本视频用得越多，其剩余片段扣分越多，鼓励分散到多素材）；
+    - 跨视频/历史使用降权（recently_used 中的片段扣分，不硬禁）。
+    """
+    asset_quality = asset_quality or {}
+    asset_freshness = asset_freshness or {}
+    used_in_video = used_in_video or set()
+    asset_usage_counts = asset_usage_counts or {}
+    recently_used = recently_used or set()
+
+    quality = asset_quality.get(clip.asset_id, 0.6)
+
+    if target_duration > 0:
+        coverage = min(1.0, clip.duration / target_duration)
+        overshoot = max(0.0, (clip.duration - target_duration) / target_duration)
+        duration_fit = max(0.0, coverage - 0.15 * overshoot)
+    else:
+        duration_fit = 0.5
+
+    freshness = asset_freshness.get(clip.asset_id, 0.5)
+    unused_bonus = 0.0 if clip.id in used_in_video else 1.0
+
+    # 素材均衡：该素材已被本视频选用 k 次，其片段逐次扣分
+    times_used = asset_usage_counts.get(clip.asset_id, 0)
+    balance = 1.0 / (1.0 + times_used)
+
+    # 跨视频/历史使用降权（不硬禁）
+    history_penalty = 0.35 if clip.id in recently_used else 0.0
+
+    score = (
+        W_QUALITY * quality
+        + W_DURATION_FIT * duration_fit
+        + W_FRESHNESS * freshness
+        + W_UNUSED_BONUS * unused_bonus
+        + W_ASSET_BALANCE * balance
+        - history_penalty
+    )
+    return score
+
+
+def select_atom_clips(
+    candidates: list[AssetAtomClip],
+    *,
+    target_duration: float = 0.0,
+    used_atom_clip_ids: set[str] | None = None,
+    asset_usage_counts: dict[str, int] | None = None,
+    recently_used_atom_ids: set[str] | None = None,
+    required_count: int = 1,
+    limit: int = 0,
+    asset_quality: dict[str, float] | None = None,
+    asset_freshness: dict[str, float] | None = None,
+    rng: random.Random | None = None,
+) -> list[ScoredAtomClip]:
+    """为一个目标槽位从候选原子片段中评分选片（纯函数）.
+
+    Args:
+        candidates: 候选原子片段（可跨多素材）。
+        target_duration: 槽位目标时长（秒）。
+        used_atom_clip_ids: 本视频已用过的原子片段 ID（硬排除，同片段不重复）。
+        asset_usage_counts: 本视频各素材已选片段数（均衡评分用）。
+        recently_used_atom_ids: 跨视频/历史成片用过的片段 ID（降权，不硬禁）。
+        required_count: 整个视频需要的片段总数（预留，供覆盖策略判断）。
+        limit: 最多返回条数；<=0 表示返回全部排序结果。
+        asset_quality / asset_freshness: 评分注入。
+        rng: 可选随机源（测试注入）。
+
+    Returns:
+        评分降序的 ScoredAtomClip 列表（已排除本视频用过的片段）。
+    """
+    rng = rng or random.Random()
+    used = used_atom_clip_ids or set()
+    asset_usage_counts = asset_usage_counts or {}
+    recently_used = recently_used_atom_ids or set()
+
+    available = [c for c in candidates if c.id not in used]
+    scored: list[ScoredAtomClip] = []
+    for clip in available:
+        base = score_atom_clip(
+            clip,
+            target_duration=target_duration,
+            asset_quality=asset_quality,
+            asset_freshness=asset_freshness,
+            used_in_video=used,
+            asset_usage_counts=asset_usage_counts,
+            recently_used=recently_used,
+            required_count=required_count,
+            total_candidates=len(candidates),
+        )
+        noise = rng.uniform(0.0, SCORE_NOISE_MAX)
+        scored.append(ScoredAtomClip(clip=clip, score=base + noise))
+
+    scored.sort(key=lambda s: s.score, reverse=True)
+    if limit and limit > 0:
+        return scored[:limit]
+    return scored
+
+
+def clips_to_segments(clips: list[AssetAtomClip]) -> dict[str, list[tuple[float, float]]]:
+    """把选中的原子片段转换为旧的 {asset_id: [(start, end), ...]} 区间结构.
+
+    用于与现有跨变体区间避让（variant_plan_selector / metadata.used_segments）对接。
+    原子片段级天然不重叠，同素材多片段直接形成多段不重叠区间。
+    """
+    segments: dict[str, list[tuple[float, float]]] = {}
+    for clip in clips:
+        segments.setdefault(clip.asset_id, []).append((clip.start_time, clip.end_time))
+    for asset_id in segments:
+        segments[asset_id].sort()
+    return segments
+
+
+def estimate_required_clip_count(
+    voice_total_duration: float,
+    average_clip_duration: float = 4.5,
+) -> int:
+    """配音总时长 / 平均片段时长 ≈ 需要的片段数（至少 1）。"""
+    if voice_total_duration <= 0 or average_clip_duration <= 0:
+        return 1
+    return max(1, round(voice_total_duration / average_clip_duration))
+
+
+def reselect_clips_from_atoms(
+    source_clips: list[dict[str, Any]],
+    candidates: list[AssetAtomClip],
+    *,
+    historical_atom_ids: set[str] | None = None,
+    batch_used_atom_ids: set[str] | None = None,
+    rng: random.Random | None = None,
+) -> list[dict[str, Any]] | None:
+    """#1970 变体重选的原子片段级实现.
+
+    与 variant_plan_selector.reselect_clips_for_variant 对应：保留源 plan 的
+    片段骨架（order/clip_type/文案/转场），从候选原子片段中为每个 main 片段
+    选取一个原子片段；同变体/批次内同一片段不可重复，历史成片用过的片段降权。
+
+    Returns:
+        新 clips_data（dict 列表，含 asset_id/atom_clip_id/start_time/duration），
+        候选不足（main 片段多于去重后片段数）时返回 None，由调用方回退整条素材路径。
+        非 main 片段（intro/outro 等）原样保留不分配素材。
+    """
+    if not source_clips or not candidates:
+        return None
+
+    rng = rng or random.Random()
+    main_indexes = [i for i, c in enumerate(source_clips) if c.get("clip_type", "main") == "main"]
+    if len(main_indexes) > len({c.id for c in candidates}):
+        return None
+
+    used: set[str] = set(batch_used_atom_ids or ())
+    result: list[dict[str, Any]] = [dict(c) for c in source_clips]
+    asset_usage: dict[str, int] = {}
+
+    for idx in main_indexes:
+        skeleton = source_clips[idx]
+        target_duration = float(skeleton.get("duration") or 0.0)
+        ranked = select_atom_clips(
+            candidates,
+            target_duration=target_duration,
+            used_atom_clip_ids=used,
+            asset_usage_counts=asset_usage,
+            recently_used_atom_ids=historical_atom_ids or set(),
+            required_count=len(main_indexes),
+            limit=1,
+            rng=rng,
+        )
+        if not ranked:
+            return None
+        picked = ranked[0]
+        # 段长：片段短于槽位时取片段全长（渲染末帧冻结铺满），长于槽位时按槽位时长 trim
+        new_duration = picked.duration if target_duration <= 0 else min(target_duration, picked.duration)
+        result[idx].update(
+            {
+                "asset_id": picked.asset_id,
+                "atom_clip_id": picked.atom_clip_id,
+                "start_time": round(picked.start_time, 3),
+                "duration": round(new_duration, 3),
+            }
+        )
+        used.add(picked.atom_clip_id)
+        asset_usage[picked.asset_id] = asset_usage.get(picked.asset_id, 0) + 1
+
+    return result
@@ -0,0 +1,215 @@
+"""素材原子切片服务 — #1970 智能剪辑流程重构 P1.
+
+切片规则（见 docs/smart-edit-flow-redesign-20260916.md §1）：
+- 3~6 秒一个片段，具体时长在此范围内随机（避免固定节奏）
+- 切点附近 0.5 秒内有 scdet 镜头切换点时，切点偏移到切换处
+  （复用素材 metadata 中已缓存的 scene_change_points，不重新计算）
+- <6 秒素材整条作为一个片段，不切
+- 最后一个片段不足 3 秒的合并到前一个；超过 3 秒独立成段
+- 片段是逻辑索引，不物理切割视频文件
+
+片段在内存中计算；持久化由上层调用 repository 完成，保证本模块可单测、无 IO 依赖。
+"""
+
+from __future__ import annotations
+
+import random
+
+from packages.domain.asset_atom_clip import AssetAtomClip
+
+# 切片参数（集中常量，便于后续抽配置）
+MIN_CLIP_SECONDS = 3.0
+MAX_CLIP_SECONDS = 6.0
+# 切点与 scdet 切换点的对齐窗口
+SCENE_SNAP_WINDOW = 0.5
+# 末段最小独立时长：不足则并入前一段
+MIN_TAIL_SECONDS = 3.0
+# 浮点比较容差
+_EPS = 0.05
+
+
+def _round3(value: float) -> float:
+    return round(float(value), 3)
+
+
+def _snap_to_scene(
+    cut: float,
+    scene_points: list[float] | None,
+    lower: float,
+    upper: float,
+) -> tuple[float, float | None]:
+    """将切点 ``cut`` 对齐到窗口内最近的 scdet 切换点.
+
+    Args:
+        cut: 原始切点（秒）。
+        scene_points: 候选切换点（秒，已排序），可为空。
+        lower: 允许偏移的下界（不早于当前片段起点）。
+        upper: 允许偏移的上界（不晚于素材总时长）。
+
+    Returns:
+        (对齐后的切点, 命中的切换点)；未命中返回 (cut, None)。
+    """
+    if not scene_points:
+        return cut, None
+
+    best: float | None = None
+    best_dist = SCENE_SNAP_WINDOW
+    for point in scene_points:
+        # 切换点必须严格落在片段内部（不能与边界重合），且在窗口内
+        if point <= lower + _EPS or point >= upper - _EPS:
+            continue
+        dist = abs(point - cut)
+        if dist <= best_dist:
+            best_dist = dist
+            best = point
+    if best is None:
+        return cut, None
+    return _round3(best), _round3(best)
+
+
+def compute_atom_clips(
+    asset_id: str,
+    duration: float,
+    *,
+    scene_change_points: list[float] | None = None,
+    tags: list[str] | None = None,
+    rng: random.Random | None = None,
+) -> list[AssetAtomClip]:
+    """根据素材时长计算原子片段（纯函数，不落库）.
+
+    Args:
+        asset_id: 素材 ID。
+        duration: 素材总时长（秒）。
+        scene_change_points: metadata 中缓存的 scdet 切换点（秒）。
+        tags: 继承自素材的标签。
+        rng: 可选随机源（测试可注入固定种子）。
+
+    Returns:
+        有序的原子片段列表（clip_index 从 0 开始）。
+    """
+    if duration <= 0:
+        return []
+
+    r = rng or random.Random()
+    points = _normalize_scene_points(scene_change_points, duration)
+
+    # <6 秒素材整条作为一个片段，不切
+    if duration < MAX_CLIP_SECONDS:
+        return [
+            AssetAtomClip.create(
+                asset_id=asset_id,
+                start_time=0.0,
+                end_time=_round3(duration),
+                clip_index=0,
+                tags=list(tags or []),
+            )
+        ]
+
+    boundaries: list[float] = [0.0]
+    scene_hits: dict[int, float] = {}
+
+    cursor = 0.0
+    while duration - cursor > MAX_CLIP_SECONDS + _EPS:
+        # 在 [3, 6] 内随机决定本段目标时长
+        target_len = r.uniform(MIN_CLIP_SECONDS, MAX_CLIP_SECONDS)
+        raw_cut = cursor + target_len
+        if raw_cut >= duration - _EPS:
+            break
+        cut, hit = _snap_to_scene(raw_cut, points, lower=cursor, upper=duration)
+
+        # 对齐后若导致本段短于 3 秒（切换点太靠近段首），放弃对齐
+        if cut - cursor < MIN_CLIP_SECONDS - _EPS:
+            cut = _round3(raw_cut)
+            hit = None
+
+        boundaries.append(_round3(cut))
+        if hit is not None:
+            scene_hits[len(boundaries) - 1] = hit
+        cursor = cut
+
+    boundaries.append(_round3(duration))
+
+    # 末段处理：最后一个片段不足 3 秒则合并到前一个
+    if len(boundaries) >= 3:
+        tail_start = boundaries[-2]
+        tail_len = duration - tail_start
+        if tail_len < MIN_TAIL_SECONDS - _EPS:
+            boundaries.pop(-2)
+
+    clips: list[AssetAtomClip] = []
+    for index in range(len(boundaries) - 1):
+        start = boundaries[index]
+        end = boundaries[index + 1]
+        if end - start < _EPS:
+            continue
+        # 片段尾部对齐的切换点 = 该片段右边界（若它来自 snap）
+        scene_at = scene_hits.get(index + 1)
+        clips.append(
+            AssetAtomClip.create(
+                asset_id=asset_id,
+                start_time=start,
+                end_time=end,
+                clip_index=index,
+                tags=list(tags or []),
+                scene_change_at=scene_at,
+            )
+        )
+    return clips
+
+
+def compute_fallback_clips(
+    asset_id: str,
+    duration: float,
+    *,
+    tags: list[str] | None = None,
+    clip_seconds: float = 4.5,
+) -> list[AssetAtomClip]:
+    """兜底切片：atom_clips 未就绪时，内存中按固定步长临时均匀切片（不存库）.
+
+    与 :func:`compute_atom_clips` 的区别：不随机、不对齐切点，
+    产出的片段标记 ``is_fallback=True``。
+    """
+    if duration <= 0:
+        return []
+
+    step = min(max(clip_seconds, MIN_CLIP_SECONDS), MAX_CLIP_SECONDS)
+    clips: list[AssetAtomClip] = []
+    cursor = 0.0
+    index = 0
+    while cursor < duration - _EPS:
+        end = min(cursor + step, duration)
+        clips.append(
+            AssetAtomClip.create(
+                asset_id=asset_id,
+                start_time=_round3(cursor),
+                end_time=_round3(end),
+                clip_index=index,
+                tags=list(tags or []),
+                is_fallback=True,
+            )
+        )
+        cursor = end
+        index += 1
+
+    # 末段不足 3 秒合并
+    if len(clips) >= 2 and clips[-1].duration < MIN_TAIL_SECONDS - _EPS:
+        last = clips.pop()
+        prev = clips[-1]
+        merged = AssetAtomClip.create(
+            asset_id=asset_id,
+            start_time=prev.start_time,
+            end_time=last.end_time,
+            clip_index=prev.clip_index,
+            tags=list(tags or []),
+            is_fallback=True,
+        )
+        clips[-1] = merged
+    return clips
+
+
+def _normalize_scene_points(points: list[float] | None, duration: float) -> list[float]:
+    """清洗切换点：去重、排序、限定在 (0, duration) 内。"""
+    if not points:
+        return []
+    cleaned = sorted({round(float(p), 3) for p in points if 0 < float(p) < duration})
+    return cleaned
@@ -65,6 +65,7 @@ class EditPlanClip:
    order: int
    template_clip_config_id: str = ""
    asset_id: str = ""
+    atom_clip_id: str = ""
    text_content: str = ""
    start_time: float = 0.0
    duration: float = 0.0
@@ -85,6 +86,7 @@ class EditPlanClip:
        *,
        template_clip_config_id: str = "",
        asset_id: str = "",
+        atom_clip_id: str = "",
        text_content: str = "",
        start_time: float = 0.0,
        duration: float = 0.0,
@@ -117,6 +119,7 @@ class EditPlanClip:
            order=order,
            template_clip_config_id=template_clip_config_id.strip() if template_clip_config_id else "",
            asset_id=asset_id.strip() if asset_id else "",
+            atom_clip_id=atom_clip_id.strip() if atom_clip_id else "",
            text_content=text_content.strip(),
            start_time=start_time,
            duration=duration,
@@ -127,16 +130,25 @@ class EditPlanClip:
            config=config or {},
        )

-    def assign_asset(self, asset_id: str, *, start_time: float | None = None) -> None:
+    def assign_asset(
+        self,
+        asset_id: str,
+        *,
+        start_time: float | None = None,
+        atom_clip_id: str | None = None,
+    ) -> None:
        """分配素材

        Args:
            asset_id: 素材 ID
            start_time: 可选，素材播放起始时间（秒）。如果提供且在有效范围内，则设置；否则保持默认 0.0
+            atom_clip_id: 可选，选中的原子片段 ID（#1970 原子化切片）。
        """
        if not asset_id.strip():
            raise ValueError("asset_id 不能为空")
        self.asset_id = asset_id.strip()
+        if atom_clip_id is not None:
+            self.atom_clip_id = atom_clip_id.strip() if atom_clip_id else ""
        if start_time is not None and start_time >= 0:
            self.start_time = start_time
        self.updated_at = datetime.now(UTC)
@@ -12,9 +12,21 @@ else:


 class EditingMode(StrEnum):
-    """剪辑模式枚举"""
+    """剪辑模式枚举。

-    ONE_TAKE = "one_take"  # 顺序拼接模式
-    PIP = "pip"  # 画中画模式
-    VOICE_OVER = "voice_over"  # 口播+B-roll模式
-    VOICE_PIP = "voice_pip"  # 口播+画中画组合模式
+    #1970 智能剪辑流程重构（2026-09）后，剪辑组装模式改由
+    ``CreateGenerationTaskRequest.assembly_mode``（'random'/'narrative'）表达。
+    本枚举仅保留模板体系仍在使用的模式；以下三个模式标记 deprecated，
+    不主动删除代码（pip/voice_pip 在路由入口已统一映射为 one_take），
+    待确认无存量引用后在技术债清理中移除：
+
+    - ONE_TAKE（deprecated）：顺序拼接，等同 assembly_mode='random'
+    - PIP（deprecated）：画中画已下线，入口映射 one_take
+    - VOICE_PIP（deprecated）：口播+画中画已下线，入口映射 one_take
+    - VOICE_OVER：保留，口播+B-roll 模板仍在使用
+    """
+
+    ONE_TAKE = "one_take"  # deprecated（#1970）：顺序拼接，等同 assembly_mode='random'
+    PIP = "pip"  # deprecated（#1970）：画中画已下线，入口映射 one_take
+    VOICE_OVER = "voice_over"  # 口播+B-roll模式（保留）
+    VOICE_PIP = "voice_pip"  # deprecated（#1970）：口播+画中画已下线，入口映射 one_take
@@ -0,0 +1,132 @@
+"""叙事剪辑素材标签匹配 — #1970 PR3.
+
+叙事模式下，选片在现有评分（smart_match / atom_clip_selector）之前先做一层
+文案标签匹配：
+
+- 文案 tags 与素材 tag 名归一化后求交集；
+- 命中任一标签的素材作为「优先候选池」，未命中的作为普通池；
+- 调用方对优先池跑现有 smart_select_assets，数量不足时用普通池补足
+  （无任何匹配 → 完全降级为现有随机逻辑，行为与改造前一致）。
+
+纯函数模块：标签 id→名称映射由调用方查 TagModel 后注入，不直接碰 DB。
+"""
+
+from __future__ import annotations
+
+from typing import Any, Iterable
+
+# 标签归一化后仍短于此长度的标签不参与匹配（避免「的」「是」这类噪声短词）
+MIN_TAG_LEN = 2
+
+
+def normalize_tag(tag: Any) -> str:
+    """标签归一化：去空白、小写。数字/英文统一小写，中文不受影响。"""
+    if tag is None:
+        return ""
+    return str(tag).strip().lower()
+
+
+def _normalize_tags(tags: Iterable[Any]) -> set[str]:
+    out: set[str] = set()
+    for t in tags or []:
+        norm = normalize_tag(t)
+        if len(norm) >= MIN_TAG_LEN:
+            out.add(norm)
+    return out
+
+
+def build_asset_tag_name_index(tag_names_by_id: dict[str, Any]) -> dict[str, set[str]]:
+    """构造 asset_id → 归一化标签名集合 的索引。
+
+    Args:
+        tag_names_by_id: {asset_id: [标签名或标签id, ...]}，允许混入 None/空值
+    """
+    index: dict[str, set[str]] = {}
+    for asset_id, names in (tag_names_by_id or {}).items():
+        index[asset_id] = _normalize_tags(names)
+    return index
+
+
+def match_assets_by_script_tags(
+    assets: list[Any],
+    *,
+    script_tags: Iterable[Any],
+    tag_names_by_id: dict[str, Any] | None = None,
+) -> tuple[list[Any], list[Any]]:
+    """按文案标签把素材拆成「命中池 / 未命中池」，保持输入相对顺序。
+
+    Args:
+        assets: 候选素材（domain Asset，需有 id 与 tag_ids）。
+        script_tags: 文案 tags（字符串数组，名称语义）。
+        tag_names_by_id: asset_id → 素材标签名列表；素材只有 tag_ids 时由调用方
+            查 TagModel 名称后传入。为空则视为无素材命中。
+
+    Returns:
+        (matched, unmatched)：命中任一文案标签的素材 / 其余素材。
+        文案无有效标签时 matched 为空（调用方直接走随机逻辑）。
+    """
+    wanted = _normalize_tags(script_tags)
+    if not wanted:
+        return [], list(assets)
+
+    name_index = build_asset_tag_name_index(tag_names_by_id or {})
+    matched: list[Any] = []
+    unmatched: list[Any] = []
+    for asset in assets:
+        asset_id = str(getattr(asset, "id", "") or "")
+        names = set(name_index.get(asset_id, set()))
+        # 兼容素材自身带字符串 tags（旧链路/测试替身）
+        raw_tags = getattr(asset, "tags", None)
+        if raw_tags:
+            names |= _normalize_tags(raw_tags)
+        if names & wanted:
+            matched.append(asset)
+        else:
+            unmatched.append(asset)
+    return matched, unmatched
+
+
+def pick_narrative_assets(
+    assets: list[Any],
+    *,
+    script_tags: Iterable[Any],
+    tag_names_by_id: dict[str, Any] | None = None,
+    limit: int | None = None,
+    rng: Any = None,
+) -> list[Any]:
+    """叙事模式选片：标签命中池优先，不足部分从未命中池按现有评分补齐。
+
+    本函数只负责「标签优先 + 兜底降级」的顺序编排；评分仍复用
+    smart_match.smart_select_assets（质量/时长/新鲜度/未使用 + 随机噪声），
+    不重写评分维度。
+
+    Args:
+        assets: ready 视频素材候选（调用方负责状态/类型过滤）。
+        script_tags / tag_names_by_id: 见 match_assets_by_script_tags。
+        limit: 需要的素材数量；None 表示全部（命中池 + 全部未命中池）。
+        rng: 注入 smart_select_assets 的随机源（可复现）。
+
+    Returns:
+        选中的素材列表。无任何标签命中时等价于对全量跑 smart_select_assets。
+    """
+    from packages.domain.smart_match import smart_select_assets
+
+    matched, unmatched = match_assets_by_script_tags(
+        assets,
+        script_tags=script_tags,
+        tag_names_by_id=tag_names_by_id,
+    )
+
+    need = limit if (limit is not None and limit > 0) else None
+
+    if not matched:
+        # 完全降级：与改造前随机混剪同一逻辑
+        return [r.asset for r in smart_select_assets(assets, kind="video", limit=need, rng=rng)]
+
+    picked = [r.asset for r in smart_select_assets(matched, kind="video", limit=need, rng=rng)]
+    if need is not None and len(picked) < need and unmatched:
+        rest_need = need - len(picked)
+        picked.extend(r.asset for r in smart_select_assets(unmatched, kind="video", limit=rest_need, rng=rng))
+    elif need is None:
+        picked.extend(r.asset for r in smart_select_assets(unmatched, kind="video", rng=rng))
+    return picked
@@ -0,0 +1,57 @@
+"""素材原子片段仓储接口定义。"""
+
+from abc import ABC, abstractmethod
+
+from packages.domain.asset_atom_clip import AssetAtomClip
+
+
+class AssetAtomClipRepository(ABC):
+    @abstractmethod
+    def create(self, clip: AssetAtomClip) -> AssetAtomClip:
+        """创建一条原子片段记录。"""
+        pass
+
+    @abstractmethod
+    def batch_create(self, clips: list[AssetAtomClip]) -> list[AssetAtomClip]:
+        """批量创建原子片段记录。"""
+        pass
+
+    @abstractmethod
+    def find_by_asset(self, asset_id: str) -> list[AssetAtomClip]:
+        """查找某个素材的所有原子片段，按 clip_index 排序。"""
+        pass
+
+    @abstractmethod
+    def find_by_id(self, clip_id: str) -> AssetAtomClip | None:
+        """按 ID 查找单个原子片段。"""
+        pass
+
+    @abstractmethod
+    def find_by_ids(self, clip_ids: list[str]) -> list[AssetAtomClip]:
+        """批量查找原子片段。"""
+        pass
+
+    @abstractmethod
+    def delete_by_asset(self, asset_id: str) -> int:
+        """删除某素材的所有原子片段（级联删除），返回删除数量。"""
+        pass
+
+    @abstractmethod
+    def count_by_asset(self, asset_id: str) -> int:
+        """统计某素材的原子片段数量。"""
+        pass
+
+    @abstractmethod
+    def find_candidates_for_selection(
+        self,
+        asset_ids: list[str],
+        *,
+        min_duration: float | None = None,
+        max_duration: float | None = None,
+        limit: int = 100,
+    ) -> list[AssetAtomClip]:
+        """按素材集合和时长条件查找候选原子片段，按 clip_index 排序。
+
+        选片逻辑一次拉取多条素材的候选片段时使用，避免 N+1 查询。
+        """
+        pass
@@ -0,0 +1,218 @@
+"""#1970 PR3 schema 校验 + 路由辅助函数测试。"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+from app.api.routes import generation_tasks as gt
+from app.schemas.generation_task import CreateGenerationTaskRequest
+from pydantic import ValidationError
+
+# ── schema ─────────────────────────────────────────────────────────────────
+
+
+def _base_payload(**overrides):
+    payload = dict(
+        template_id="tpl1",
+        asset_ids=["a1", "a2"],
+        duration=30,
+        title_text="t",
+        editing_mode="voice_over",
+    )
+    payload.update(overrides)
+    return payload
+
+
+class TestAssemblySchema:
+    def test_defaults(self):
+        req = CreateGenerationTaskRequest(**_base_payload())
+        assert req.assembly_mode == "random"
+        assert req.script_id == ""
+        assert req.tts_voice_id == ""
+        assert req.tts_voice_source == "preset"
+        assert req.video_ratio == ""  # 空串=沿用模板默认（前端新流程显式传 9:16）
+        assert req.dedup_enabled is True
+
+    def test_narrative_accepts_fields(self):
+        req = CreateGenerationTaskRequest(
+            **_base_payload(
+                assembly_mode="narrative",
+                script_id="s1",
+                tts_voice_id="longxiaochun",
+                tts_voice_source="clone",
+                video_ratio="16:9",
+            )
+        )
+        assert req.assembly_mode == "narrative"
+        assert req.script_id == "s1"
+
+    def test_bad_assembly_mode_rejected(self):
+        with pytest.raises(ValidationError):
+            CreateGenerationTaskRequest(**_base_payload(assembly_mode="movie"))
+
+    def test_bad_voice_source_rejected(self):
+        with pytest.raises(ValidationError):
+            CreateGenerationTaskRequest(**_base_payload(tts_voice_source="elevenlabs"))
+
+    def test_bad_video_ratio_rejected(self):
+        with pytest.raises(ValidationError):
+            CreateGenerationTaskRequest(**_base_payload(video_ratio="4:5"))
+
+    def test_narrative_without_script_rejected(self):
+        with pytest.raises(ValidationError) as ei:
+            CreateGenerationTaskRequest(**_base_payload(assembly_mode="narrative"))
+        assert "script_id" in str(ei.value)
+
+    def test_narrative_without_voice_rejected(self):
+        with pytest.raises(ValidationError) as ei:
+            CreateGenerationTaskRequest(**_base_payload(assembly_mode="narrative", script_id="s1"))
+        assert "tts_voice_id" in str(ei.value)
+
+    def test_random_mode_ignores_script_absence(self):
+        req = CreateGenerationTaskRequest(**_base_payload())
+        assert req.assembly_mode == "random"
+
+
+# ── _select_assets_from_library 的叙事分支 ─────────────────────────────────
+
+
+@dataclass
+class _Asset:
+    id: str
+    status: object = field(default_factory=lambda: SimpleNamespace(value="ready"))
+    mime_type: str = "video/mp4"
+    tags: list[str] = field(default_factory=list)
+    tag_ids: list[str] = field(default_factory=list)
+    file_type: str = "video"
+    quality_score: float | None = None
+    duration: float = 8.0
+    created_at: object = None
+    metadata: dict = field(default_factory=dict)
+
+
+class TestNarrativeSelectInRoute:
+    def test_narrative_tags_prioritize_matched(self):
+        assets = [
+            _Asset("a1", tags=["工厂"]),
+            _Asset("a2", tags=["旅游"]),
+            _Asset("a3", tags=["工厂"]),
+        ]
+        picked = gt._select_assets_from_library(assets, mode="all", count=2, script_tags=["工厂"])
+        assert set(picked) == {"a1", "a3"}
+
+    def test_narrative_no_match_falls_back_to_full_pool(self):
+        assets = [_Asset("a1", tags=["工厂"]), _Asset("a2", tags=["旅游"])]
+        picked = gt._select_assets_from_library(assets, mode="all", count=2, script_tags=["美食"])
+        assert set(picked) == {"a1", "a2"}
+
+    def test_tag_ids_via_index(self):
+        assets = [_Asset("a1", tag_ids=["t1"]), _Asset("a2", tag_ids=["t2"])]
+        picked = gt._select_assets_from_library(
+            assets,
+            mode="all",
+            count=1,
+            script_tags=["教程"],
+            tag_names_by_id={"a1": ["教程"], "a2": ["旅游"]},
+        )
+        assert picked == ["a1"]
+
+    def test_no_script_tags_smart_path_unchanged(self):
+        assets = [_Asset("a1"), _Asset("a2")]
+        picked = gt._select_assets_from_library(assets, mode="smart", count=1)
+        assert picked  # 非空即可，评分逻辑由 smart_match 自己的测试覆盖
+
+
+# ── _load_asset_tag_names（DB 替身） ────────────────────────────────────────
+
+
+class _FakeRow:
+    def __init__(self, **kw):
+        self.__dict__.update(kw)
+
+
+class _FakeQuery:
+    def __init__(self, rows):
+        self._rows = rows
+
+    def filter(self, *a, **k):
+        return self
+
+    def all(self):
+        return self._rows
+
+
+class _FakeDb:
+    def __init__(self, name_rows, link_rows):
+        self._maps = {
+            "names": name_rows,
+            "links": link_rows,
+        }
+
+    def query(self, *cols):
+        # _load_asset_tag_names 两次查询：第一次取 (id, name)，第二次取 (asset_id, tag_id)
+        keys = tuple(getattr(c, "key", None) for c in cols)
+        if keys and keys[0] == "id":
+            return _FakeQuery(self._maps["names"])
+        return _FakeQuery(self._maps["links"])
+
+
+@dataclass
+class _TagIdAsset:
+    id: str
+    tag_ids: list[str]
+
+
+class TestLoadAssetTagNames:
+    def test_builds_index(self):
+        assets = [_TagIdAsset("a1", ["t1", "t2"]), _TagIdAsset("a2", ["t2"])]
+        db = _FakeDb(
+            name_rows=[_FakeRow(id="t1", name="工厂"), _FakeRow(id="t2", name="带货")],
+            link_rows=[
+                ("a1", "t1"),
+                ("a1", "t2"),
+                ("a2", "t2"),
+            ],
+        )
+        idx = gt._load_asset_tag_names(db, assets, "u1")
+        assert idx == {"a1": ["工厂", "带货"], "a2": ["带货"]}
+
+    def test_no_tag_ids_returns_empty(self):
+        assert gt._load_asset_tag_names(_FakeDb([], []), [_TagIdAsset("a1", [])], "u1") == {}
+
+    def test_query_failure_degrades_empty(self):
+        class BoomQuery:
+            def filter(self, *a, **k):
+                raise RuntimeError("db down")
+
+        class BoomDb:
+            def query(self, *a):
+                return BoomQuery()
+
+        idx = gt._load_asset_tag_names(BoomDb(), [_TagIdAsset("a1", ["t1"])], "u1")
+        assert idx == {}
+
+
+# ── _resolve_output_dimensions ─────────────────────────────────────────────
+
+
+class TestResolveOutputDimensions:
+    def _req(self, ratio="", width=1280, height=720):
+        return CreateGenerationTaskRequest(**_base_payload(video_ratio=ratio, output_width=width, output_height=height))
+
+    def test_known_ratios(self):
+        assert gt._resolve_output_dimensions(self._req("9:16")) == (1080, 1920)
+        assert gt._resolve_output_dimensions(self._req("16:9")) == (1920, 1080)
+        assert gt._resolve_output_dimensions(self._req("1:1")) == (1080, 1080)
+        assert gt._resolve_output_dimensions(self._req("4:3")) == (1440, 1080)
+        assert gt._resolve_output_dimensions(self._req("3:4")) == (1080, 1440)
+
+    def test_old_call_default_kept_when_no_ratio(self):
+        assert gt._resolve_output_dimensions(self._req("")) == (1280, 720)
+
+    def test_explicit_dimensions_take_precedence(self):
+        # 非旧默认值（720p）的显式分辨率优先于 ratio 映射
+        req = self._req("9:16", width=1440, height=2560)
+        assert gt._resolve_output_dimensions(req) == (1440, 2560)
@@ -0,0 +1,110 @@
+"""#1970 原子片段 resolver 单元测试：DB 加载 + 内存兜底."""
+
+from __future__ import annotations
+
+from packages.domain.asset_atom_clip import AssetAtomClip
+from packages.domain.atom_clip_resolver import (
+    flatten_candidates,
+    load_atom_clips_for_assets,
+)
+
+
+def _atom(asset_id: str, idx: int, start: float, end: float) -> AssetAtomClip:
+    return AssetAtomClip(
+        id=f"{asset_id}-clip-{idx}",
+        asset_id=asset_id,
+        start_time=start,
+        end_time=end,
+        duration=round(end - start, 3),
+        clip_index=idx,
+    )
+
+
+class FakeAtomRepo:
+    def __init__(self, by_asset):
+        self._by_asset = by_asset
+
+    def find_candidates_for_selection(self, asset_ids, *, limit=0):
+        out = []
+        for aid in asset_ids:
+            out.extend(self._by_asset.get(aid, []))
+        return out
+
+    def find_by_asset(self, asset_id):
+        return list(self._by_asset.get(asset_id, []))
+
+
+class _Asset:
+    def __init__(self, duration):
+        self.duration = duration
+
+
+class FakeAssetRepo:
+    def __init__(self, durations):
+        self._durations = durations
+
+    def get(self, asset_id):
+        d = self._durations.get(asset_id)
+        return _Asset(d) if d is not None else None
+
+
+class TestLoadAtomClips:
+    def test_persisted_clips_loaded_sorted(self):
+        clips = [_atom("a", 1, 4.5, 9.0), _atom("a", 0, 0.0, 4.5)]
+        repo = FakeAtomRepo({"a": clips})
+        result = load_atom_clips_for_assets(["a"], atom_clip_repo=repo)
+        assert [c.clip_index for c in result["a"]] == [0, 1]
+
+    def test_dedup_asset_ids_preserves_order(self):
+        repo = FakeAtomRepo({"a": [_atom("a", 0, 0, 4)], "b": [_atom("b", 0, 0, 4)]})
+        result = load_atom_clips_for_assets(["a", "b", "a"], atom_clip_repo=repo)
+        assert list(result.keys()) == ["a", "b"]
+
+    def test_fallback_when_no_persisted_clips(self):
+        """老素材没有 atom_clips 时，内存按 3-6 秒均匀切片，标记 is_fallback。"""
+        atom_repo = FakeAtomRepo({})
+        asset_repo = FakeAssetRepo({"old": 20.0})
+        result = load_atom_clips_for_assets(["old"], atom_clip_repo=atom_repo, asset_repo=asset_repo)
+        assert "old" in result
+        clips = result["old"]
+        assert clips
+        assert all(c.is_fallback for c in clips)
+        assert abs(clips[-1].end_time - 20.0) < 0.01
+
+    def test_missing_duration_skipped(self):
+        atom_repo = FakeAtomRepo({})
+        asset_repo = FakeAssetRepo({})
+        result = load_atom_clips_for_assets(["ghost"], atom_clip_repo=atom_repo, asset_repo=asset_repo)
+        assert result == {}
+
+    def test_no_asset_repo_skips_empty_assets(self):
+        atom_repo = FakeAtomRepo({})
+        result = load_atom_clips_for_assets(["a"], atom_clip_repo=atom_repo, asset_repo=None)
+        assert result == {}
+
+    def test_mixed_persisted_and_fallback(self):
+        atom_repo = FakeAtomRepo({"new": [_atom("new", 0, 0, 5)]})
+        asset_repo = FakeAssetRepo({"new": 5.0, "old": 10.0})
+        result = load_atom_clips_for_assets(["new", "old"], atom_clip_repo=atom_repo, asset_repo=asset_repo)
+        assert not result["new"][0].is_fallback
+        assert all(c.is_fallback for c in result["old"])
+
+    def test_repo_exception_falls_back(self):
+        class BrokenRepo(FakeAtomRepo):
+            def find_candidates_for_selection(self, asset_ids, *, limit=0):
+                raise RuntimeError("db down")
+
+        asset_repo = FakeAssetRepo({"a": 9.0})
+        result = load_atom_clips_for_assets(["a"], atom_clip_repo=BrokenRepo({}), asset_repo=asset_repo)
+        assert result["a"]
+        assert all(c.is_fallback for c in result["a"])
+
+    def test_empty_input(self):
+        assert load_atom_clips_for_assets([], atom_clip_repo=FakeAtomRepo({})) == {}
+
+
+class TestFlatten:
+    def test_flatten_order(self):
+        clips = flatten_candidates({"a": [_atom("a", 0, 0, 4)], "b": [_atom("b", 0, 0, 4), _atom("b", 1, 4, 8)]})
+        assert len(clips) == 3
+        assert clips[0].asset_id == "a"
@@ -0,0 +1,205 @@
+"""#1970 原子片段级选片核心单元测试（纯函数，不依赖 DB）."""
+
+from __future__ import annotations
+
+import random
+
+from packages.domain.asset_atom_clip import AssetAtomClip
+from packages.domain.atom_clip_selector import (
+    clips_to_segments,
+    estimate_required_clip_count,
+    reselect_clips_from_atoms,
+    score_atom_clip,
+    select_atom_clips,
+)
+from packages.domain.atom_clip_service import compute_atom_clips
+
+
+def _clip(asset_id: str, start: float, end: float, clip_id: str = "") -> AssetAtomClip:
+    return (
+        AssetAtomClip.create(
+            asset_id=asset_id,
+            start_time=start,
+            end_time=end,
+            clip_index=int(start),
+        )
+        if not clip_id
+        else AssetAtomClip(
+            id=clip_id,
+            asset_id=asset_id,
+            start_time=start,
+            end_time=end,
+            duration=round(end - start, 3),
+            clip_index=0,
+        )
+    )
+
+
+class TestEstimateCount:
+    def test_basic(self):
+        assert estimate_required_clip_count(30.0, 4.5) == 7
+        assert estimate_required_clip_count(18.0, 4.0) == round(18 / 4)
+
+    def test_invalid_inputs_returns_one(self):
+        assert estimate_required_clip_count(0) == 1
+        assert estimate_required_clip_count(10, 0) == 1
+        assert estimate_required_clip_count(-1) == 1
+
+
+class TestScore:
+    def test_unused_beats_used(self):
+        c = _clip("a1", 0, 4)
+        s_unused = score_atom_clip(c, target_duration=4.0, used_in_video=set())
+        s_used = score_atom_clip(c, target_duration=4.0, used_in_video={c.id})
+        assert s_unused > s_used
+
+    def test_duration_fit_better_when_closer(self):
+        target = 4.0
+        exact = score_atom_clip(_clip("a", 0, 4.0), target_duration=target)
+        short = score_atom_clip(_clip("b", 0, 1.5), target_duration=target)
+        assert exact > short
+
+    def test_history_penalty(self):
+        c = _clip("a1", 0, 4)
+        normal = score_atom_clip(c, target_duration=4.0)
+        penalized = score_atom_clip(c, target_duration=4.0, recently_used={c.id})
+        assert normal > penalized
+
+    def test_asset_balance_penalizes_repeated_asset(self):
+        c1 = _clip("a", 0, 4)
+        first = score_atom_clip(c1, target_duration=4.0, asset_usage_counts={})
+        third = score_atom_clip(c1, target_duration=4.0, asset_usage_counts={"a": 2})
+        assert first > third
+
+
+class TestSelect:
+    def test_no_duplicate_atom_within_video(self):
+        pool = compute_atom_clips("a", 30.0, rng=random.Random(1))
+        used: set[str] = set()
+        usage: dict[str, int] = {}
+        chosen = []
+        rng = random.Random(5)
+        for _ in range(4):
+            ranked = select_atom_clips(
+                pool,
+                target_duration=4.0,
+                used_atom_clip_ids=used,
+                asset_usage_counts=usage,
+                required_count=4,
+                limit=1,
+                rng=rng,
+            )
+            assert ranked
+            pick = ranked[0]
+            assert pick.atom_clip_id not in used
+            chosen.append(pick)
+            used.add(pick.atom_clip_id)
+            usage[pick.asset_id] = usage.get(pick.asset_id, 0) + 1
+        assert len(used) == 4
+
+    def test_same_asset_different_clips_allowed(self):
+        pool = compute_atom_clips("a", 30.0, rng=random.Random(2))
+        used: set[str] = set()
+        usage: dict[str, int] = {}
+        rng = random.Random(7)
+        picked_assets = set()
+        for _ in range(3):
+            pick = select_atom_clips(
+                pool,
+                target_duration=4.0,
+                used_atom_clip_ids=used,
+                asset_usage_counts=usage,
+                limit=1,
+                rng=rng,
+            )[0]
+            used.add(pick.atom_clip_id)
+            usage[pick.asset_id] = usage.get(pick.asset_id, 0) + 1
+            picked_assets.add(pick.asset_id)
+        # 单素材池允许同素材多片段
+        assert picked_assets == {"a"}
+        assert len(used) == 3
+
+    def test_exhausted_pool_returns_empty(self):
+        pool = [_clip("a", 0, 4)]
+        ranked = select_atom_clips(pool, used_atom_clip_ids={pool[0].id}, target_duration=4.0)
+        assert ranked == []
+
+    def test_recently_used_deprioritized_not_hard_blocked(self):
+        # 两个片段，recent 中包含更合适的那个；它应被降权但不会从候选中消失
+        fresh = _clip("a", 0, 2.0, clip_id="fresh")
+        recent = _clip("b", 0, 4.0, clip_id="recent")
+        ranked = select_atom_clips(
+            [fresh, recent],
+            target_duration=4.0,
+            recently_used_atom_ids={"recent"},
+            limit=2,
+            rng=random.Random(0),  # 噪声 0 不影响
+        )
+        ids = [r.atom_clip_id for r in ranked]
+        assert set(ids) == {"fresh", "recent"}
+        # 降权 + 噪声可能导致排序不稳定，只验证 recent 仍在候选中（不硬禁）
+
+    def test_limit(self):
+        pool = compute_atom_clips("a", 40.0, rng=random.Random(4))
+        ranked = select_atom_clips(pool, target_duration=4.0, limit=3)
+        assert len(ranked) == 3
+        scores = [r.score for r in ranked]
+        assert scores == sorted(scores, reverse=True)
+
+
+class TestClipsToSegments:
+    def test_grouped_by_asset_sorted(self):
+        clips = [
+            _clip("a", 10, 14),
+            _clip("a", 0, 4),
+            _clip("b", 2, 6),
+        ]
+        segs = clips_to_segments(clips)
+        assert segs["a"] == [(0, 4), (10, 14)]
+        assert segs["b"] == [(2, 6)]
+
+
+class TestReselectFromAtoms:
+    def _src(self, n):
+        return [{"order": i, "clip_type": "main", "duration": 4.0, "start_time": 0.0} for i in range(n)]
+
+    def test_skeleton_preserved_and_unique(self):
+        pool = compute_atom_clips("a", 30.0, rng=random.Random(11)) + compute_atom_clips(
+            "b", 30.0, rng=random.Random(12)
+        )
+        out = reselect_clips_from_atoms(self._src(5), pool, rng=random.Random(13))
+        assert out is not None
+        assert len(out) == 5
+        ids = [c["atom_clip_id"] for c in out]
+        assert len(set(ids)) == 5
+        for c in out:
+            assert c["asset_id"]
+            assert c["start_time"] >= 0
+            assert c["duration"] > 0
+
+    def test_insufficient_candidates_returns_none(self):
+        pool = compute_atom_clips("a", 10.0, rng=random.Random(1))
+        assert reselect_clips_from_atoms(self._src(20), pool) is None
+
+    def test_non_main_clips_left_untouched(self):
+        pool = compute_atom_clips("a", 30.0, rng=random.Random(8))
+        src = [
+            {"order": 0, "clip_type": "intro", "duration": 2.0, "asset_id": "fixed"},
+            {"order": 1, "clip_type": "main", "duration": 4.0},
+        ]
+        out = reselect_clips_from_atoms(src, pool, rng=random.Random(3))
+        assert out is not None
+        assert out[0]["asset_id"] == "fixed"
+        assert "atom_clip_id" not in out[0]
+        assert out[1].get("atom_clip_id")
+
+    def test_empty_inputs(self):
+        assert reselect_clips_from_atoms([], [_clip("a", 0, 4)]) is None
+        assert reselect_clips_from_atoms(self._src(2), []) is None
+
+    def test_batch_used_excluded(self):
+        pool = compute_atom_clips("a", 30.0, rng=random.Random(21))
+        batch_used = {pool[0].id}
+        out = reselect_clips_from_atoms(self._src(3), pool, batch_used_atom_ids=batch_used, rng=random.Random(22))
+        assert out is not None
+        assert pool[0].id not in {c["atom_clip_id"] for c in out}
@@ -0,0 +1,151 @@
+"""#1970 素材原子化切片逻辑单元测试（纯函数，不依赖 DB）."""
+
+from __future__ import annotations
+
+import random
+
+import pytest
+
+from packages.domain.asset_atom_clip import AssetAtomClip
+from packages.domain.atom_clip_service import (
+    MAX_CLIP_SECONDS,
+    MIN_CLIP_SECONDS,
+    compute_atom_clips,
+    compute_fallback_clips,
+)
+
+
+class TestComputeAtomClips:
+    def test_short_asset_under_6s_single_clip(self):
+        """<6 秒素材整条作为一个片段，不切。"""
+        for dur in (0.1, 3.0, 5.99):
+            clips = compute_atom_clips("a1", dur, rng=random.Random(1))
+            assert len(clips) == 1
+            assert clips[0].start_time == 0.0
+            assert abs(clips[0].end_time - dur) < 0.01
+            assert clips[0].clip_index == 0
+
+    def test_exactly_6s_single_clip(self):
+        clips = compute_atom_clips("a1", 6.0, rng=random.Random(1))
+        assert len(clips) == 1
+        assert clips[0].start_time == 0.0
+
+    def test_zero_and_negative_duration_returns_empty(self):
+        assert compute_atom_clips("a1", 0) == []
+        assert compute_atom_clips("a1", -1.0) == []
+
+    @pytest.mark.parametrize("seed", range(30))
+    def test_clips_in_3_to_6_range(self, seed):
+        """除末段外，每段时长在 3~6 秒；末段 >=3 秒。"""
+        clips = compute_atom_clips("a1", 60.0, rng=random.Random(seed))
+        assert len(clips) >= 2
+        for clip in clips[:-1]:
+            assert MIN_CLIP_SECONDS - 0.06 <= clip.duration <= MAX_CLIP_SECONDS + 0.06
+        # 末段 >=3（不足 3 应已合并）
+        assert clips[-1].duration >= MIN_CLIP_SECONDS - 0.06
+
+    @pytest.mark.parametrize("dur", [6.01, 7.0, 9.0, 12.3, 30.0, 45.3, 100.0])
+    def test_full_coverage_no_gaps_no_overlap(self, dur):
+        clips = compute_atom_clips("a1", dur, rng=random.Random(int(dur * 100) % 10000))
+        assert abs(clips[0].start_time) < 0.001
+        assert abs(clips[-1].end_time - dur) < 0.01
+        for prev, nxt in zip(clips, clips[1:], strict=False):
+            assert abs(prev.end_time - nxt.start_time) < 0.001
+
+    def test_clip_index_sequential(self):
+        clips = compute_atom_clips("a1", 40.0, rng=random.Random(5))
+        assert [c.clip_index for c in clips] == list(range(len(clips)))
+
+    def test_tail_shorter_than_3s_merges_into_previous(self):
+        """末段不足 3 秒必须合并到前一段。"""
+        # 多跑种子，保证任何随机结果都不存在 <3s 的末段
+        for seed in range(100):
+            clips = compute_atom_clips("a1", 7.5, rng=random.Random(seed))
+            assert clips[-1].duration >= MIN_CLIP_SECONDS - 0.06
+            assert abs(clips[-1].end_time - 7.5) < 0.01
+
+    def test_tail_between_3_and_6_stands_alone(self):
+        """末段 >=3 秒独立成段。"""
+        found_standalone = False
+        for seed in range(100):
+            clips = compute_atom_clips("a1", 9.5, rng=random.Random(seed))
+            if len(clips) == 2:
+                found_standalone = True
+                assert clips[-1].duration >= MIN_CLIP_SECONDS - 0.06
+        assert found_standalone, "9.5s 至少在某些种子下应切为两段"
+
+    def test_scene_change_snap_within_window(self):
+        """切点 0.5s 窗口内有切换点时，切点对齐到切换处。"""
+        aligned = 0
+        for seed in range(500):
+            clips = compute_atom_clips("a1", 20.0, scene_change_points=[4.52], rng=random.Random(seed))
+            if any(c.scene_change_at == 4.52 for c in clips):
+                aligned += 1
+                hit = next(c for c in clips if c.scene_change_at == 4.52)
+                # 命中片段的右边界即切换点
+                assert abs(hit.end_time - 4.52) < 0.001
+        assert aligned > 0
+
+    def test_scene_change_outside_window_not_force_aligned(self):
+        """窗口外的切换点不应强行对齐。"""
+        clips = compute_atom_clips("a1", 30.0, scene_change_points=[15.0], rng=random.Random(1))
+        for c in clips:
+            if c.scene_change_at is not None:
+                assert abs(c.end_time - c.scene_change_at) < 0.001
+
+    def test_scene_snap_never_creates_sub_3s_clip(self):
+        """对齐不能导致片段短于 3 秒。"""
+        for seed in range(100):
+            clips = compute_atom_clips("a1", 40.0, scene_change_points=[3.2, 6.3, 9.4], rng=random.Random(seed))
+            for c in clips:
+                assert c.duration >= MIN_CLIP_SECONDS - 0.06
+
+    def test_scene_points_out_of_duration_ignored(self):
+        clips = compute_atom_clips("a1", 20.0, scene_change_points=[-1.0, 25.0, 4.0], rng=random.Random(3))
+        assert all(c.scene_change_at != -1.0 and c.scene_change_at != 25.0 for c in clips)
+
+    def test_tags_inherited(self):
+        clips = compute_atom_clips("a1", 30.0, tags=["t1", "t2"], rng=random.Random(2))
+        assert all(c.tags == ["t1", "t2"] for c in clips)
+
+    def test_random_not_fixed_rhythm(self):
+        """随机切片：不同种子产出的切点集合应不同（避免固定节奏）。"""
+        cuts1 = [c.end_time for c in compute_atom_clips("a1", 60.0, rng=random.Random(1))]
+        cuts2 = [c.end_time for c in compute_atom_clips("a1", 60.0, rng=random.Random(2))]
+        assert cuts1 != cuts2
+
+    def test_seed_reproducible(self):
+        """相同种子结果可复现。"""
+        a = [(c.start_time, c.end_time) for c in compute_atom_clips("a1", 60.0, rng=random.Random(42))]
+        b = [(c.start_time, c.end_time) for c in compute_atom_clips("a1", 60.0, rng=random.Random(42))]
+        assert a == b
+
+
+class TestComputeFallbackClips:
+    def test_fallback_marked_and_uniform(self):
+        clips = compute_fallback_clips("a1", 20.0, clip_seconds=4.5)
+        assert clips
+        assert all(c.is_fallback for c in clips)
+        for prev, nxt in zip(clips, clips[1:], strict=False):
+            assert abs(prev.end_time - nxt.start_time) < 0.001
+        assert abs(clips[-1].end_time - 20.0) < 0.01
+
+    def test_fallback_tail_merge(self):
+        """11.5s = 4.5+4.5+2.5 → 末段 2.5<3 合并 → 4.5+7.0。"""
+        clips = compute_fallback_clips("a1", 11.5, clip_seconds=4.5)
+        assert len(clips) == 2
+        assert abs(clips[-1].duration - 7.0) < 0.01
+
+    def test_fallback_short_asset(self):
+        clips = compute_fallback_clips("a1", 2.0)
+        assert len(clips) == 1
+        assert clips[0].is_fallback
+
+    def test_fallback_invalid_duration(self):
+        assert compute_fallback_clips("a1", 0) == []
+        assert compute_fallback_clips("a1", -5) == []
+
+    def test_fallback_clip_has_no_persisted_id(self):
+        clips = compute_fallback_clips("a1", 10.0)
+        # 兜底片段仍有运行时 id（dataclass 生成），但 is_fallback 是判别标记
+        assert all(isinstance(c, AssetAtomClip) for c in clips)
@@ -0,0 +1,181 @@
+"""#1970 PlanGeneratorService 原子片段选片端到端单元测试.
+
+用 SQLite 内存库 + 真实仓储验证：注入 atom_clip_repo 后，正式生成（非预览）
+从原子片段选片，EditPlanClip.atom_clip_id 落库；预览模式保持旧路径。
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
+os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+
+import pytest
+from app.services.plan_generator_service import PlanGeneratorService
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from packages.adapters.sqlalchemy_impl.asset_atom_clip_repository import (
+    SQLAlchemyAssetAtomClipRepository,
+)
+from packages.adapters.sqlalchemy_impl.models import Base
+from packages.domain.asset_atom_clip import AssetAtomClip
+from packages.domain.edit_template import EditTemplate, EditTemplateStatus
+from packages.domain.editing_mode import EditingMode
+from packages.domain.template_clip_config import ClipType, TemplateClipConfig
+
+
+class _FakeAsset:
+    def __init__(self, aid, duration):
+        self.id = aid
+        self.duration = duration
+        self.quality_score = 60.0
+        self.metadata = {}
+        self.created_at = None
+
+
+class FakeAssetRepo:
+    def __init__(self, durations):
+        self._durations = durations
+
+    def get(self, aid):
+        return _FakeAsset(aid, self._durations[aid]) if aid in self._durations else None
+
+
+@pytest.fixture()
+def db_session():
+    engine = create_engine("sqlite://")
+    # 只建相关表，避免全模型依赖
+    Base.metadata.create_all(
+        engine,
+        tables=[
+            Base.metadata.tables["edit_plans"],
+            Base.metadata.tables["edit_plan_clips"],
+            Base.metadata.tables["asset_atom_clips"],
+        ],
+    )
+    connection = engine.connect()
+    Session = sessionmaker(bind=connection)
+    session = Session()
+    yield session
+    session.close()
+    connection.close()
+
+
+def _template(mode=EditingMode.ONE_TAKE.value):
+    return EditTemplate(
+        id="tpl-1",
+        name="测试模板",
+        editing_mode=mode,
+        status=EditTemplateStatus.ACTIVE,
+    )
+
+
+def _clip_configs(n=3):
+    return [
+        TemplateClipConfig(
+            id=f"cfg-{i}",
+            template_id="tpl-1",
+            clip_type=ClipType.MAIN,
+            order=i,
+            min_duration=3.0,
+            max_duration=6.0,
+        )
+        for i in range(n)
+    ]
+
+
+class TestAtomClipPlanGeneration:
+    def test_generation_uses_atom_clips(self, db_session):
+        atom_repo = SQLAlchemyAssetAtomClipRepository(db_session)
+        # 两个素材各 30s，各切若干片段
+        clips_a = [AssetAtomClip.create("asset-a", i * 5.0, i * 5.0 + 5.0, i) for i in range(6)]
+        clips_b = [AssetAtomClip.create("asset-b", i * 5.0, i * 5.0 + 5.0, i) for i in range(6)]
+        atom_repo.batch_create(clips_a + clips_b)
+        db_session.commit()
+
+        svc = PlanGeneratorService(
+            db_session,
+            asset_repo=FakeAssetRepo({"asset-a": 30.0, "asset-b": 30.0}),
+            atom_clip_repo=atom_repo,
+        )
+        result = svc.generate_from_template(
+            template=_template(),
+            clip_configs=_clip_configs(3),
+            asset_ids=["asset-a", "asset-b"],
+            created_by_user_id="user-1",
+        )
+        clips = result["clips"]
+        assert len(clips) == 3
+        # 每个 clip 都绑定了原子片段
+        atom_ids = [c.atom_clip_id for c in clips]
+        assert all(atom_ids)
+        # 同一原子片段一个视频只用一次
+        assert len(set(atom_ids)) == 3
+        # start_time/duration 与选中片段一致
+        for c in clips:
+            assert c.start_time >= 0
+            assert 0 < c.duration <= 6.0 + 0.01
+        # asset_id 与 atom_clip 归属一致
+        for c in clips:
+            assert c.asset_id.startswith("asset-")
+
+    def test_fallback_when_atom_clips_not_ready(self, db_session):
+        """素材没有 atom_clips 时内存兜底切片，仍能选出片段。"""
+        atom_repo = SQLAlchemyAssetAtomClipRepository(db_session)
+        svc = PlanGeneratorService(
+            db_session,
+            asset_repo=FakeAssetRepo({"old-asset": 20.0}),
+            atom_clip_repo=atom_repo,
+        )
+        result = svc.generate_from_template(
+            template=_template(),
+            clip_configs=_clip_configs(3),
+            asset_ids=["old-asset"],
+            created_by_user_id="user-1",
+        )
+        clips = result["clips"]
+        # 兜底片段不落库、无持久 ID，clip 不绑定 atom_clip_id（回退旧路径）或绑定运行时 ID
+        # 关键：必须成功选出素材，不报错
+        assert all(c.asset_id == "old-asset" for c in clips)
+
+    def test_preview_mode_keeps_legacy_path(self, db_session):
+        """随机预览模式走旧路径，不要求 atom clips。"""
+        atom_repo = SQLAlchemyAssetAtomClipRepository(db_session)
+        svc = PlanGeneratorService(
+            db_session,
+            asset_repo=FakeAssetRepo({"asset-a": 30.0, "asset-b": 30.0, "asset-c": 30.0}),
+            atom_clip_repo=atom_repo,
+        )
+        result = svc.generate_from_template(
+            template=_template(),
+            clip_configs=_clip_configs(3),
+            asset_ids=["asset-a", "asset-b", "asset-c"],
+            created_by_user_id="user-1",
+            random_preview=True,
+        )
+        clips = result["clips"]
+        assert len(clips) == 3
+        assert {c.asset_id for c in clips} == {"asset-a", "asset-b", "asset-c"}
+        # 预览路径不绑定 atom_clip_id
+        assert all(not c.atom_clip_id for c in clips)
+
+    def test_no_atom_repo_uses_legacy_path(self, db_session):
+        """未注入 atom_clip_repo（旧调用方）时行为不变。"""
+        svc = PlanGeneratorService(
+            db_session,
+            asset_repo=FakeAssetRepo({"asset-a": 30.0, "asset-b": 30.0, "asset-c": 30.0}),
+        )
+        result = svc.generate_from_template(
+            template=_template(),
+            clip_configs=_clip_configs(3),
+            asset_ids=["asset-a", "asset-b", "asset-c"],
+            created_by_user_id="user-1",
+        )
+        clips = result["clips"]
+        assert len(clips) == 3
+        assert {c.asset_id for c in clips} == {"asset-a", "asset-b", "asset-c"}
@@ -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
@@ -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
@@ -0,0 +1,167 @@
+"""#1970 PR3 叙事模式文案标签匹配纯函数测试。"""
+
+from __future__ import annotations
+
+import random
+from dataclasses import dataclass, field
+
+import pytest
+
+from packages.domain.narrative_match import (
+    build_asset_tag_name_index,
+    match_assets_by_script_tags,
+    normalize_tag,
+    pick_narrative_assets,
+)
+
+
+@dataclass
+class FakeAsset:
+    id: str
+    tag_ids: list[str] = field(default_factory=list)
+    tags: list[str] = field(default_factory=list)
+    status: str = "ready"
+    file_type: str = "video"
+    duration: float = 10.0
+    quality_score: float | None = None
+    created_at: object = None
+    metadata: dict = field(default_factory=dict)
+
+
+# ── normalize_tag ──────────────────────────────────────────────────────────
+
+
+class TestNormalizeTag:
+    def test_strip_and_lower(self):
+        assert normalize_tag("  带货 ") == "带货"
+        assert normalize_tag("Factory") == "factory"
+
+    def test_none_and_non_string(self):
+        assert normalize_tag(None) == ""
+        assert normalize_tag(123) == "123"
+
+    def test_short_tag_filtered_by_normalize_set(self):
+        # 单字噪声标签不参与匹配（_normalize_tags 层过滤）
+        from packages.domain.narrative_match import _normalize_tags
+
+        assert _normalize_tags(["的", " a ", "工厂"]) == {"工厂"}
+
+
+# ── match_assets_by_script_tags ────────────────────────────────────────────
+
+
+class TestMatchSplit:
+    def test_split_by_tag_names(self):
+        assets = [
+            FakeAsset("a1", tags=["工厂"]),
+            FakeAsset("a2", tags=["旅游"]),
+            FakeAsset("a3", tags=["工厂", "车间"]),
+        ]
+        matched, unmatched = match_assets_by_script_tags(assets, script_tags=["工厂"])
+        assert [a.id for a in matched] == ["a1", "a3"]
+        assert [a.id for a in unmatched] == ["a2"]
+
+    def test_case_insensitive(self):
+        assets = [FakeAsset("a1", tags=["Factory"])]
+        matched, unmatched = match_assets_by_script_tags(assets, script_tags=["FACTORY"])
+        assert [a.id for a in matched] == ["a1"]
+        assert unmatched == []
+
+    def test_tag_ids_via_name_index(self):
+        assets = [FakeAsset("a1", tag_ids=["t1"]), FakeAsset("a2", tag_ids=["t2"])]
+        index = {"a1": ["测评"], "a2": ["vlog"]}
+        matched, unmatched = match_assets_by_script_tags(assets, script_tags=["测评"], tag_names_by_id=index)
+        assert [a.id for a in matched] == ["a1"]
+        assert [a.id for a in unmatched] == ["a2"]
+
+    def test_empty_script_tags_degrades_all_unmatched(self):
+        assets = [FakeAsset("a1", tags=["工厂"])]
+        matched, unmatched = match_assets_by_script_tags(assets, script_tags=[])
+        assert matched == []
+        assert [a.id for a in unmatched] == ["a1"]
+
+    def test_no_match_degrades(self):
+        assets = [FakeAsset("a1", tags=["工厂"]), FakeAsset("a2", tags=["车间"])]
+        matched, unmatched = match_assets_by_script_tags(assets, script_tags=["美食"])
+        assert matched == []
+        assert {a.id for a in unmatched} == {"a1", "a2"}
+
+    def test_order_preserved(self):
+        assets = [FakeAsset(f"a{i}", tags=["x" if i % 2 else "工厂"]) for i in range(6)]
+        matched, _ = match_assets_by_script_tags(assets, script_tags=["工厂"])
+        assert [a.id for a in matched] == ["a0", "a2", "a4"]
+
+    def test_build_index_ignores_blank(self):
+        # 空白/None/单字符噪声标签均不参与匹配
+        idx = build_asset_tag_name_index({"a1": [" 工厂 ", "", None, "A"]})
+        assert idx == {"a1": {"工厂"}}
+
+
+# ── pick_narrative_assets ──────────────────────────────────────────────────
+
+
+class TestPickNarrativeAssets:
+    def _assets(self):
+        # smart_match 需要 created_at（None 走 recency 兜底）
+        import datetime as dt
+
+        old = dt.datetime(2020, 1, 1, tzinfo=dt.UTC)
+        return [
+            FakeAsset("match1", tags=["工厂"], created_at=old),
+            FakeAsset("nomatch1", tags=["旅游"], created_at=old),
+            FakeAsset("match2", tags=["工厂"], created_at=old),
+            FakeAsset("nomatch2", tags=["美食"], created_at=old),
+        ]
+
+    def test_matched_pool_prioritized(self):
+        picked = pick_narrative_assets(self._assets(), script_tags=["工厂"], limit=2, rng=random.Random(0))
+        assert {a.id for a in picked} <= {"match1", "match2"}
+        assert all(a.id.startswith("match") for a in picked)
+
+    def test_fallback_fills_from_unmatched(self):
+        picked = pick_narrative_assets(self._assets(), script_tags=["工厂"], limit=4, rng=random.Random(0))
+        ids = {a.id for a in picked}
+        assert ids == {"match1", "match2", "nomatch1", "nomatch2"}
+        # 命中池排在前面
+        assert picked[0].id.startswith("match")
+        assert picked[1].id.startswith("match")
+
+    def test_no_tag_match_equals_random_selection(self):
+        assets = self._assets()
+        picked = pick_narrative_assets(assets, script_tags=["不存在"], limit=3, rng=random.Random(42))
+        assert len(picked) == 3
+
+    def test_empty_tags_selects_all_pool(self):
+        assets = self._assets()
+        picked = pick_narrative_assets(assets, script_tags=[], limit=None, rng=random.Random(1))
+        assert len(picked) == 4
+
+    def test_limit_none_returns_all_with_matched_first(self):
+        picked = pick_narrative_assets(self._assets(), script_tags=["工厂"], limit=None, rng=random.Random(1))
+        assert len(picked) == 4
+        assert {a.id for a in picked[:2]} == {"match1", "match2"}
+
+    def test_tag_ids_index_path(self):
+        assets = [FakeAsset("a1", tag_ids=["t1"]), FakeAsset("a2", tag_ids=["t2"])]
+        # 补 created_at
+        import datetime as dt
+
+        for a in assets:
+            a.created_at = dt.datetime(2020, 1, 1, tzinfo=dt.UTC)
+        picked = pick_narrative_assets(
+            assets,
+            script_tags=["教程"],
+            tag_names_by_id={"a1": ["教程"], "a2": ["旅游"]},
+            limit=1,
+            rng=random.Random(0),
+        )
+        assert [a.id for a in picked] == ["a1"]
+
+    def test_deterministic_with_seed(self):
+        r1 = pick_narrative_assets(self._assets(), script_tags=["工厂"], limit=4, rng=random.Random(7))
+        r2 = pick_narrative_assets(self._assets(), script_tags=["工厂"], limit=4, rng=random.Random(7))
+        assert [a.id for a in r1] == [a.id for a in r2]
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-q"])
@@ -0,0 +1,454 @@
+"""#1970 PR3 叙事前置服务 narrative_service 单元测试（不依赖真实 PG/OSS/CosyVoice）。"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+
+from apps.api.app.services import narrative_service as ns
+from apps.api.app.services.narrative_service import (
+    NarrativeError,
+    _resolve_voice,
+    _save_tts_job_as_voice_asset,
+    prepare_narrative_voice,
+)
+from packages.adapters.sqlalchemy_impl.models import ScriptModel
+from packages.domain.tts_job import TTSJob, TTSJobStatus
+
+# ── fakes ──────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class FakeProfile:
+    id: str = "prof-1"
+    user_id: str = "u1"
+    voice_id: str = "cv-voice-1"
+
+
+class FakeCloneRepo:
+    def __init__(self, profile: FakeProfile | None = None):
+        self._profile = profile
+
+    def get(self, pid: str) -> FakeProfile | None:
+        if self._profile and self._profile.id == pid:
+            return self._profile
+        return None
+
+
+class FakeQuery:
+    def __init__(self, script: ScriptModel | None):
+        self._script = script
+
+    def filter(self, *conditions):
+        # 服务端写 filter(...).filter(...) 链式调用；归属/ID 已在 FakeDb 构造时过滤
+        return self
+
+    def first(self):
+        return self._script
+
+
+class FakeDb:
+    def __init__(self, script: ScriptModel | None, *, current_user: str = "u1", query_script_id: str = "script-1"):
+        self._script = script
+        self._current_user = current_user
+        self._query_script_id = query_script_id
+
+    def query(self, model):
+        visible = self._script
+        if visible is not None and (visible.user_id != self._current_user or visible.id != self._query_script_id):
+            visible = None
+        return FakeQuery(visible)
+
+
+def _make_script(*, user_id: str = "u1", content: str = "这是一段口播文案", title: str = "测试文案", tags=None):
+    return ScriptModel(
+        id="script-1",
+        user_id=user_id,
+        title=title,
+        content=content,
+        segments=[],
+        tags=tags if tags is not None else ["带货"],
+    )
+
+
+@dataclass
+class FakeLibrary:
+    id: str = "lib-voice"
+    project_id: str = "p1"
+    kind: Any = field(default_factory=lambda: SimpleNamespace(value="voice"))
+
+
+@dataclass
+class FakeProject:
+    id: str = "p1"
+
+
+class FakeProjectRepo:
+    def __init__(self, projects=None):
+        self._projects = projects if projects is not None else [FakeProject()]
+
+    def find_accessible_projects(self, user_id):
+        return self._projects
+
+
+class FakeLibraryRepo:
+    def __init__(self, libs=None):
+        self._libs = libs if libs is not None else [FakeLibrary()]
+        self.created: list = []
+
+    def find_by_project(self, project_id):
+        return list(self._libs)
+
+    def create(self, library):
+        self.created.append(library)
+        return library
+
+
+@dataclass
+class FakeAsset:
+    id: str = "asset-new"
+    duration: float | None = 12.0
+
+
+class FakeAssetRepo:
+    def __init__(self):
+        self.created: list = []
+
+    def create(self, asset):
+        wrapped = FakeAsset(id="asset-new", duration=getattr(asset, "duration", None))
+        self.created.append(asset)
+        return wrapped
+
+
+class FakeStorage:
+    def __init__(self, *, fail_download: bool = False):
+        self.fail_download = fail_download
+        self.uploaded: list = []
+
+    def download_asset(self, source, dest_path) -> bool:
+        if self.fail_download:
+            return False
+        dest_path.write_bytes(b"FAKEAUDIO")
+        return True
+
+    def upload_file(self, path, key, content_type="", **kwargs):
+        self.uploaded.append((key, content_type))
+
+    def delete_file(self, key):
+        pass
+
+
+class FakeTTSRepo:
+    def __init__(self, job: TTSJob):
+        self.job = job
+        self.saved: list[TTSJob] = []
+
+    def create(self, job: TTSJob) -> TTSJob:
+        self.saved.append(job)
+        self.job = job
+        return job
+
+    def update(self, job: TTSJob) -> TTSJob:
+        self.job = job
+        return job
+
+    def get(self, job_id: str) -> TTSJob | None:
+        return self.job if self.job.id == job_id else None
+
+
+class FakeCosyVoice:
+    pass
+
+
+def _make_completed_job() -> TTSJob:
+    job = TTSJob.create(
+        user_id="u1",
+        input_text="这是一段口播文案",
+        voice_id="cv-voice-1",
+        voice_clone_profile_id="",
+        format="mp3",
+        sample_rate=22050,
+    )
+    job.mark_processing()
+    job.mark_completed(
+        output_audio_url="https://oss/tts/output/job-1.mp3",
+        output_audio_key="tts/output/job-1.mp3",
+        duration=12.5,
+    )
+    return job
+
+
+# ── _resolve_voice ─────────────────────────────────────────────────────────
+
+
+class TestResolveVoice:
+    def test_preset_returns_id_directly_when_no_profile(self):
+        voice_id, clone_id = _resolve_voice(
+            user_id="u1",
+            tts_voice_id="longxiaochun",
+            tts_voice_source="preset",
+            voice_clone_repository=FakeCloneRepo(None),
+        )
+        assert voice_id == "longxiaochun"
+        assert clone_id == ""
+
+    def test_preset_id_that_is_clone_profile_uuid_resolves(self):
+        repo = FakeCloneRepo(FakeProfile())
+        voice_id, clone_id = _resolve_voice(
+            user_id="u1",
+            tts_voice_id="prof-1",
+            tts_voice_source="preset",
+            voice_clone_repository=repo,
+        )
+        assert voice_id == "cv-voice-1"
+        assert clone_id == "prof-1"
+
+    def test_clone_source(self):
+        voice_id, clone_id = _resolve_voice(
+            user_id="u1",
+            tts_voice_id="prof-1",
+            tts_voice_source="clone",
+            voice_clone_repository=FakeCloneRepo(FakeProfile()),
+        )
+        assert voice_id == "cv-voice-1"
+        assert clone_id == "prof-1"
+
+    def test_clone_missing_404(self):
+        with pytest.raises(NarrativeError) as ei:
+            _resolve_voice(
+                user_id="u1",
+                tts_voice_id="nope",
+                tts_voice_source="clone",
+                voice_clone_repository=FakeCloneRepo(None),
+            )
+        assert ei.value.status_code == 404
+
+    def test_clone_other_user_403(self):
+        repo = FakeCloneRepo(FakeProfile(user_id="someone-else"))
+        with pytest.raises(NarrativeError) as ei:
+            _resolve_voice(
+                user_id="u1",
+                tts_voice_id="prof-1",
+                tts_voice_source="clone",
+                voice_clone_repository=repo,
+            )
+        assert ei.value.status_code == 403
+
+    def test_clone_not_ready_400(self):
+        repo = FakeCloneRepo(FakeProfile(voice_id=""))
+        with pytest.raises(NarrativeError) as ei:
+            _resolve_voice(
+                user_id="u1",
+                tts_voice_id="prof-1",
+                tts_voice_source="clone",
+                voice_clone_repository=repo,
+            )
+        assert ei.value.status_code == 400
+
+
+# ── save asset ─────────────────────────────────────────────────────────────
+
+
+class TestSaveVoiceAsset:
+    def _deps(self, **storage_kw):
+        return dict(
+            user_id="u1",
+            name="测试配音",
+            project_repository=FakeProjectRepo(),
+            asset_library_repository=FakeLibraryRepo(),
+            asset_repository=FakeAssetRepo(),
+            storage_service=FakeStorage(**storage_kw),
+        )
+
+    def test_save_creates_asset(self):
+        job = _make_completed_job()
+        deps = self._deps()
+        asset = _save_tts_job_as_voice_asset(job=job, **deps)
+        assert asset.id == "asset-new"
+        assert deps["asset_repository"].created[0].mime_type == "audio/mpeg"
+        assert deps["storage_service"].uploaded[0][0] == "uploads/voice/tts/" + job.id + ".mp3"
+
+    def test_no_project_raises(self):
+        job = _make_completed_job()
+        deps = self._deps()
+        deps["project_repository"] = FakeProjectRepo(projects=[])
+        with pytest.raises(NarrativeError):
+            _save_tts_job_as_voice_asset(job=job, **deps)
+
+    def test_download_fail_raises_502(self):
+        job = _make_completed_job()
+        deps = self._deps(fail_download=True)
+        with pytest.raises(NarrativeError) as ei:
+            _save_tts_job_as_voice_asset(job=job, **deps)
+        assert ei.value.status_code == 502
+
+    def test_job_without_output_raises(self):
+        job = TTSJob.create(user_id="u1", input_text="x", voice_id="v", voice_clone_profile_id="")
+        with pytest.raises(NarrativeError) as ei:
+            _save_tts_job_as_voice_asset(job=job, **self._deps())
+        assert ei.value.status_code == 502
+
+
+# ── prepare_narrative_voice 主流程（monkeypatch workflow） ─────────────────
+
+
+class TestPrepareNarrativeVoice:
+    def _deps(self, db_script=None, *, has_script=True, clone_profile=None, storage_fail=False, points_enabled=False):
+        job = _make_completed_job()
+        return dict(
+            db=FakeDb(db_script if db_script is not None else (_make_script() if has_script else None)),
+            user_id="u1",
+            script_id="script-1",
+            tts_voice_id="longxiaochun",
+            tts_voice_source="preset",
+            tts_repository=FakeTTSRepo(job),
+            cosyvoice_service=FakeCosyVoice(),
+            voice_clone_repository=FakeCloneRepo(clone_profile),
+            asset_repository=FakeAssetRepo(),
+            asset_library_repository=FakeLibraryRepo(),
+            project_repository=FakeProjectRepo(),
+            storage_service=FakeStorage(fail_download=storage_fail),
+            points_enabled=points_enabled,
+        )
+
+    def test_success_returns_context(self, monkeypatch):
+        captured = {}
+
+        class FakeWorkflow:
+            def __init__(self, *, repository, cosyvoice_service):
+                captured["repo"] = repository
+                self._repo = repository
+
+            def start_synthesis(self, job_id):
+                job = self._repo.get(job_id)
+                job.mark_processing()
+                job.mark_completed(
+                    output_audio_url="https://oss/tts/output/x.mp3",
+                    output_audio_key="tts/output/x.mp3",
+                    duration=12.5,
+                )
+                return job
+
+            def poll_and_process_synthesis(self, job_id, timeout=120.0):
+                return self._repo.get(job_id)
+
+        monkeypatch.setattr(ns, "TTSWorkflowService", FakeWorkflow)
+        ctx = prepare_narrative_voice(**self._deps())
+        assert ctx.voice_asset_id == "asset-new"
+        assert ctx.tts_job_id
+        assert ctx.audio_duration == pytest.approx(12.5)
+        assert ctx.script.tags == ["带货"]
+
+    def test_script_missing_404(self):
+        deps = self._deps(has_script=False)
+        with pytest.raises(NarrativeError) as ei:
+            prepare_narrative_voice(**deps)
+        assert ei.value.status_code == 404
+
+    def test_script_other_user_404(self):
+        deps = self._deps(db_script=_make_script(user_id="other"))
+        with pytest.raises(NarrativeError) as ei:
+            prepare_narrative_voice(**deps)
+        assert ei.value.status_code == 404
+
+    def test_empty_content_400(self):
+        deps = self._deps(db_script=_make_script(content="   "))
+        with pytest.raises(NarrativeError) as ei:
+            prepare_narrative_voice(**deps)
+        assert ei.value.status_code == 400
+
+    def test_synth_failure_raises_502(self, monkeypatch):
+        class FailingWorkflow:
+            def __init__(self, *, repository, cosyvoice_service):
+                self._repo = repository
+
+            def start_synthesis(self, job_id):
+                raise RuntimeError("cosyvoice down")
+
+            def process_synthesis_failure(self, job_id, error):
+                return None
+
+        monkeypatch.setattr(ns, "TTSWorkflowService", FailingWorkflow)
+        with pytest.raises(NarrativeError) as ei:
+            prepare_narrative_voice(**self._deps())
+        assert ei.value.status_code == 502
+        assert "配音合成失败" in ei.value.message
+
+    def test_points_insufficient_402(self, monkeypatch):
+        class FakePoints:
+            def deduct_points(self, *a, **k):
+                return {"success": False, "balance": 0}
+
+        monkeypatch.setattr(ns, "PointsService", lambda: FakePoints())
+        deps = self._deps(points_enabled=True)
+        with pytest.raises(NarrativeError) as ei:
+            prepare_narrative_voice(**deps)
+        assert ei.value.status_code == 402
+
+    def test_points_refund_on_failure(self, monkeypatch):
+        class FakePoints:
+            def __init__(self):
+                self.refunded = 0
+
+            def deduct_points(self, *a, **k):
+                return {"success": True, "balance": 100}
+
+            def refund_points(self, user_id, amount, source, db, ref_id="", **k):
+                self.refunded += amount
+
+        points = FakePoints()
+        monkeypatch.setattr(ns, "PointsService", lambda: points)
+
+        class FailingWorkflow:
+            def __init__(self, *, repository, cosyvoice_service):
+                pass
+
+            def start_synthesis(self, job_id):
+                raise RuntimeError("boom")
+
+            def process_synthesis_failure(self, job_id, error):
+                return None
+
+        monkeypatch.setattr(ns, "TTSWorkflowService", FailingWorkflow)
+        deps = self._deps(points_enabled=True)
+        with pytest.raises(NarrativeError):
+            prepare_narrative_voice(**deps)
+        assert points.refunded > 0
+
+    def test_clone_source_resolves_profile(self, monkeypatch):
+        captured = {}
+
+        class FakeWorkflow:
+            def __init__(self, *, repository, cosyvoice_service):
+                self._repo = repository
+                captured["cosy"] = cosyvoice_service
+
+            def start_synthesis(self, job_id):
+                job = self._repo.get(job_id)
+                captured["voice_id"] = job.voice_id
+                job.mark_processing()
+                job.mark_completed(
+                    output_audio_url="https://oss/tts/output/x.mp3",
+                    output_audio_key="tts/output/x.mp3",
+                    duration=12.5,
+                )
+                return job
+
+            def poll_and_process_synthesis(self, job_id, timeout=120.0):
+                return self._repo.get(job_id)
+
+        monkeypatch.setattr(ns, "TTSWorkflowService", FakeWorkflow)
+        deps = self._deps(clone_profile=FakeProfile())
+        deps["tts_voice_id"] = "prof-1"
+        deps["tts_voice_source"] = "clone"
+        prepare_narrative_voice(**deps)
+        assert captured["voice_id"] == "cv-voice-1"
+
+
+if __name__ == "__main__":
+    import pytest as _pytest
+
+    _pytest.main([__file__, "-q"])
@@ -187,29 +187,39 @@ def test_asr_not_configured_returns_503(fake_user):
 # ── ASR 转写失败 → 502 ────────────────────────────────────────


-def test_asr_transcription_failure_returns_502(fake_user):
-    """ASR 转写异常 → 502（被 _direct_url_download_and_local_asr 包装）。"""
-    from app.services.script_asr_service import ASRTranscriptionError
+def test_asr_transcription_failure_returns_503_with_desc_fallback(fake_user):
+    """ASR 转写异常（MediaKit+本地都失败）→ desc 兜底；desc 也空则 503。"""
+    from app.api.routes import scripts_ai
+    from app.services.mediakit_client import MediaKitError
+    from fastapi import HTTPException

-    scripts_ai = _import_target()
    body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/xxxxx/")

-    # MediaKit 失败
    fake_mk = mock.MagicMock()
    fake_mk.is_available = True
-    from app.services.mediakit_client import MediaKitError
-
    fake_mk.asr_submit.side_effect = MediaKitError("ASR failed", code="TaskFailed")

+    # desc 为空 → 最终 503（stage=asr）
    with _fake_resolver_success(desc=""):
+        with mock.patch("app.api.routes.scripts_ai.get_mediakit_client", return_value=fake_mk):
+            with mock.patch(
+                "app.api.routes.scripts_ai._direct_url_download_and_local_asr",
+                side_effect=HTTPException(status_code=502, detail="语音识别失败: No module named 'apps.worker'"),
+            ):
+                with pytest.raises(HTTPException) as exc:
+                    scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
+                assert exc.value.status_code == status.HTTP_503_SERVICE_UNAVAILABLE
+
+    # desc 非空 → desc 兜底成功，返回 200
+    with _fake_resolver_success(desc="这是视频文案描述"):
        with mock.patch("app.api.routes.scripts_ai.get_mediakit_client", return_value=fake_mk):
            with mock.patch(
                "app.api.routes.scripts_ai._direct_url_download_and_local_asr",
                side_effect=HTTPException(status_code=502, detail="语音识别失败"),
            ):
-                with pytest.raises(HTTPException) as exc:
-                    scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
-                assert exc.value.status_code == status.HTTP_502_BAD_GATEWAY
+                resp = scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
+                assert resp.text == "这是视频文案描述"
+                assert resp.duration_seconds == 0.0


 # ── 下载超时 → 504 ────────────────────────────────────────────
@@ -297,3 +297,74 @@ class TestResolveLatestPlanByTemplate:
        with caplog.at_level("WARNING"):
            assert resolve_latest_plan_by_template(db, template_id="tpl", user_id="u") is None
        assert any("查找最新plan失败" in rec.message for rec in caplog.records)
+
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# collect_plan_atom_clip_ids (#1970)
+# ═══════════════════════════════════════════════════════════════════════════════
+
+
+def _make_atom_clip(atom_clip_id):
+    c = MagicMock()
+    c.atom_clip_id = atom_clip_id
+    return c
+
+
+class TestCollectPlanAtomClipIds:
+    def test_empty_plan_returns_empty(self):
+        from app.services.generation_common import collect_plan_atom_clip_ids
+
+        repo = MagicMock()
+        repo.list_by_plan.return_value = []
+        assert collect_plan_atom_clip_ids("p1", repo) == []
+
+    def test_collects_non_empty_ids_and_ignores_blank(self):
+        from app.services.generation_common import collect_plan_atom_clip_ids
+
+        repo = MagicMock()
+        repo.list_by_plan.side_effect = [
+            [
+                _make_atom_clip("atom-1"),
+                _make_atom_clip(""),
+                _make_atom_clip("atom-2"),
+            ],
+            [],
+        ]
+        assert collect_plan_atom_clip_ids("p1", repo) == ["atom-1", "atom-2"]
+
+    def test_missing_attribute_treated_as_blank(self):
+        from app.services.generation_common import collect_plan_atom_clip_ids
+
+        legacy = MagicMock()
+        del legacy.atom_clip_id  # 旧对象无该属性
+        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
@@ -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)