diff --git a/apps/api/app/api/routes/templates_editor/_fallback.py b/apps/api/app/api/routes/templates_editor/_fallback.py
index deafea778..031cf7d12 100755
--- a/apps/api/app/api/routes/templates_editor/_fallback.py
+++ b/apps/api/app/api/routes/templates_editor/_fallback.py
@@ -27,18 +27,14 @@ from packages.domain.edit_plan import EditPlanStatus
 logger = logging.getLogger(__name__)
 
 
-def _auto_fallback_draft_to_editing(
-    svc: EditPlanService, plan_id: str, plan_check
-) -> None:
+def _auto_fallback_draft_to_editing(svc: EditPlanService, plan_id: str, plan_check) -> None:
     """自动兜底 1: draft → editing"""
     if plan_check.status == EditPlanStatus.DRAFT:
         logger.info("模板编辑器自动兜底: plan=%s draft→editing", plan_id)
         svc.transition_status(plan_id, EditPlanStatus.EDITING)
 
 
-def _auto_fallback_copy_template_clips(
-    svc: EditPlanService, plan_id: str, plan_check, db: Session
-) -> None:
+def _auto_fallback_copy_template_clips(svc: EditPlanService, plan_id: str, plan_check, db: Session) -> None:
     """自动兜底 2: 无片段 + 有 template_id → 从模板复制片段配置"""
     existing_clips = svc.count_clips(plan_id)
     if existing_clips == 0 and plan_check.template_id:
