diff --git a/apps/api/app/api/routes/templates_editor/clips.py b/apps/api/app/api/routes/templates_editor/clips.py
index d43965aa8..515e3eaa4 100755
--- a/apps/api/app/api/routes/templates_editor/clips.py
+++ b/apps/api/app/api/routes/templates_editor/clips.py
@@ -25,6 +25,7 @@ from app.services.edit_template_service import EditTemplateService
 from fastapi import APIRouter, Depends, HTTPException, Query, status
 
 from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
+from packages.domain.plan_generator_utils import _calc_random_start_time
 
 from .dependencies import get_draft_plan_id, get_editor_services
 from .schemas import (
@@ -45,6 +46,9 @@ from .schemas import (
 logger = logging.getLogger(__name__)
 router = APIRouter(tags=["Template Editor"])
 
+# 编辑器默认片段时长（秒）
+_DEFAULT_EDITOR_CLIP_DURATION = 5.0
+
 
 def _clip_to_response(clip, asset_url: str | None = None) -> EditorClipResponse:
     """统一构造片段响应 — 与 edit_plan_clips 表字段完全对齐"""
@@ -155,10 +159,7 @@ def list_draft_clips(
     url_map = _build_asset_url_map(asset_ids, asset_repo)
 
     return EditorClipListResponse(
-        items=[
-            _clip_to_response(c, asset_url=url_map.get(getattr(c, "asset_id", "") or ""))
-            for c in clips
-        ],
+        items=[_clip_to_response(c, asset_url=url_map.get(getattr(c, "asset_id", "") or "")) for c in clips],
         total=total,
     )
 
@@ -272,9 +273,7 @@ def split_draft_clip(
     try:
         result = plan_svc.split_clip(clip_id, body.split_time)
     except ValueError as exc:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)
-        ) from exc
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
     left = result["left_clip"]
     right = result["right_clip"]
     asset_ids = [getattr(left, "asset_id", "") or "", getattr(right, "asset_id", "") or ""]
@@ -304,9 +303,7 @@ def merge_draft_clips(
     try:
         merged = plan_svc.merge_clips(body.clip_ids)
     except ValueError as exc:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)
-        ) from exc
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
     asset_id = getattr(merged, "asset_id", "") or ""
     url_map = _build_asset_url_map([asset_id], asset_repo) if asset_id else {}
     return {
@@ -360,28 +357,90 @@ def create_clips_from_assets_editor(
     body: ClipsFromAssetsRequest,
     plan_id: str = Depends(get_draft_plan_id),
     services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
+    asset_repo: SQLAlchemyAssetRepository = Depends(get_asset_repository),
     current_user: AuthenticatedUser = Depends(get_current_user),
 ) -> ClipsFromAssetsResponse:
-    """从素材批量创建片段"""
+    """从素材批量创建片段（支持同一素材切多个片段 + 随机起始时间 + 去重）.
+
+    逻辑：
+    1. 模板要求 N 个片段，必须创建 N 个（不管素材有几个）
+    2. 素材数量 < 片段数量时，同一素材轮询切多个片段
+    3. 每个片段从素材中随机选取不重复时间段
+    4. 素材时长不足 5s 时缩短 clip duration
+    5. 新片段追加到时间线末尾（order 在现有最大值基础上递增）
+    """
     _, plan_svc = services
+
+    required_count = body.required_clips_count if body.required_clips_count is not None else len(body.asset_ids)
+
+    # 去重后批量获取素材实际时长，避免重复查询
+    unique_asset_ids = list(dict.fromkeys(body.asset_ids))
+    asset_durations: dict[str, float] = {}
+    for asset_id in unique_asset_ids:
+        asset = asset_repo.get(asset_id)
+        if asset and hasattr(asset, "duration"):
+            asset_durations[asset_id] = float(asset.duration or 0.0)
+
+    # 计算追加起始 order：当前 plan 已有片段的最大 order + 1
+    # list_clips 返回 List[EditPlanClip]
+    existing_clips_list = plan_svc.list_clips(plan_id)
+    next_order = max((c.order for c in existing_clips_list), default=-1) + 1
+
+    used_segments: dict[str, list[tuple[float, float]]] = {}
     clips = []
-    for i, asset_id in enumerate(body.asset_ids):
+
+    for i in range(required_count):
+        # 轮询分配素材：素材不够时同一素材切多个片段
+        asset_id = body.asset_ids[i % len(body.asset_ids)]
+
+        # 素材时长不足时缩短 clip duration
+        asset_total = asset_durations.get(asset_id, 0.0)
+        if asset_total > 0:
+            clip_duration = min(_DEFAULT_EDITOR_CLIP_DURATION, asset_total)
+        else:
+            clip_duration = _DEFAULT_EDITOR_CLIP_DURATION
+
+        # 计算随机 start_time，避开已使用的时间段
+        start_time = _calc_random_start_time(asset_id, clip_duration, asset_durations, used_segments)
+        if start_time is None:
+            # 素材时长信息缺失，无法计算随机起始时间
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"素材 {asset_id} 时长信息缺失，无法创建片段",
+            )
+
+        # 记录已使用时间段（用于去重）
+        used_segments.setdefault(asset_id, []).append((start_time, start_time + clip_duration))
+
         try:
             clip = plan_svc.create_clip(
                 plan_id,
-                clip_type="main",
-                order=body.start_order + i if hasattr(body, "start_order") else i,
-                duration=5.0,
+                clip_type=body.clip_type or "main",
+                order=next_order + i,
+                duration=clip_duration,
+                start_time=start_time,
                 asset_id=asset_id,
             )
             clips.append(clip)
-        except ValueError:
-            pass
+        except ValueError as exc:
+            logger.error(
+                "创建片段失败: plan_id=%s asset_id=%s order=%d error=%s",
+                plan_id,
+                asset_id,
+                next_order + i,
+                exc,
+                exc_info=True,
+            )
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"创建片段失败: {exc}",
+            ) from exc
 
     logger.info(
-        "模板编辑器从素材创建片段: template_id=%s plan_id=%s count=%d by user=%s",
+        "模板编辑器从素材创建片段: template_id=%s plan_id=%s required=%d actual=%d by user=%s",
         template_id,
         plan_id,
+        required_count,
         len(clips),
         current_user.user.id,
     )
