diff --git a/apps/api/app/api/routes/edit_plans.py b/apps/api/app/api/routes/edit_plans.py
index de0a5c5bf..ec347a32a 100755
--- a/apps/api/app/api/routes/edit_plans.py
+++ b/apps/api/app/api/routes/edit_plans.py
@@ -50,6 +50,7 @@ class EditPlanCreateRequest(BaseModel):
     config: dict[str, Any] = Field(default_factory=dict, description="计划配置 (JSON)")
     total_duration: float = Field(default=0.0, ge=0.0, description="总时长 (秒)")
     project_id: str = Field(default="", description="所属项目 ID")
+    asset_ids: List[str] = Field(default_factory=list, description="素材 ID 列表，创建时自动分配给片段")
 
 
 class EditPlanUpdateRequest(BaseModel):
@@ -419,7 +420,7 @@ def create_plan(
         result = generator.generate_from_template(
             template=template,
             clip_configs=clip_configs,
-            asset_ids=[],
+            asset_ids=body.asset_ids,
             project_id=project_id,
             created_by_user_id=current_user.user.id,
             name=body.name,
@@ -445,6 +446,13 @@ def create_plan(
             total_duration=body.total_duration if body.total_duration > 0 else None,
         )
 
+    # 把 asset_ids 写入 plan.config，供生成时兜底分配使用
+    if body.asset_ids:
+        current_config = plan.config or {}
+        if current_config.get("asset_ids") != body.asset_ids:
+            current_config["asset_ids"] = body.asset_ids
+            plan = svc.update_plan(plan.id, config=normalize_plan_config(current_config))
+
     logger.info(
         "创建剪辑计划: id=%s name=%s clips=%d by user=%s",
         plan.id,
diff --git a/apps/worker/video_processing/oss_helpers.py b/apps/worker/video_processing/oss_helpers.py
index 3e486ed51..6ba88ac7a 100755
--- a/apps/worker/video_processing/oss_helpers.py
+++ b/apps/worker/video_processing/oss_helpers.py
@@ -15,6 +15,8 @@ from urllib.parse import urlparse
 
 import oss2
 
+from packages.shared.config import get_shared_settings
+
 logger = logging.getLogger(__name__)
 
 # OSS 上传配置
@@ -31,14 +33,18 @@ OSS_MULTIPART_NUM_THREADS = 3  # 分片上传并发数
 def oss_settings() -> tuple[str, str, str, str] | None:
     """获取 OSS 配置。
 
+    统一使用 SharedSettings 读取配置，与 SharedStorageService 保持一致，
+    支持从 .env 文件加载，避免两套配置路径不一致。
+
     Returns:
         (access_key_id, access_key_secret, endpoint, bucket_name) 元组，
         配置缺失时返回 None。
     """
-    access_key_id = os.getenv("OSS_ACCESS_KEY_ID")
-    access_key_secret = os.getenv("OSS_ACCESS_KEY_SECRET")
-    endpoint = os.getenv("OSS_ENDPOINT")
-    bucket_name = os.getenv("OSS_BUCKET_NAME")
+    settings = get_shared_settings()
+    access_key_id = settings.oss_access_key_id
+    access_key_secret = settings.oss_access_key_secret
+    endpoint = settings.oss_endpoint
+    bucket_name = settings.oss_bucket_name
     if not all([access_key_id, access_key_secret, endpoint, bucket_name]):
         return None
     return access_key_id, access_key_secret, endpoint, bucket_name
diff --git a/apps/worker/video_processing/render_adapter.py b/apps/worker/video_processing/render_adapter.py
index c6611d814..46ca718c1 100755
--- a/apps/worker/video_processing/render_adapter.py
+++ b/apps/worker/video_processing/render_adapter.py
@@ -27,6 +27,7 @@ from video_processing.unified_render_service import UnifiedRenderService
 
 from packages.adapters.sqlalchemy_impl.edit_plan_clip_repository import SQLAlchemyEditPlanClipRepository
 from packages.adapters.sqlalchemy_impl.edit_plan_repository import SQLAlchemyEditPlanRepository
+from packages.adapters.sqlalchemy_impl.models import AssetModel
 from packages.domain.edit_plan import EditPlanStatus
 from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus
 
@@ -294,10 +295,15 @@ class RenderAdapter:
             except Exception:
                 logger.exception("进度回调失败")
 
-    @staticmethod
-    def _download_assets(clips: list[EditPlanClip], work_dir: Path) -> tuple[dict[str, Path], list[str], list[str]]:
+    def _download_assets(
+        self, clips: list[EditPlanClip], work_dir: Path
+    ) -> tuple[dict[str, Path], list[str], list[str]]:
         """下载片段素材到本地。
 
+        先通过 asset_id 批量查询 assets 表获取 file_url（OSS存储路径），
+        再用 file_url 作为 OSS key 下载。asset_id 是 UUID 主键，
+        不能直接当作 OSS 存储路径使用。
+
         Returns:
             (asset_path_map, rendered_clip_ids, failed_clip_ids)
             - asset_path_map: asset_id → local_path 映射（下载成功的）
@@ -312,6 +318,13 @@ class RenderAdapter:
         failed_clip_ids: list[str] = []
         seen_asset_ids: set[str] = set()
 
+        # 批量查询素材的 file_url（OSS 存储路径）
+        clip_asset_ids = [c.asset_id for c in clips if c.asset_id]
+        asset_storage_map: dict[str, str] = {}
+        if clip_asset_ids:
+            assets = self._db.query(AssetModel).filter(AssetModel.id.in_(clip_asset_ids)).all()
+            asset_storage_map = {a.id: a.file_url for a in assets if a.file_url}
+
         for clip in clips:
             asset_id = clip.asset_id
             if not asset_id:
@@ -328,11 +341,23 @@ class RenderAdapter:
 
             seen_asset_ids.add(asset_id)
 
-            # 生成安全的本地文件名
-            safe_name = f"clip_{clip.order:04d}_{abs(hash(asset_id)) % 100000:05d}.mp4"
+            # 从素材表获取 OSS 存储路径（file_url）
+            storage_key = asset_storage_map.get(asset_id)
+            if not storage_key:
+                logger.warning(
+                    "素材无 file_url，跳过下载: clip_id=%s asset_id=%s",
+                    clip.id,
+                    asset_id,
+                )
+                failed_clip_ids.append(clip.id)
+                continue
+
+            # 生成安全的本地文件名（保留原始扩展名）
+            ext = Path(storage_key).suffix or ".mp4"
+            safe_name = f"clip_{clip.order:04d}_{abs(hash(asset_id)) % 100000:05d}{ext}"
             local_path = asset_dir / safe_name
 
-            if download_asset(asset_id, local_path):
+            if download_asset(storage_key, local_path):
                 asset_path_map[asset_id] = local_path
                 rendered_clip_ids.append(clip.id)
                 logger.debug("素材下载成功: clip_id=%s asset_id=%s", clip.id, asset_id[:60])
diff --git a/tests/unit/test_edit_plans_api.py b/tests/unit/test_edit_plans_api.py
index 3314a3ce8..5e44e6575 100755
--- a/tests/unit/test_edit_plans_api.py
+++ b/tests/unit/test_edit_plans_api.py
@@ -238,6 +238,47 @@ class TestCreatePlan:
         assert call_kwargs["name"] == "我的剪辑计划"
         assert call_kwargs["created_by_user_id"] == "user-001"
 
+    @patch("app.api.routes.edit_plans.EditTemplateService")
+    @patch("app.services.PlanGeneratorService")
+    def test_create_with_asset_ids(self, mock_generator_cls, mock_template_svc_cls, client):
+        """创建计划时传入 asset_ids，应传递给生成器并写入 plan.config"""
+        c, repo = client
+
+        mock_template_svc = MagicMock()
+        mock_template_svc.get_template_or_raise.return_value = MagicMock(
+            id="tpl-001",
+            name="测试模板",
+            config={},
+        )
+        mock_template_svc.list_clip_configs.return_value = []
+        mock_template_svc_cls.return_value = mock_template_svc
+
+        mock_gen = MagicMock()
+        test_plan = self._make_test_plan(name="带素材计划")
+        test_plan.status = EditPlanStatus.EDITING
+        repo.create(test_plan)
+        mock_gen.generate_from_template.return_value = {
+            "plan": test_plan,
+            "clips": [],
+        }
+        mock_generator_cls.return_value = mock_gen
+
+        asset_ids = ["asset-001", "asset-002", "asset-003"]
+        resp = c.post(
+            "/api/v1/edit-plans",
+            json={
+                "template_id": "tpl-001",
+                "name": "带素材计划",
+                "asset_ids": asset_ids,
+            },
+        )
+        assert resp.status_code == 201
+
+        # 验证 asset_ids 传递给了生成器
+        mock_gen.generate_from_template.assert_called_once()
+        call_kwargs = mock_gen.generate_from_template.call_args[1]
+        assert call_kwargs["asset_ids"] == asset_ids
+
     @patch("app.api.routes.edit_plans.EditTemplateService")
     @patch("app.services.PlanGeneratorService")
     def test_create_minimal(self, mock_generator_cls, mock_template_svc_cls, client):
diff --git a/tests/unit/test_oss_upload_crash_fix.py b/tests/unit/test_oss_upload_crash_fix.py
index 806e9b4cf..36d868b7f 100755
--- a/tests/unit/test_oss_upload_crash_fix.py
+++ b/tests/unit/test_oss_upload_crash_fix.py
@@ -9,7 +9,6 @@
 
 from __future__ import annotations
 
-import os
 import tempfile
 import time
 from pathlib import Path
@@ -27,14 +26,9 @@ class TestOSSBucketConnectTimeout:
 
         mock_bucket_instance = MagicMock()
         with (
-            patch.dict(
-                os.environ,
-                {
-                    "OSS_ACCESS_KEY_ID": "test-key",
-                    "OSS_ACCESS_KEY_SECRET": "test-secret",
-                    "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                    "OSS_BUCKET_NAME": "test-bucket",
-                },
+            patch(
+                "video_processing.oss_helpers.oss_settings",
+                return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
             ),
             patch("video_processing.oss_helpers.oss2.Auth"),
             patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls,
@@ -53,7 +47,7 @@ class TestOSSBucketConnectTimeout:
         """OSS 配置缺失时返回 None."""
         from video_processing.oss_helpers import oss_bucket
 
-        with patch.dict(os.environ, {}, clear=True):
+        with patch("video_processing.oss_helpers.oss_settings", return_value=None):
             bucket = oss_bucket()
             assert bucket is None
 
@@ -80,14 +74,9 @@ class TestUploadToOSSMultipart:
             mock_bucket = MagicMock()
 
             with (
-                patch.dict(
-                    os.environ,
-                    {
-                        "OSS_ACCESS_KEY_ID": "test-key",
-                        "OSS_ACCESS_KEY_SECRET": "test-secret",
-                        "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                        "OSS_BUCKET_NAME": "test-bucket",
-                    },
+                patch(
+                    "video_processing.oss_helpers.oss_settings",
+                    return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
                 ),
                 patch("video_processing.oss_helpers.oss2.Auth"),
                 patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
@@ -113,14 +102,9 @@ class TestUploadToOSSMultipart:
             mock_bucket = MagicMock()
 
             with (
-                patch.dict(
-                    os.environ,
-                    {
-                        "OSS_ACCESS_KEY_ID": "test-key",
-                        "OSS_ACCESS_KEY_SECRET": "test-secret",
-                        "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                        "OSS_BUCKET_NAME": "test-bucket",
-                    },
+                patch(
+                    "video_processing.oss_helpers.oss_settings",
+                    return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
                 ),
                 patch("video_processing.oss_helpers.oss2.Auth"),
                 patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
@@ -167,14 +151,9 @@ class TestUploadToOSSTimeout:
 
         try:
             with (
-                patch.dict(
-                    os.environ,
-                    {
-                        "OSS_ACCESS_KEY_ID": "test-key",
-                        "OSS_ACCESS_KEY_SECRET": "test-secret",
-                        "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                        "OSS_BUCKET_NAME": "test-bucket",
-                    },
+                patch(
+                    "video_processing.oss_helpers.oss_settings",
+                    return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
                 ),
                 patch("video_processing.oss_helpers.oss2.Auth"),
                 patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
@@ -200,14 +179,9 @@ class TestUploadToOSSTimeout:
 
         try:
             with (
-                patch.dict(
-                    os.environ,
-                    {
-                        "OSS_ACCESS_KEY_ID": "test-key",
-                        "OSS_ACCESS_KEY_SECRET": "test-secret",
-                        "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                        "OSS_BUCKET_NAME": "test-bucket",
-                    },
+                patch(
+                    "video_processing.oss_helpers.oss_settings",
+                    return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
                 ),
                 patch("video_processing.oss_helpers.oss2.Auth"),
                 patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
@@ -227,7 +201,7 @@ class TestUploadToOSSTimeout:
         file_path = Path(small_file.name)
 
         try:
-            with patch.dict(os.environ, {}, clear=True):
+            with patch("video_processing.oss_helpers.oss_settings", return_value=None):
                 url = upload_to_oss(file_path, "test/noconfig.mp4")
                 assert url is None
         finally:
diff --git a/tests/unit/test_p02_worker_oss_fix.py b/tests/unit/test_p02_worker_oss_fix.py
index 58f7d37fd..f9c12826b 100755
--- a/tests/unit/test_p02_worker_oss_fix.py
+++ b/tests/unit/test_p02_worker_oss_fix.py
@@ -8,7 +8,7 @@
 
 from __future__ import annotations
 
-import os
+from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 # ── oss_bucket endpoint scheme 修复 ──────────────────────────────────────────
@@ -23,20 +23,13 @@ class TestOSSBucketEndpointScheme:
 
         mock_bucket_instance = MagicMock()
         with (
-            patch.dict(
-                os.environ,
-                {
-                    "OSS_ACCESS_KEY_ID": "test-key",
-                    "OSS_ACCESS_KEY_SECRET": "test-secret",
-                    "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                    "OSS_BUCKET_NAME": "test-bucket",
-                },
+            patch(
+                "video_processing.oss_helpers.oss_settings",
+                return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
             ),
             patch("video_processing.oss_helpers.oss2.Auth"),
             patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls,
         ):
-            # 清除缓存，确保重新创建
-
             bucket = oss_bucket()
 
             assert bucket is mock_bucket_instance
@@ -52,19 +45,13 @@ class TestOSSBucketEndpointScheme:
 
         mock_bucket_instance = MagicMock()
         with (
-            patch.dict(
-                os.environ,
-                {
-                    "OSS_ACCESS_KEY_ID": "test-key",
-                    "OSS_ACCESS_KEY_SECRET": "test-secret",
-                    "OSS_ENDPOINT": "https://oss-cn-hangzhou.aliyuncs.com",
-                    "OSS_BUCKET_NAME": "test-bucket",
-                },
+            patch(
+                "video_processing.oss_helpers.oss_settings",
+                return_value=("test-key", "test-secret", "https://oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
             ),
             patch("video_processing.oss_helpers.oss2.Auth"),
             patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls,
         ):
-
             oss_bucket()
 
             call_args = mock_bucket_cls.call_args
@@ -79,19 +66,13 @@ class TestOSSBucketEndpointScheme:
 
         mock_bucket_instance = MagicMock()
         with (
-            patch.dict(
-                os.environ,
-                {
-                    "OSS_ACCESS_KEY_ID": "test-key",
-                    "OSS_ACCESS_KEY_SECRET": "test-secret",
-                    "OSS_ENDPOINT": "http://oss-cn-hangzhou.aliyuncs.com",
-                    "OSS_BUCKET_NAME": "test-bucket",
-                },
+            patch(
+                "video_processing.oss_helpers.oss_settings",
+                return_value=("test-key", "test-secret", "http://oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
             ),
             patch("video_processing.oss_helpers.oss2.Auth"),
             patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls,
         ):
-
             oss_bucket()
 
             call_args = mock_bucket_cls.call_args
@@ -102,17 +83,7 @@ class TestOSSBucketEndpointScheme:
         """凭证缺失时返回 None."""
         from video_processing.oss_helpers import oss_bucket
 
-        with patch.dict(
-            os.environ,
-            {
-                "OSS_ACCESS_KEY_ID": "",
-                "OSS_ACCESS_KEY_SECRET": "",
-                "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                "OSS_BUCKET_NAME": "test-bucket",
-            },
-            clear=True,
-        ):
-
+        with patch("video_processing.oss_helpers.oss_settings", return_value=None):
             bucket = oss_bucket()
             assert bucket is None
 
@@ -128,17 +99,15 @@ class TestGetSignedDownloadUrl:
         from video_processing.oss_helpers import get_signed_download_url
 
         mock_bucket = MagicMock()
-        mock_bucket.sign_url.return_value = "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4?OSSAccessKeyId=xxx&Expires=xxx&Signature=xxx"
+        mock_bucket.sign_url.return_value = (
+            "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4"
+            "?OSSAccessKeyId=xxx&Expires=xxx&Signature=xxx"
+        )
 
         with (
-            patch.dict(
-                os.environ,
-                {
-                    "OSS_ACCESS_KEY_ID": "test-key",
-                    "OSS_ACCESS_KEY_SECRET": "test-secret",
-                    "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                    "OSS_BUCKET_NAME": "test-bucket",
-                },
+            patch(
+                "video_processing.oss_helpers.oss_settings",
+                return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
             ),
             patch("video_processing.oss_helpers.oss2.Auth"),
             patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
@@ -159,14 +128,9 @@ class TestGetSignedDownloadUrl:
         )
 
         with (
-            patch.dict(
-                os.environ,
-                {
-                    "OSS_ACCESS_KEY_ID": "test-key",
-                    "OSS_ACCESS_KEY_SECRET": "test-secret",
-                    "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                    "OSS_BUCKET_NAME": "test-bucket",
-                },
+            patch(
+                "video_processing.oss_helpers.oss_settings",
+                return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
             ),
             patch("video_processing.oss_helpers.oss2.Auth"),
             patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
@@ -183,7 +147,7 @@ class TestGetSignedDownloadUrl:
         """bucket 为 None 时返回 None（不抛异常）."""
         from video_processing.oss_helpers import get_signed_download_url
 
-        with patch.dict(os.environ, {}, clear=True):
+        with patch("video_processing.oss_helpers.oss_settings", return_value=None):
             result = get_signed_download_url("generated/test.mp4")
             assert result is None
 
@@ -195,14 +159,9 @@ class TestGetSignedDownloadUrl:
         mock_bucket.sign_url.side_effect = Exception("sign failed")
 
         with (
-            patch.dict(
-                os.environ,
-                {
-                    "OSS_ACCESS_KEY_ID": "test-key",
-                    "OSS_ACCESS_KEY_SECRET": "test-secret",
-                    "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                    "OSS_BUCKET_NAME": "test-bucket",
-                },
+            patch(
+                "video_processing.oss_helpers.oss_settings",
+                return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
             ),
             patch("video_processing.oss_helpers.oss2.Auth"),
             patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
@@ -224,17 +183,10 @@ class TestUploadToOSSReturnsHTTPS:
         mock_bucket = MagicMock()
         mock_bucket.put_object_from_file = MagicMock()
 
-        from pathlib import Path
-
         with (
-            patch.dict(
-                os.environ,
-                {
-                    "OSS_ACCESS_KEY_ID": "test-key",
-                    "OSS_ACCESS_KEY_SECRET": "test-secret",
-                    "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                    "OSS_BUCKET_NAME": "test-bucket",
-                },
+            patch(
+                "video_processing.oss_helpers.oss_settings",
+                return_value=("test-key", "test-secret", "oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
             ),
             patch("video_processing.oss_helpers.oss2.Auth"),
             patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
@@ -252,17 +204,10 @@ class TestUploadToOSSReturnsHTTPS:
         mock_bucket = MagicMock()
         mock_bucket.put_object_from_file = MagicMock()
 
-        from pathlib import Path
-
         with (
-            patch.dict(
-                os.environ,
-                {
-                    "OSS_ACCESS_KEY_ID": "test-key",
-                    "OSS_ACCESS_KEY_SECRET": "test-secret",
-                    "OSS_ENDPOINT": "https://oss-cn-hangzhou.aliyuncs.com",
-                    "OSS_BUCKET_NAME": "test-bucket",
-                },
+            patch(
+                "video_processing.oss_helpers.oss_settings",
+                return_value=("test-key", "test-secret", "https://oss-cn-hangzhou.aliyuncs.com", "test-bucket"),
             ),
             patch("video_processing.oss_helpers.oss2.Auth"),
             patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
diff --git a/tests/unit/test_render_adapter.py b/tests/unit/test_render_adapter.py
index e8dfd8d9d..034886911 100755
--- a/tests/unit/test_render_adapter.py
+++ b/tests/unit/test_render_adapter.py
@@ -70,13 +70,33 @@ def _make_clip(
 def _make_adapter(
     plan: FakePlan | None = None,
     clips: list[FakeClip] | None = None,
+    asset_url_map: dict[str, str] | None = None,
 ) -> tuple[RenderAdapter, MagicMock, MagicMock]:
     """创建测试用的 RenderAdapter 及 mock repo。
 
+    Args:
+        plan: 模拟的剪辑计划
+        clips: 模拟的片段列表
+        asset_url_map: asset_id → file_url 映射，用于 mock assets 表查询
+
     Returns:
         (adapter, mock_plan_repo, mock_clip_repo)
     """
     mock_db = MagicMock()
+
+    # mock assets 表查询：_download_assets 用
+    if asset_url_map is not None:
+        mock_query = MagicMock()
+        mock_assets = []
+        for aid, url in asset_url_map.items():
+            m = MagicMock()
+            m.id = aid
+            m.file_url = url
+            mock_assets.append(m)
+        mock_query.all.return_value = mock_assets
+        mock_query.filter.return_value = mock_query
+        mock_db.query.return_value = mock_query
+
     adapter = RenderAdapter(mock_db)
 
     # 替换内部 repo
@@ -229,7 +249,8 @@ class TestRenderPlan:
 
         plan = FakePlan(id="plan_001", status="editing")
         clips = [_make_clip("c1", order=0, duration=5.0)]
-        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+        asset_url_map = {"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4"}
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)
 
         result = adapter.render_plan("plan_001")
 
@@ -243,7 +264,7 @@ class TestRenderPlan:
         """完整渲染流程成功。"""
 
         # 素材下载成功
-        def _fake_download(asset_id, local_path):
+        def _fake_download(storage_key, local_path):
             local_path.parent.mkdir(parents=True, exist_ok=True)
             local_path.write_bytes(b"fake video data")
             return True
@@ -269,7 +290,11 @@ class TestRenderPlan:
             _make_clip("c1", order=0, duration=5.0),
             _make_clip("c2", order=1, duration=5.0),
         ]
-        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+        asset_url_map = {
+            "asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4",
+            "asset_c2.mp4": "https://test-bucket.oss.com/assets/asset_c2.mp4",
+        }
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)
 
         result = adapter.render_plan(
             "plan_001",
@@ -295,7 +320,7 @@ class TestRenderPlan:
     def test_progress_callback(self, mock_download, tmp_path):
         """进度回调被正确触发。"""
 
-        def _fake_download(asset_id, local_path):
+        def _fake_download(storage_key, local_path):
             local_path.parent.mkdir(parents=True, exist_ok=True)
             local_path.write_bytes(b"fake data")
             return True
@@ -310,7 +335,8 @@ class TestRenderPlan:
 
             plan = FakePlan(id="plan_001", status="editing")
             clips = [_make_clip("c1", order=0, duration=5.0)]
-            adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+            asset_url_map = {"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4"}
+            adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)
 
             progress_values = []
 
@@ -333,8 +359,8 @@ class TestRenderPlan:
         """部分素材下载失败时，只使用成功的素材。"""
         download_results = [True, False, True]  # 3个素材中2个成功
 
-        def _fake_download(asset_id, local_path):
-            idx = hash(asset_id) % 3
+        def _fake_download(storage_key, local_path):
+            idx = hash(storage_key) % 3
             if download_results[idx]:
                 local_path.parent.mkdir(parents=True, exist_ok=True)
                 local_path.write_bytes(b"fake data")
@@ -357,11 +383,16 @@ class TestRenderPlan:
             with patch("video_processing.render_adapter.upload_to_oss", return_value="https://example.com/out.mp4"):
                 plan = FakePlan(id="plan_001", status="editing")
                 clips = [
-                    _make_clip("c1", order=0, duration=3.0, asset_id="asset_001.mp4"),
-                    _make_clip("c2", order=1, duration=3.0, asset_id="asset_002.mp4"),
-                    _make_clip("c3", order=2, duration=3.0, asset_id="asset_003.mp4"),
+                    _make_clip("c1", order=0, duration=3.0, asset_id="asset_001"),
+                    _make_clip("c2", order=1, duration=3.0, asset_id="asset_002"),
+                    _make_clip("c3", order=2, duration=3.0, asset_id="asset_003"),
                 ]
-                adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+                asset_url_map = {
+                    "asset_001": "https://test-bucket.oss.com/assets/001.mp4",
+                    "asset_002": "https://test-bucket.oss.com/assets/002.mp4",
+                    "asset_003": "https://test-bucket.oss.com/assets/003.mp4",
+                }
+                adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)
 
                 result = adapter.render_plan(
                     "plan_001",
@@ -377,38 +408,75 @@ class TestRenderPlan:
 
 
 class TestDownloadAssets:
+    @staticmethod
+    def _make_mock_db(asset_url_map: dict[str, str]):
+        """构造 mock db，根据 asset_id 返回对应的 AssetModel.file_url。"""
+        mock_db = MagicMock()
+        mock_query = MagicMock()
+
+        def _fake_filter(query):
+            # 模拟 .filter(AssetModel.id.in_([...])).all()
+            mock_assets = []
+            for asset_id, file_url in asset_url_map.items():
+                mock_asset = MagicMock()
+                mock_asset.id = asset_id
+                mock_asset.file_url = file_url
+                mock_assets.append(mock_asset)
+            mock_query.all.return_value = mock_assets
+            return mock_query
+
+        mock_query.filter.side_effect = _fake_filter
+        mock_db.query.return_value = mock_query
+        return mock_db
+
     @patch("video_processing.render_adapter.download_asset")
     def test_all_download_success(self, mock_download, tmp_path):
         """全部素材下载成功。"""
         mock_download.return_value = True
 
+        asset_url_map = {
+            "asset_001": "https://bucket.oss.com/videos/key1.mp4",
+            "asset_002": "https://bucket.oss.com/videos/key2.mp4",
+        }
+        mock_db = self._make_mock_db(asset_url_map)
+        adapter = RenderAdapter(mock_db)
+
         clips = [
-            _make_clip("c1", order=0, asset_id="key1.mp4"),
-            _make_clip("c2", order=1, asset_id="key2.mp4"),
+            _make_clip("c1", order=0, asset_id="asset_001"),
+            _make_clip("c2", order=1, asset_id="asset_002"),
         ]
 
-        asset_path_map, rendered_ids, failed_ids = RenderAdapter._download_assets(clips, tmp_path)
+        asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)
 
         assert len(asset_path_map) == 2
-        assert "key1.mp4" in asset_path_map
-        assert "key2.mp4" in asset_path_map
+        assert "asset_001" in asset_path_map
+        assert "asset_002" in asset_path_map
         assert len(rendered_ids) == 2
         assert len(failed_ids) == 0
         assert mock_download.call_count == 2
+        # 验证传给 download_asset 的是 file_url 而非 asset_id
+        call_keys = [call[0][0] for call in mock_download.call_args_list]
+        assert "https://bucket.oss.com/videos/key1.mp4" in call_keys
+        assert "https://bucket.oss.com/videos/key2.mp4" in call_keys
 
     @patch("video_processing.render_adapter.download_asset")
     def test_empty_asset_id_skipped(self, mock_download, tmp_path):
         """空 asset_id 的片段被跳过。"""
-        clips = [
-            _make_clip("c1", order=0, asset_id=""),
-            _make_clip("c2", order=1, asset_id="key2.mp4"),
-        ]
         mock_download.return_value = True
 
-        asset_path_map, rendered_ids, failed_ids = RenderAdapter._download_assets(clips, tmp_path)
+        asset_url_map = {"asset_002": "https://bucket.oss.com/videos/key2.mp4"}
+        mock_db = self._make_mock_db(asset_url_map)
+        adapter = RenderAdapter(mock_db)
+
+        clips = [
+            _make_clip("c1", order=0, asset_id=""),
+            _make_clip("c2", order=1, asset_id="asset_002"),
+        ]
+
+        asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)
 
         assert len(asset_path_map) == 1
-        assert "key2.mp4" in asset_path_map
+        assert "asset_002" in asset_path_map
         assert "c1" in failed_ids
         assert "c2" in rendered_ids
         assert mock_download.call_count == 1  # 只调用了一次下载
@@ -418,11 +486,15 @@ class TestDownloadAssets:
         """全部下载失败返回空字典。"""
         mock_download.return_value = False
 
+        asset_url_map = {"asset_001": "https://bucket.oss.com/videos/key1.mp4"}
+        mock_db = self._make_mock_db(asset_url_map)
+        adapter = RenderAdapter(mock_db)
+
         clips = [
-            _make_clip("c1", order=0, asset_id="key1.mp4"),
+            _make_clip("c1", order=0, asset_id="asset_001"),
         ]
 
-        asset_path_map, rendered_ids, failed_ids = RenderAdapter._download_assets(clips, tmp_path)
+        asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)
 
         assert len(asset_path_map) == 0
         assert len(rendered_ids) == 0
@@ -431,22 +503,34 @@ class TestDownloadAssets:
     @patch("video_processing.render_adapter.download_asset")
     def test_partial_download_failure(self, mock_download, tmp_path):
         """部分下载失败时正确区分成功/失败。"""
-        results = {"key1.mp4": True, "key2.mp4": False, "key3.mp4": True}
+        results = {
+            "https://bucket.oss.com/videos/key1.mp4": True,
+            "https://bucket.oss.com/videos/key2.mp4": False,
+            "https://bucket.oss.com/videos/key3.mp4": True,
+        }
 
-        def _fake_download(asset_id, local_path):
+        def _fake_download(storage_key, local_path):
             local_path.parent.mkdir(parents=True, exist_ok=True)
             local_path.write_bytes(b"fake")
-            return results.get(asset_id, False)
+            return results.get(storage_key, False)
 
         mock_download.side_effect = _fake_download
 
+        asset_url_map = {
+            "asset_001": "https://bucket.oss.com/videos/key1.mp4",
+            "asset_002": "https://bucket.oss.com/videos/key2.mp4",
+            "asset_003": "https://bucket.oss.com/videos/key3.mp4",
+        }
+        mock_db = self._make_mock_db(asset_url_map)
+        adapter = RenderAdapter(mock_db)
+
         clips = [
-            _make_clip("c1", order=0, asset_id="key1.mp4"),
-            _make_clip("c2", order=1, asset_id="key2.mp4"),
-            _make_clip("c3", order=2, asset_id="key3.mp4"),
+            _make_clip("c1", order=0, asset_id="asset_001"),
+            _make_clip("c2", order=1, asset_id="asset_002"),
+            _make_clip("c3", order=2, asset_id="asset_003"),
         ]
 
-        asset_path_map, rendered_ids, failed_ids = RenderAdapter._download_assets(clips, tmp_path)
+        asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)
 
         assert len(asset_path_map) == 2
         assert "c1" in rendered_ids
@@ -458,18 +542,44 @@ class TestDownloadAssets:
         """同一素材被多个 clip 引用时只下载一次。"""
         mock_download.return_value = True
 
+        asset_url_map = {"asset_shared": "https://bucket.oss.com/videos/shared.mp4"}
+        mock_db = self._make_mock_db(asset_url_map)
+        adapter = RenderAdapter(mock_db)
+
         clips = [
-            _make_clip("c1", order=0, asset_id="shared.mp4"),
-            _make_clip("c2", order=1, asset_id="shared.mp4"),
+            _make_clip("c1", order=0, asset_id="asset_shared"),
+            _make_clip("c2", order=1, asset_id="asset_shared"),
         ]
 
-        asset_path_map, rendered_ids, failed_ids = RenderAdapter._download_assets(clips, tmp_path)
+        asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)
 
         assert len(asset_path_map) == 1
         assert mock_download.call_count == 1
         assert "c1" in rendered_ids
         assert "c2" in rendered_ids
 
+    @patch("video_processing.render_adapter.download_asset")
+    def test_asset_without_file_url_skipped(self, mock_download, tmp_path):
+        """素材在assets表中无file_url时跳过下载，标记为失败。"""
+        # 构造返回 asset 但 file_url 为空
+        mock_db = MagicMock()
+        mock_query = MagicMock()
+        mock_asset = MagicMock()
+        mock_asset.id = "asset_no_url"
+        mock_asset.file_url = ""
+        mock_query.all.return_value = [mock_asset]
+        mock_query.filter.return_value = mock_query
+        mock_db.query.return_value = mock_query
+
+        adapter = RenderAdapter(mock_db)
+        clips = [_make_clip("c1", order=0, asset_id="asset_no_url")]
+
+        asset_path_map, rendered_ids, failed_ids = adapter._download_assets(clips, tmp_path)
+
+        assert len(asset_path_map) == 0
+        assert "c1" in failed_ids
+        mock_download.assert_not_called()
+
 
 # ── _prepare_bgm 测试 ────────────────────────────────────────────────────────
 
@@ -674,7 +784,7 @@ class TestRenderPlanWithBgmAsr:
     def test_bgm_passed_to_render_service(self, mock_download, mock_render_cls, mock_upload, tmp_path):
         """BGM 路径被正确传递给 UnifiedRenderService。"""
 
-        def _fake_download(asset_id, local_path):
+        def _fake_download(storage_key, local_path):
             local_path.parent.mkdir(parents=True, exist_ok=True)
             local_path.write_bytes(b"fake video")
             return True
@@ -702,8 +812,9 @@ class TestRenderPlanWithBgmAsr:
             },
         )
         clips = [_make_clip("c1", order=0, duration=5.0)]
+        asset_url_map = {"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4"}
 
-        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)
 
         # Mock BGM 下载
         with patch.object(adapter, "_prepare_bgm", return_value=str(tmp_path / "bgm.mp3")):
@@ -724,7 +835,7 @@ class TestRenderPlanWithBgmAsr:
     def test_asr_service_passed_to_render_service(self, mock_download, mock_render_cls, mock_upload, tmp_path):
         """ASR 服务被正确传递给 UnifiedRenderService。"""
 
-        def _fake_download(asset_id, local_path):
+        def _fake_download(storage_key, local_path):
             local_path.parent.mkdir(parents=True, exist_ok=True)
             local_path.write_bytes(b"fake video")
             return True
@@ -745,8 +856,9 @@ class TestRenderPlanWithBgmAsr:
         mock_asr = MagicMock()
         plan = FakePlan(id="plan_asr")
         clips = [_make_clip("c1", order=0, duration=5.0)]
+        asset_url_map = {"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4"}
 
-        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)
 
         with patch.object(adapter, "_prepare_bgm", return_value=None):
             with patch.object(adapter, "_get_asr_service", return_value=mock_asr):
@@ -764,16 +876,16 @@ class TestRenderPlanWithBgmAsr:
     @patch("video_processing.render_adapter.download_asset")
     def test_rendered_and_failed_clip_ids_in_result(self, mock_download, mock_render_cls, mock_upload, tmp_path):
         """渲染结果中包含成功和失败的 clip id 列表。"""
-        # 成功/失败映射，通过 asset_id 区分
+        # 成功/失败映射，通过 storage_key（file_url）区分
         download_map = {
-            "good_001.mp4": True,
-            "bad_002.mp4": False,
-            "good_003.mp4": True,
+            "https://test-bucket.oss.com/assets/good_001.mp4": True,
+            "https://test-bucket.oss.com/assets/bad_002.mp4": False,
+            "https://test-bucket.oss.com/assets/good_003.mp4": True,
         }
 
-        def _fake_download(asset_id, local_path):
+        def _fake_download(storage_key, local_path):
             local_path.parent.mkdir(parents=True, exist_ok=True)
-            ok = download_map.get(asset_id, False)
+            ok = download_map.get(storage_key, False)
             if ok:
                 local_path.write_bytes(b"fake video")
             return ok
@@ -793,12 +905,17 @@ class TestRenderPlanWithBgmAsr:
 
         plan = FakePlan(id="plan_mixed")
         clips = [
-            _make_clip("c_good1", order=0, asset_id="good_001.mp4"),
-            _make_clip("c_bad", order=1, asset_id="bad_002.mp4"),
-            _make_clip("c_good2", order=2, asset_id="good_003.mp4"),
+            _make_clip("c_good1", order=0, asset_id="good_001"),
+            _make_clip("c_bad", order=1, asset_id="bad_002"),
+            _make_clip("c_good2", order=2, asset_id="good_003"),
         ]
+        asset_url_map = {
+            "good_001": "https://test-bucket.oss.com/assets/good_001.mp4",
+            "bad_002": "https://test-bucket.oss.com/assets/bad_002.mp4",
+            "good_003": "https://test-bucket.oss.com/assets/good_003.mp4",
+        }
 
-        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map)
 
         result = adapter.render_plan(
             "plan_mixed",