@@ -53,15 +49,15 @@ def _auto_fallback_copy_template_clips(
             for cfg in configs:
                 svc.create_clip(
                     plan_id=plan_id,
-                    clip_type=cfg.clip_type.value
-                    if hasattr(cfg.clip_type, "value")
-                    else cfg.clip_type,
+                    clip_type=cfg.clip_type.value if hasattr(cfg.clip_type, "value") else cfg.clip_type,
                     order=cfg.order,
                     template_clip_config_id=cfg.id,
                     duration=cfg.default_duration,
-                    transition_effect=cfg.transition_effect.value
-                    if hasattr(cfg.transition_effect, "value")
-                    else cfg.transition_effect,
+                    transition_effect=(
+                        cfg.transition_effect.value
+                        if hasattr(cfg.transition_effect, "value")
+                        else cfg.transition_effect
+                    ),
                 )
             logger.info(
                 "模板编辑器自动兜底: plan=%s 从 template_clip_configs 复制了 %d 个片段",
@@ -90,17 +86,14 @@ def _auto_fallback_copy_template_clips(
             )
 
 
-def _auto_fallback_assign_assets(
-    svc: EditPlanService, plan_id: str, plan_check
-) -> list:
+def _auto_fallback_assign_assets(svc: EditPlanService, plan_id: str, plan_check) -> list:
     """自动兜底 3: 为没有素材的片段分配素材。返回剩余无素材片段列表。"""
     all_clips = svc.list_clips(plan_id)
     clips_without_asset = [c for c in all_clips if not c.asset_id]
     config_asset_ids = (plan_check.config or {}).get("asset_ids", [])
 
     logger.info(
-        "模板编辑器自动兜底3 诊断: plan=%s total_clips=%d "
-        "clips_without_asset=%d config_asset_ids=%r",
+        "模板编辑器自动兜底3 诊断: plan=%s total_clips=%d " "clips_without_asset=%d config_asset_ids=%r",
         plan_id,
         len(all_clips),
         len(clips_without_asset),
@@ -161,43 +154,74 @@ def _auto_fallback_auto_material_mode(
     clips_without_asset: list,
     asset_library_repo: Any,
     asset_repo: Any,
+    user_id: str = "",
 ) -> None:
-    """自动兜底 4: 项目有视频素材库时自动选素材"""
+    """自动兜底 4: 自动选素材分配给无素材片段
+
+    查找策略（按优先级）：
+    1. plan 有 project_id → 从项目素材库查找
+    2. plan 无 project_id 但有 user_id → 从用户上传的素材中查找
+    """
     if not clips_without_asset:
         return
-    if not plan_check.project_id:
+
+    ready_videos: list = []
+    source_desc = ""
+
+    # 策略 1: 通过 project_id 查找项目素材库
+    if plan_check.project_id:
+        libs = asset_library_repo.find_by_project(plan_check.project_id)
+        video_lib = None
+        for lib in libs:
+            lib_kind = lib.kind.value if hasattr(lib.kind, "value") else lib.kind
+            if lib_kind == "video":
+                video_lib = lib
+                break
+        if video_lib:
+            assets = asset_repo.find_by_library(video_lib.id)
+            ready_videos = [
+                a
+                for a in assets
+                if (a.status.value if hasattr(a.status, "value") else a.status) == "ready"
+                and a.mime_type
+                and a.mime_type.startswith("video")
+            ]
+            source_desc = f"素材库 {video_lib.name}"
+
+    # 策略 2: 通过 user_id 查找用户上传的素材
+    if not ready_videos and user_id and hasattr(asset_repo, "find_ready_videos_by_user"):
+        logger.info(
+            "模板编辑器自动兜底4: plan=%s project_id 为空，尝试通过 user_id=%s 查找素材",
+            plan_id,
+            user_id,
+        )
+        ready_videos = asset_repo.find_ready_videos_by_user(user_id)
+        source_desc = f"用户上传 (user_id={user_id[:8]}...)"
+
+    if not ready_videos:
+        logger.warning(
+            "模板编辑器自动兜底4: plan=%s 未找到可用素材 (project_id=%s, user_id=%s)",
+            plan_id,
+            plan_check.project_id or "(empty)",
+            user_id[:8] + "..." if user_id else "(empty)",
+        )
         return
 
     logger.info(
-        "模板编辑器自动兜底4: plan=%s 自动选素材分配给 %d 个无素材片段",
+        "模板编辑器自动兜底4: plan=%s 自动选素材分配给 %d 个无素材片段 (来源: %s, 共 %d 个)",
         plan_id,
         len(clips_without_asset),
+        source_desc,
+        len(ready_videos),
+    )
+    random.shuffle(ready_videos)
+    for i, clip in enumerate(clips_without_asset):
+        asset = ready_videos[i % len(ready_videos)]
+        svc.assign_asset(clip.id, asset.id)
+    logger.info(
+        "模板编辑器自动兜底4: plan=%s 从 %s 分配了 %d 个素材给 %d 个片段",
+        plan_id,
+        source_desc,
+        len(ready_videos),
+        len(clips_without_asset),
     )
-    libs = asset_library_repo.find_by_project(plan_check.project_id)
-    video_lib = None
-    for lib in libs:
-        lib_kind = lib.kind.value if hasattr(lib.kind, "value") else lib.kind
-        if lib_kind == "video":
-            video_lib = lib
-            break
-
-    if video_lib:
-        assets = asset_repo.find_by_library(video_lib.id)
-        ready_videos = [
-            a
-            for a in assets
-            if (a.status.value if hasattr(a.status, "value") else a.status) == "ready"
-            and a.mime_type
-            and a.mime_type.startswith("video")
-        ]
-        if ready_videos:
-            random.shuffle(ready_videos)
-            for i, clip in enumerate(clips_without_asset):
-                asset = ready_videos[i % len(ready_videos)]
-                svc.assign_asset(clip.id, asset.id)
-            logger.info(
-                "模板编辑器自动兜底4: plan=%s 从素材库 %s 分配了 %d 个素材",
-                plan_id,
-                video_lib.name,
-                len(ready_videos),
-            )
diff --git a/apps/api/app/api/routes/templates_editor/generation.py b/apps/api/app/api/routes/templates_editor/generation.py
index c9bda4ce1..b4e152290 100755
--- a/apps/api/app/api/routes/templates_editor/generation.py
+++ b/apps/api/app/api/routes/templates_editor/generation.py
@@ -71,20 +71,22 @@ def generate_editor_draft(
     _auto_fallback_copy_template_clips(plan_svc, plan_id, plan_check, db)
     clips_without_asset = _auto_fallback_assign_assets(plan_svc, plan_id, plan_check)
     _auto_fallback_auto_material_mode(
-        plan_svc, plan_id, plan_check, clips_without_asset, asset_library_repo, asset_repo
+        plan_svc,
+        plan_id,
+        plan_check,
+        clips_without_asset,
+        asset_library_repo,
+        asset_repo,
+        user_id=str(current_user.user.id),
     )
 
     # 检查是否可生成（含最后防线自动修复 + 诊断日志）
     try:
         can_gen, reason = plan_svc.can_generate(plan_id)
     except ValueError as exc:
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)
-        ) from exc
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
     if not can_gen:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST, detail=reason
-        )
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=reason)
 
     try:
         clip_count = plan_svc.mark_clips_ready(plan_id)