diff --git a/apps/api/app/api/routes/templates_editor/schemas.py b/apps/api/app/api/routes/templates_editor/schemas.py
index 8135c9a5e..2403d22b5 100755
--- a/apps/api/app/api/routes/templates_editor/schemas.py
+++ b/apps/api/app/api/routes/templates_editor/schemas.py
@@ -167,6 +167,7 @@ class ClipsFromAssetsRequest(BaseModel):
 
     asset_ids: List[str] = Field(..., min_length=1, max_length=200, description="素材 ID 列表，按顺序追加到时间线末尾")
     clip_type: str = Field(default="main", description="片段类型，默认 main")
+    required_clips_count: Optional[int] = Field(default=None, ge=1, le=200, description="要求创建的片段数量；不传则等于素材数量")
 
 
 class ClipsFromAssetsResponse(BaseModel):
diff --git a/tests/unit/test_editor_clips_random_start.py b/tests/unit/test_editor_clips_random_start.py
new file mode 100644
index 000000000..e2336ef72
--- /dev/null
+++ b/tests/unit/test_editor_clips_random_start.py
@@ -0,0 +1,396 @@
+"""测试编辑器 from-assets 端点：同一素材切多个片段 + 随机起始 + 去重.
+
+覆盖：
+- required_clips_count 精确控制片段数量
+- 素材不足时同一素材轮询切多个片段
+- 随机 start_time + used_segments 去重
+- 素材时长不足时 clip duration 缩短
+- 向后兼容（不传 required_clips_count 时等于素材数量）
+- order 追加到时间线末尾
+- start_time=None 时抛出 400
+- create_clip 失败时抛出 400 并记录日志
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+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 fastapi import HTTPException
+
+TEST_PLAN_ID = "plan-draft-001"
+TEST_USER_ID = "user-001"
+
+
+def _make_auth_user():
+    auth = MagicMock()
+    auth.user.id = TEST_USER_ID
+    auth.user.email = "test@example.com"
+    auth.user.display_name = "测试用户"
+    auth.user_id = TEST_USER_ID
+    return auth
+
+
+def _make_mock_clip(clip_id, order, duration, start_time=0.0, asset_id=""):
+    clip = MagicMock()
+    clip.id = clip_id
+    clip.plan_id = TEST_PLAN_ID
+    clip.clip_type = "main"
+    clip.order = order
+    clip.duration = duration
+    clip.start_time = start_time
+    clip.text_content = ""
+    clip.transition_effect = "cut"
+    clip.transition_duration = 0.0
+    clip.playback_speed = 1.0
+    clip.config = {}
+    clip.asset_id = asset_id
+    clip.status = "pending"
+    clip.template_clip_config_id = ""
+    clip.created_at = None
+    clip.updated_at = None
+    return clip
+
+
+def _make_mock_asset(asset_id, duration):
+    asset = MagicMock()
+    asset.id = asset_id
+    asset.duration = duration
+    return asset
+
+
+def _create_clips(plan_id, clip_type, order, duration=0.0, start_time=0.0, asset_id="", **kw):
+    return _make_mock_clip(
+        clip_id=f"clip-{order}",
+        order=order,
+        duration=duration,
+        start_time=start_time,
+        asset_id=asset_id,
+    )
+
+
+def _make_plan_svc(existing_clips=None):
+    svc = MagicMock()
+    svc.get_plan_or_raise = MagicMock()
+    svc.create_clip = MagicMock(side_effect=_create_clips)
+    svc.list_clips = MagicMock(return_value=existing_clips or [])
+    return svc
+
+
+class TestEditorClipsRequiredCount:
+    """测试 required_clips_count 控制片段数量 + 同素材多片段."""
+
+    @patch("app.api.routes.templates_editor.clips.get_storage_service")
+    def test_creates_exactly_required_clips_count(self, mock_storage):
+        """required_clips_count=4 时，即使只有2个素材也创建4个片段."""
+        from app.api.routes.templates_editor.clips import (
+            create_clips_from_assets_editor,
+        )
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        mock_plan_svc = _make_plan_svc()
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(side_effect=lambda aid: _make_mock_asset(aid, {"a1": 30.0, "a2": 20.0}[aid]))
+
+        body = ClipsFromAssetsRequest(asset_ids=["a1", "a2"], required_clips_count=4)
+
+        result = create_clips_from_assets_editor(
+            template_id="tpl-001",
+            body=body,
+            plan_id=TEST_PLAN_ID,
+            services=(MagicMock(), mock_plan_svc),
+            asset_repo=mock_asset_repo,
+            current_user=_make_auth_user(),
+        )
+
+        assert result.created_count == 4
+        assert mock_plan_svc.create_clip.call_count == 4
+
+        # 验证轮询分配：a1, a2, a1, a2
+        calls = mock_plan_svc.create_clip.call_args_list
+        assert calls[0].kwargs["asset_id"] == "a1"
+        assert calls[1].kwargs["asset_id"] == "a2"
+        assert calls[2].kwargs["asset_id"] == "a1"
+        assert calls[3].kwargs["asset_id"] == "a2"
+
+    @patch("app.api.routes.templates_editor.clips.get_storage_service")
+    def test_orders_append_to_existing_timeline(self, mock_storage):
+        """时间线已有2个片段时，新片段 order 应从 2 开始连续递增."""
+        from app.api.routes.templates_editor.clips import (
+            create_clips_from_assets_editor,
+        )
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        # 模拟已有 order=0, order=1 的片段
+        existing = [_make_mock_clip("old-1", 0, 5.0), _make_mock_clip("old-2", 1, 5.0)]
+        mock_plan_svc = _make_plan_svc(existing_clips=existing)
+
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))
+
+        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=3)
+
+        create_clips_from_assets_editor(
+            template_id="tpl-001",
+            body=body,
+            plan_id=TEST_PLAN_ID,
+            services=(MagicMock(), mock_plan_svc),
+            asset_repo=mock_asset_repo,
+            current_user=_make_auth_user(),
+        )
+
+        calls = mock_plan_svc.create_clip.call_args_list
+        assert calls[0].kwargs["order"] == 2
+        assert calls[1].kwargs["order"] == 3
+        assert calls[2].kwargs["order"] == 4
+
+    @patch("app.api.routes.templates_editor.clips.get_storage_service")
+    def test_orders_start_at_zero_when_empty(self, mock_storage):
+        """空时间线时 order 从 0 开始."""
+        from app.api.routes.templates_editor.clips import (
+            create_clips_from_assets_editor,
+        )
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        mock_plan_svc = _make_plan_svc(existing_clips=[])
+
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))
+
+        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=3)
+
+        create_clips_from_assets_editor(
+            template_id="tpl-001",
+            body=body,
+            plan_id=TEST_PLAN_ID,
+            services=(MagicMock(), mock_plan_svc),
+            asset_repo=mock_asset_repo,
+            current_user=_make_auth_user(),
+        )
+
+        calls = mock_plan_svc.create_clip.call_args_list
+        assert calls[0].kwargs["order"] == 0
+        assert calls[1].kwargs["order"] == 1
+        assert calls[2].kwargs["order"] == 2
+
+    @patch("app.api.routes.templates_editor.clips.get_storage_service")
+    def test_backward_compatible_default_count(self, mock_storage):
+        """不传 required_clips_count 时，片段数等于素材数（向后兼容）."""
+        from app.api.routes.templates_editor.clips import (
+            create_clips_from_assets_editor,
+        )
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        mock_plan_svc = _make_plan_svc()
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(side_effect=lambda aid: _make_mock_asset(aid, 30.0))
+
+        body = ClipsFromAssetsRequest(asset_ids=["a1", "a2", "a3"])
+
+        result = create_clips_from_assets_editor(
+            template_id="tpl-001",
+            body=body,
+            plan_id=TEST_PLAN_ID,
+            services=(MagicMock(), mock_plan_svc),
+            asset_repo=mock_asset_repo,
+            current_user=_make_auth_user(),
+        )
+
+        assert result.created_count == 3
+        assert mock_plan_svc.create_clip.call_count == 3
+
+
+class TestEditorClipsDurationAndStartTime:
+    """测试素材时长获取、clip duration 缩短、start_time 传入."""
+
+    @patch("app.api.routes.templates_editor.clips.get_storage_service")
+    def test_clip_duration_shortened_for_short_assets(self, mock_storage):
+        """素材只有 3s 时 clip duration 缩短到 3.0."""
+        from app.api.routes.templates_editor.clips import (
+            create_clips_from_assets_editor,
+        )
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        mock_plan_svc = _make_plan_svc()
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("short", 3.0))
+
+        body = ClipsFromAssetsRequest(asset_ids=["short"])
+
+        create_clips_from_assets_editor(
+            template_id="tpl-001",
+            body=body,
+            plan_id=TEST_PLAN_ID,
+            services=(MagicMock(), mock_plan_svc),
+            asset_repo=mock_asset_repo,
+            current_user=_make_auth_user(),
+        )
+
+        calls = mock_plan_svc.create_clip.call_args_list
+        assert len(calls) == 1
+        assert calls[0].kwargs["duration"] == 3.0
+
+    @patch("app.api.routes.templates_editor.clips.get_storage_service")
+    def test_start_time_passed_to_create_clip(self, mock_storage):
+        """验证 _calc_random_start_time 返回值被传入 create_clip."""
+        from app.api.routes.templates_editor.clips import (
+            create_clips_from_assets_editor,
+        )
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        mock_plan_svc = _make_plan_svc()
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))
+
+        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=2)
+
+        with patch(
+            "app.api.routes.templates_editor.clips._calc_random_start_time",
+            side_effect=[12.5, 18.0],
+        ) as mock_calc:
+            create_clips_from_assets_editor(
+                template_id="tpl-001",
+                body=body,
+                plan_id=TEST_PLAN_ID,
+                services=(MagicMock(), mock_plan_svc),
+                asset_repo=mock_asset_repo,
+                current_user=_make_auth_user(),
+            )
+
+        assert mock_calc.call_count == 2
+        calls = mock_plan_svc.create_clip.call_args_list
+        assert calls[0].kwargs["start_time"] == 12.5
+        assert calls[1].kwargs["start_time"] == 18.0
+
+    @patch("app.api.routes.templates_editor.clips.get_storage_service")
+    def test_asset_durations_deduped(self, mock_storage):
+        """asset_ids 有重复时只查询一次素材时长."""
+        from app.api.routes.templates_editor.clips import (
+            create_clips_from_assets_editor,
+        )
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        mock_plan_svc = _make_plan_svc()
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))
+
+        # a1 出现 3 次，但时长只应查一次
+        body = ClipsFromAssetsRequest(asset_ids=["a1", "a1", "a1"])
+
+        create_clips_from_assets_editor(
+            template_id="tpl-001",
+            body=body,
+            plan_id=TEST_PLAN_ID,
+            services=(MagicMock(), mock_plan_svc),
+            asset_repo=mock_asset_repo,
+            current_user=_make_auth_user(),
+        )
+
+        # 去重后只调用 1 次
+        assert mock_asset_repo.get.call_count == 1
+
+    @patch("app.api.routes.templates_editor.clips.get_storage_service")
+    def test_used_segments_passed_to_calc(self, mock_storage):
+        """同一素材切多个片段时，used_segments 应被维护并传入."""
+        from app.api.routes.templates_editor.clips import (
+            create_clips_from_assets_editor,
+        )
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        mock_plan_svc = _make_plan_svc()
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))
+
+        body = ClipsFromAssetsRequest(asset_ids=["a1"], required_clips_count=3)
+
+        captured_used_segments = []
+
+        def fake_calc(asset_id, clip_duration, asset_durations, used_segments):
+            captured_used_segments.append({aid: list(segs) for aid, segs in (used_segments or {}).items()})
+            return (len(captured_used_segments) - 1) * 5.0
+
+        with patch(
+            "app.api.routes.templates_editor.clips._calc_random_start_time",
+            side_effect=fake_calc,
+        ):
+            create_clips_from_assets_editor(
+                template_id="tpl-001",
+                body=body,
+                plan_id=TEST_PLAN_ID,
+                services=(MagicMock(), mock_plan_svc),
+                asset_repo=mock_asset_repo,
+                current_user=_make_auth_user(),
+            )
+
+        assert captured_used_segments[0] == {}
+        assert captured_used_segments[1] == {"a1": [(0.0, 5.0)]}
+        assert captured_used_segments[2] == {"a1": [(0.0, 5.0), (5.0, 10.0)]}
+
+
+class TestEditorClipsErrorHandling:
+    """测试异常处理."""
+
+    @patch("app.api.routes.templates_editor.clips.get_storage_service")
+    def test_none_start_time_raises_400(self, mock_storage):
+        """_calc_random_start_time 返回 None 时应抛出 HTTPException 400."""
+        from app.api.routes.templates_editor.clips import (
+            create_clips_from_assets_editor,
+        )
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        mock_plan_svc = _make_plan_svc()
+        # asset_repo.get 返回 None → asset_durations 为空 → _calc_random_start_time 返回 None
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(return_value=None)
+
+        body = ClipsFromAssetsRequest(asset_ids=["missing-asset"])
+
+        with pytest.raises(HTTPException) as exc_info:
+            create_clips_from_assets_editor(
+                template_id="tpl-001",
+                body=body,
+                plan_id=TEST_PLAN_ID,
+                services=(MagicMock(), mock_plan_svc),
+                asset_repo=mock_asset_repo,
+                current_user=_make_auth_user(),
+            )
+
+        assert exc_info.value.status_code == 400
+        assert "时长信息缺失" in exc_info.value.detail
+
+    @patch("app.api.routes.templates_editor.clips.get_storage_service")
+    def test_create_clip_value_error_raises_400(self, mock_storage):
+        """create_clip 抛出 ValueError 时应转为 HTTPException 400."""
+        from app.api.routes.templates_editor.clips import (
+            create_clips_from_assets_editor,
+        )
+        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest
+
+        mock_plan_svc = _make_plan_svc()
+        mock_plan_svc.create_clip = MagicMock(side_effect=ValueError("计划不存在"))
+
+        mock_asset_repo = MagicMock()
+        mock_asset_repo.get = MagicMock(return_value=_make_mock_asset("a1", 30.0))
+
+        body = ClipsFromAssetsRequest(asset_ids=["a1"])
+
+        with pytest.raises(HTTPException) as exc_info:
+            create_clips_from_assets_editor(
+                template_id="tpl-001",
+                body=body,
+                plan_id=TEST_PLAN_ID,
+                services=(MagicMock(), mock_plan_svc),
+                asset_repo=mock_asset_repo,
+                current_user=_make_auth_user(),
+            )
+
+        assert exc_info.value.status_code == 400
+        assert "创建片段失败" in exc_info.value.detail