@@ -123,9 +125,7 @@ def generate_editor_draft(
 
         return EditPlanGenerateResponse(
             plan_id=plan_id,
-            plan_status=updated_plan.status.value
-            if hasattr(updated_plan.status, "value")
-            else updated_plan.status,
+            plan_status=updated_plan.status.value if hasattr(updated_plan.status, "value") else updated_plan.status,
             generation_task_id=gen_task.id,
             clip_count=clip_count,
         )
@@ -160,9 +160,7 @@ def get_editor_generation_status(
     try:
         gen_status = plan_svc.get_generation_status(plan_id)
     except ValueError as exc:
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)
-        ) from exc
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
 
     plan = gen_status["plan"]
     clips = gen_status["clips"]
@@ -184,21 +182,15 @@ def get_editor_generation_status(
     video_url = ""
     if raw_video_url:
         try:
-            video_url = storage_service.get_download_url(
-                raw_video_url, expires_seconds=86400
-            )
+            video_url = storage_service.get_download_url(raw_video_url, expires_seconds=86400)
         except Exception as e:
-            logger.warning(
-                "生成视频签名URL失败: template_id=%s error=%s", template_id, e
-            )
+            logger.warning("生成视频签名URL失败: template_id=%s error=%s", template_id, e)
             video_url = raw_video_url
 
     progress = gen_status.get("progress", 0.0)
     error_message = gen_status.get("error_message", "")
     gen_task_status = gen_status.get("generation_task_status")
-    plan_status_val = (
-        plan.status.value if hasattr(plan.status, "value") else plan.status
-    )
+    plan_status_val = plan.status.value if hasattr(plan.status, "value") else plan.status
     if plan_status_val == "completed" and progress < 100:
         progress = 100.0
 
diff --git a/packages/adapters/sqlalchemy_impl/asset_repository.py b/packages/adapters/sqlalchemy_impl/asset_repository.py
index a5abe8bed..0c4077c14 100755
--- a/packages/adapters/sqlalchemy_impl/asset_repository.py
+++ b/packages/adapters/sqlalchemy_impl/asset_repository.py
@@ -284,6 +284,24 @@ class SQLAlchemyAssetRepository:
         )
         return int(result or 0)
 
+    def find_ready_videos_by_user(
+        self,
+        user_id: str,
+        *,
+        limit: int = 50,
+    ) -> list[Asset]:
+        """查找用户上传的所有就绪视频素材。"""
+        query = self.session.query(AssetModel).filter(
+            AssetModel.uploaded_by_user_id == user_id,
+            AssetModel.status == "ready",
+            AssetModel.file_type == "video",
+        )
+        query = query.order_by(AssetModel.created_at.desc())
+        if limit > 0:
+            query = query.limit(limit)
+        models = query.all()
+        return [self._to_domain(m) for m in models]
+
     def search_candidates(
         self,
         project_id: str,
diff --git a/tests/unit/test_fallback_user_assets.py b/tests/unit/test_fallback_user_assets.py
new file mode 100644
index 000000000..6624e07ac
--- /dev/null
+++ b/tests/unit/test_fallback_user_assets.py
@@ -0,0 +1,331 @@
+"""Unit tests for PR #1338: 确认生成兜底增强 — user_id 查找素材.
+
+覆盖:
+- SQLAlchemyAssetRepository.find_ready_videos_by_user
+- _auto_fallback_auto_material_mode 策略2 (user_id 兜底)
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "packages"))
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
+from packages.adapters.sqlalchemy_impl.models import Base
+from packages.domain import Asset, AssetStatus
+
+
+def _make_repo():
+    engine = create_engine("sqlite:///:memory:")
+    Base.metadata.create_all(engine)
+    session = sessionmaker(bind=engine)()
+    return SQLAlchemyAssetRepository(session)
+
+
+class TestFindReadyVideosByUser:
+    """SQLAlchemyAssetRepository.find_ready_videos_by_user 测试."""
+
+    def test_returns_ready_videos_for_user(self):
+        repo = _make_repo()
+        user_id = "user-abc-123"
+        v1 = Asset.create(
+            project_id="",
+            library_id="lib-1",
+            name="video1.mp4",
+            storage_key="v/v1.mp4",
+            mime_type="video/mp4",
+            status=AssetStatus.READY,
+            uploaded_by_user_id=user_id,
+        )
+        v2 = Asset.create(
+            project_id="",
+            library_id="lib-1",
+            name="video2.mp4",
+            storage_key="v/v2.mp4",
+            mime_type="video/mp4",
+            status=AssetStatus.READY,
+            uploaded_by_user_id=user_id,
+        )
+        repo.create(v1)
+        repo.create(v2)
+        result = repo.find_ready_videos_by_user(user_id)
+        assert len(result) == 2
+        assert {a.id for a in result} == {v1.id, v2.id}
+
+    def test_excludes_non_video_assets(self):
+        repo = _make_repo()
+        user_id = "user-abc-123"
+        video = Asset.create(
+            project_id="",
+            library_id="lib-1",
+            name="clip.mp4",
+            storage_key="v/clip.mp4",
+            mime_type="video/mp4",
+            status=AssetStatus.READY,
+            uploaded_by_user_id=user_id,
+        )
+        image = Asset.create(
+            project_id="",
+            library_id="lib-1",
+            name="photo.jpg",
+            storage_key="v/photo.jpg",
+            mime_type="image/jpeg",
+            status=AssetStatus.READY,
+            uploaded_by_user_id=user_id,
+        )
+        repo.create(video)
+        repo.create(image)
+        result = repo.find_ready_videos_by_user(user_id)
+        assert len(result) == 1
+        assert result[0].id == video.id
+
+    def test_excludes_non_ready_assets(self):
+        repo = _make_repo()
+        user_id = "user-abc-123"
+        ready = Asset.create(
+            project_id="",
+            library_id="lib-1",
+            name="ready.mp4",
+            storage_key="v/ready.mp4",
+            mime_type="video/mp4",
+            status=AssetStatus.READY,
+            uploaded_by_user_id=user_id,
+        )
+        uploading = Asset.create(
+            project_id="",
+            library_id="lib-1",
+            name="uploading.mp4",
+            storage_key="v/uploading.mp4",
+            mime_type="video/mp4",
+            status=AssetStatus.UPLOADING,
+            uploaded_by_user_id=user_id,
+        )
+        repo.create(ready)
+        repo.create(uploading)
+        result = repo.find_ready_videos_by_user(user_id)
+        assert len(result) == 1
+        assert result[0].id == ready.id
+
+    def test_excludes_other_users_assets(self):
+        repo = _make_repo()
+        my_video = Asset.create(
+            project_id="",
+            library_id="lib-1",
+            name="mine.mp4",
+            storage_key="v/mine.mp4",
+            mime_type="video/mp4",
+            status=AssetStatus.READY,
+            uploaded_by_user_id="user-A",
+        )
+        other_video = Asset.create(
+            project_id="",
+            library_id="lib-1",
+            name="other.mp4",
+            storage_key="v/other.mp4",
+            mime_type="video/mp4",
+            status=AssetStatus.READY,
+            uploaded_by_user_id="user-B",
+        )
+        repo.create(my_video)
+        repo.create(other_video)
+        result = repo.find_ready_videos_by_user("user-A")
+        assert len(result) == 1
+        assert result[0].id == my_video.id
+
+    def test_empty_result_for_unknown_user(self):
+        repo = _make_repo()
+        result = repo.find_ready_videos_by_user("nonexistent-user")
+        assert result == []
+
+    def test_respects_limit(self):
+        repo = _make_repo()
+        user_id = "user-abc-123"
+        for i in range(10):
+            asset = Asset.create(
+                project_id="",
+                library_id="lib-1",
+                name=f"video_{i}.mp4",
+                storage_key=f"v/v{i}.mp4",
+                mime_type="video/mp4",
+                status=AssetStatus.READY,
+                uploaded_by_user_id=user_id,
+            )
+            repo.create(asset)
+        result = repo.find_ready_videos_by_user(user_id, limit=3)
+        assert len(result) == 3
+
+
+class TestAutoFallbackAutoMaterialModeUserId:
+    """_auto_fallback_auto_material_mode user_id 兜底策略测试."""
+
+    def _make_plan_check(self, project_id="", template_id="tmpl-1"):
+        plan = MagicMock()
+        plan.project_id = project_id
+        plan.template_id = template_id
+        plan.config = {}
+        return plan
+
+    def _make_clip(self, clip_id="clip-1"):
+        clip = MagicMock()
+        clip.id = clip_id
+        clip.asset_id = ""
+        return clip
+
+    def test_skips_when_no_clips_without_asset(self):
+        from app.api.routes.templates_editor._fallback import (
+            _auto_fallback_auto_material_mode,
+        )
+
+        svc = MagicMock()
+        plan_check = self._make_plan_check()
+        _auto_fallback_auto_material_mode(
+            svc,
+            "plan-1",
+            plan_check,
+            [],
+            MagicMock(),
+            MagicMock(),
+            user_id="user-1",
+        )
+        svc.assign_asset.assert_not_called()
+
+    def test_strategy2_user_id_fallback(self):
+        from app.api.routes.templates_editor._fallback import (
+            _auto_fallback_auto_material_mode,
+        )
+
+        svc = MagicMock()
+        plan_check = self._make_plan_check(project_id="")
+        clip = self._make_clip("clip-1")
+        mock_asset = MagicMock()
+        mock_asset.id = "asset-from-user"
+        mock_asset.status = AssetStatus.READY
+        mock_asset.mime_type = "video/mp4"
+        asset_repo = MagicMock()
+        asset_repo.find_ready_videos_by_user.return_value = [mock_asset]
+        asset_library_repo = MagicMock()
+        _auto_fallback_auto_material_mode(
+            svc,
+            "plan-1",
+            plan_check,
+            [clip],
+            asset_library_repo,
+            asset_repo,
+            user_id="user-123",
+        )
+        asset_repo.find_ready_videos_by_user.assert_called_once_with("user-123")
+        svc.assign_asset.assert_called_once_with("clip-1", "asset-from-user")
+
+    def test_strategy1_takes_priority_over_strategy2(self):
+        from app.api.routes.templates_editor._fallback import (
+            _auto_fallback_auto_material_mode,
+        )
+
+        svc = MagicMock()
+        plan_check = self._make_plan_check(project_id="proj-1")
+        clip = self._make_clip("clip-1")
+        mock_lib = MagicMock()
+        mock_lib.id = "lib-video"
+        mock_lib.kind = MagicMock()
+        mock_lib.kind.value = "video"
+        mock_asset = MagicMock()
+        mock_asset.id = "asset-from-project"
+        mock_asset.status = "ready"
+        mock_asset.mime_type = "video/mp4"
+        asset_library_repo = MagicMock()
+        asset_library_repo.find_by_project.return_value = [mock_lib]
+        asset_repo = MagicMock()
+        asset_repo.find_by_library.return_value = [mock_asset]
+        _auto_fallback_auto_material_mode(
+            svc,
+            "plan-1",
+            plan_check,
+            [clip],
+            asset_library_repo,
+            asset_repo,
+            user_id="user-123",
+        )
+        asset_library_repo.find_by_project.assert_called_once_with("proj-1")
+        asset_repo.find_ready_videos_by_user.assert_not_called()
+        svc.assign_asset.assert_called_once_with("clip-1", "asset-from-project")
+
+    def test_falls_back_when_project_has_no_videos(self):
+        from app.api.routes.templates_editor._fallback import (
+            _auto_fallback_auto_material_mode,
+        )
+
+        svc = MagicMock()
+        plan_check = self._make_plan_check(project_id="proj-1")
+        clip = self._make_clip("clip-1")
+        asset_library_repo = MagicMock()
+        asset_library_repo.find_by_project.return_value = []
+        mock_asset = MagicMock()
+        mock_asset.id = "asset-from-user"
+        mock_asset.status = AssetStatus.READY
+        mock_asset.mime_type = "video/mp4"
+        asset_repo = MagicMock()
+        asset_repo.find_ready_videos_by_user.return_value = [mock_asset]
+        _auto_fallback_auto_material_mode(
+            svc,
+            "plan-1",
+            plan_check,
+            [clip],
+            asset_library_repo,
+            asset_repo,
+            user_id="user-123",
+        )
+        asset_repo.find_ready_videos_by_user.assert_called_once_with("user-123")
+        svc.assign_asset.assert_called_once_with("clip-1", "asset-from-user")
+
+    def test_no_assets_found_does_nothing(self):
+        from app.api.routes.templates_editor._fallback import (
+            _auto_fallback_auto_material_mode,
+        )
+
+        svc = MagicMock()
+        plan_check = self._make_plan_check(project_id="")
+        clip = self._make_clip("clip-1")
+        asset_repo = MagicMock()
+        asset_repo.find_ready_videos_by_user.return_value = []
+        asset_library_repo = MagicMock()
+        _auto_fallback_auto_material_mode(
+            svc,
+            "plan-1",
+            plan_check,
+            [clip],
+            asset_library_repo,
+            asset_repo,
+            user_id="user-123",
+        )
+        svc.assign_asset.assert_not_called()
+
+    def test_no_user_id_skips_strategy2(self):
+        from app.api.routes.templates_editor._fallback import (
+            _auto_fallback_auto_material_mode,
+        )
+
+        svc = MagicMock()
+        plan_check = self._make_plan_check(project_id="")
+        clip = self._make_clip("clip-1")
+        asset_repo = MagicMock()
+        asset_library_repo = MagicMock()
+        _auto_fallback_auto_material_mode(
+            svc,
+            "plan-1",
+            plan_check,
+            [clip],
+            asset_library_repo,
+            asset_repo,
+            user_id="",
+        )
+        asset_repo.find_ready_videos_by_user.assert_not_called()
+        svc.assign_asset.assert_not_called()