apps/worker/video_processing/render_adapter.py

@@ -1,14 +1,16 @@
 """统一渲染引擎适配层 — Phase 2.
 
 将 EditPlan + EditPlanClips（来自 DB）适配为 UnifiedRenderService 的输入格式，
-封装素材下载、渲染执行、结果上传的完整流程。
+封装素材下载、BGM 准备、ASR 字幕、渲染执行、结果上传的完整流程。
 
 职责：
   1. 从 DB 读取 EditPlan + EditPlanClips
   2. 下载素材到本地，构建 asset_path_map
-  3. 调用 UnifiedRenderService 执行渲染
-  4. 上传渲染结果到 OSS
-  5. 支持进度回调（对接 JobService）
+  3. 准备 BGM 音频（URL / 素材库 / 预设库）
+  4. 初始化 ASR 服务（自动字幕）
+  5. 调用 UnifiedRenderService 执行渲染
+  6. 上传渲染结果到 OSS
+  7. 支持进度回调（对接 JobService）
 """
 
 from __future__ import annotations
@@ -17,7 +19,7 @@ import logging
 import tempfile
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Callable
+from typing import Any, Callable
 
 from sqlalchemy.orm import Session
 from video_processing.oss_helpers import download_asset, upload_to_oss
@@ -46,8 +48,16 @@ class RenderAdapterResult:
     width: int = 0
     height: int = 0
     clip_count: int = 0
+    rendered_clip_ids: list[str] = None  # 成功渲染的 clip id 列表
+    failed_clip_ids: list[str] = None  # 失败的 clip id 列表
     error_message: str = ""
 
+    def __post_init__(self):
+        if self.rendered_clip_ids is None:
+            self.rendered_clip_ids = []
+        if self.failed_clip_ids is None:
+            self.failed_clip_ids = []
+
 
 ProgressCallback = Callable[[float, str], None]
 """进度回调：(progress_0_100, stage_description) → None"""
@@ -142,22 +152,34 @@ class RenderAdapter:
             self._report_progress(progress_cb, 15.0, f"下载素材（{len(ready_clips)} 个）")
 
             # 2. 下载素材
-            asset_path_map = self._download_assets(ready_clips, work_dir)
+            asset_path_map, rendered_clip_ids, failed_clip_ids = self._download_assets(ready_clips, work_dir)
             if not asset_path_map:
                 return RenderAdapterResult(
                     success=False,
                     error_message="所有素材下载失败",
                     clip_count=len(ready_clips),
+                    rendered_clip_ids=[],
+                    failed_clip_ids=failed_clip_ids,
                 )
 
+            self._report_progress(progress_cb, 35.0, "准备 BGM 音频")
+
+            # 3. 准备 BGM（从 plan.config.bgm 读取配置）
+            bgm_path = self._prepare_bgm(plan, work_dir, plan_id)
+
             self._report_progress(progress_cb, 40.0, "执行视频渲染")
 
-            # 3. 执行统一渲染
+            # 4. 初始化 ASR 服务（用于自动字幕）
+            asr_service = self._get_asr_service()
+
+            # 5. 执行统一渲染
             render_svc = UnifiedRenderService(
                 plan=plan,
                 clips=ready_clips,
                 asset_path_map=asset_path_map,
                 work_dir=work_dir,
+                bgm_path=bgm_path,
+                asr_service=asr_service,
             )
             result = render_svc.render()
 
@@ -190,6 +212,8 @@ class RenderAdapter:
                 width=result.width,
                 height=result.height,
                 clip_count=len(ready_clips),
+                rendered_clip_ids=rendered_clip_ids,
+                failed_clip_ids=failed_clip_ids,
             )
 
         except Exception as exc:
@@ -271,29 +295,153 @@ class RenderAdapter:
                 logger.exception("进度回调失败")
 
     @staticmethod
-    def _download_assets(clips: list[EditPlanClip], work_dir: Path) -> dict[str, Path]:
-        """下载片段素材到本地，返回 asset_id → local_path 映射。
+    def _download_assets(clips: list[EditPlanClip], work_dir: Path) -> tuple[dict[str, Path], list[str], list[str]]:
+        """下载片段素材到本地。
 
-        只保留下载成功的素材。
+        Returns:
+            (asset_path_map, rendered_clip_ids, failed_clip_ids)
+            - asset_path_map: asset_id → local_path 映射（下载成功的）
+            - rendered_clip_ids: 下载成功的 clip id 列表
+            - failed_clip_ids: 下载失败的 clip id 列表
         """
         asset_dir = work_dir / "assets"
         asset_dir.mkdir(exist_ok=True)
 
         asset_path_map: dict[str, Path] = {}
+        rendered_clip_ids: list[str] = []
+        failed_clip_ids: list[str] = []
+        seen_asset_ids: set[str] = set()
 
         for clip in clips:
             asset_id = clip.asset_id
             if not asset_id:
+                failed_clip_ids.append(clip.id)
                 continue
 
+            # 同一素材已下载过（多个 clip 共享同一素材）
+            if asset_id in seen_asset_ids:
+                if asset_id in asset_path_map:
+                    rendered_clip_ids.append(clip.id)
+                else:
+                    failed_clip_ids.append(clip.id)
+                continue
+
+            seen_asset_ids.add(asset_id)
+
             # 生成安全的本地文件名
             safe_name = f"clip_{clip.order:04d}_{abs(hash(asset_id)) % 100000:05d}.mp4"
             local_path = asset_dir / safe_name
 
             if download_asset(asset_id, 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])
             else:
+                failed_clip_ids.append(clip.id)
                 logger.warning("素材下载失败: clip_id=%s asset_id=%s", clip.id, asset_id[:60])
 
-        return asset_path_map
+        return asset_path_map, rendered_clip_ids, failed_clip_ids
+
+    def _prepare_bgm(self, plan, work_dir: Path, plan_id: str) -> str | None:
+        """准备 BGM 音频文件（从 plan.config.bgm 读取配置）。
+
+        支持 3 种来源（按优先级）：
+        1. audio_url — 外部直链 URL
+        2. asset_id — 素材库中的音频素材
+        3. preset_id — 预设 BGM 库
+
+        失败不阻断主流程，返回 None。
+        """
+        from urllib.parse import urlparse
+
+        plan_config = plan.config or {}
+        bgm_config = plan_config.get("bgm", {}) or {}
+
+        if not bgm_config.get("enabled", False):
+            return None
+
+        audio_url = bgm_config.get("audio_url", "") or ""
+        asset_id = bgm_config.get("asset_id", "") or ""
+        preset_id = bgm_config.get("preset_id", "") or ""
+
+        bgm_file = work_dir / "bgm.mp3"
+
+        # 优先级1：外部直链 URL
+        if audio_url:
+            try:
+                parsed = urlparse(audio_url)
+                if parsed.scheme in ("http", "https"):
+                    from video_processing.url_security import (
+                        ALLOWED_AUDIO_MIME_TYPES,
+                        safe_download_file,
+                    )
+
+                    logger.info("[plan_id=%s] [BGM] 从URL下载: %s", plan_id, audio_url[:80])
+                    safe_download_file(
+                        audio_url,
+                        str(bgm_file),
+                        purpose="bgm_download",
+                        allowed_mime_types=ALLOWED_AUDIO_MIME_TYPES,
+                        timeout=60.0,
+                    )
+                    if bgm_file.exists() and bgm_file.stat().st_size > 0:
+                        return str(bgm_file)
+            except Exception as e:
+                logger.warning("[plan_id=%s] [BGM] URL下载失败: %s", plan_id, e)
+
+        # 优先级2：素材库素材
+        if asset_id:
+            try:
+                from packages.adapters.sqlalchemy_impl.models import AssetModel
+
+                model = self._db.query(AssetModel).filter(AssetModel.id == asset_id).first()
+                if model and model.file_url:
+                    storage_key = model.file_url
+                    logger.info("[plan_id=%s] [BGM] 从素材库下载: asset_id=%s", plan_id, asset_id)
+                    ok = download_asset(storage_key, bgm_file)
+                    if ok and bgm_file.exists() and bgm_file.stat().st_size > 0:
+                        return str(bgm_file)
+            except Exception as e:
+                logger.warning("[plan_id=%s] [BGM] 素材库下载失败: %s", plan_id, e)
+
+        # 优先级3：预设 BGM 库
+        if preset_id:
+            try:
+                from packages.domain.preset_bgm import get_preset_bgm
+
+                preset = get_preset_bgm(preset_id)
+                if preset and preset.audio_url:
+                    from video_processing.url_security import (
+                        ALLOWED_AUDIO_MIME_TYPES,
+                        safe_download_file,
+                    )
+
+                    logger.info("[plan_id=%s] [BGM] 从预设库下载: preset_id=%s", plan_id, preset_id)
+                    safe_download_file(
+                        preset.audio_url,
+                        str(bgm_file),
+                        purpose="bgm_preset_download",
+                        allowed_mime_types=ALLOWED_AUDIO_MIME_TYPES,
+                        timeout=60.0,
+                    )
+                    if bgm_file.exists() and bgm_file.stat().st_size > 0:
+                        return str(bgm_file)
+            except Exception as e:
+                logger.warning("[plan_id=%s] [BGM] 预设库下载失败: %s", plan_id, e)
+
+        logger.warning("[plan_id=%s] [BGM] 所有来源都无法获取BGM，跳过", plan_id)
+        return None
+
+    @staticmethod
+    def _get_asr_service() -> Any | None:
+        """获取 ASR 服务实例（用于自动生成字幕）。
+
+        失败不阻断主流程，返回 None。
+        """
+        try:
+            from services.asr_service_factory import get_asr_service
+
+            return get_asr_service()
+        except Exception as e:
+            logger.warning("ASR 服务初始化失败，自动字幕将不可用: %s", e)
+            return None

apps/worker/worker_app/tasks/edit_plan_generation.py

@@ -39,7 +39,6 @@ from video_processing.oss_helpers import (
     download_asset,
     upload_to_oss,
 )
-from video_processing.unified_render_service import UnifiedRenderService
 
 # ── Repository imports (延迟导入避免循环依赖) ─────────────────────────────────
 
@@ -218,9 +217,6 @@ def _finalize_render_success(
 def _render_with_unified(
     plan,
     clips,
-    asset_path_map: dict[str, Path],
-    tmpdir_path: Path,
-    rendered_clip_ids: list[str],
     plan_id: str,
     generation_task_id: str,
     plan_repo,
@@ -228,31 +224,59 @@ def _render_with_unified(
     gen_task_repo,
     db,
 ) -> dict:
-    """统一渲染引擎路径（UnifiedRenderService 图层架构）。"""
-    render_service = UnifiedRenderService(
-        plan=plan,
-        clips=clips,
-        asset_path_map=asset_path_map,
-        work_dir=tmpdir_path,
-        output_width=OUTPUT_WIDTH,
-        output_height=OUTPUT_HEIGHT,
-        output_fps=int(OUTPUT_FPS),
-    )
+    """统一渲染引擎路径（通过 RenderAdapter 调用 UnifiedRenderService）。
+
+    RenderAdapter 内部处理：素材下载、BGM 准备、ASR 自动字幕、渲染执行、OSS 上传。
+    本函数只负责：业务状态更新、查重、收尾。
+    """
+    from video_processing.render_adapter import RenderAdapter
+
+    adapter = RenderAdapter(db)
+
+    # 进度回调：更新 GenerationTask 进度
+    def _progress_cb(progress: float, stage: str):
+        if not generation_task_id:
+            return
+        try:
+            gen_task = gen_task_repo.get(generation_task_id)
+            if gen_task:
+                # 映射到 30%~90% 区间（素材下载前已到 30%）
+                mapped_progress = 30.0 + progress * 0.6
+                gen_task.progress = min(mapped_progress, 95.0)
+                gen_task.append_log(
+                    stage="render_progress",
+                    message=stage,
+                    level="INFO",
+                    progress=mapped_progress,
+                )
+                gen_task_repo.update(gen_task)
+        except Exception:
+            pass
 
     try:
-        render_result = render_service.render()
+        result = adapter.render_plan(
+            plan_id=plan_id,
+            job_id=generation_task_id or plan_id,
+            progress_cb=_progress_cb,
+        )
     except Exception as render_err:
         logger.error("渲染失败(unified): %s — %s", plan_id, render_err)
         _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, f"渲染失败: {render_err}")
         return {"status": "error", "message": f"渲染失败: {render_err}"}
 
-    output_path = render_result.output_path
+    if not result.success:
+        logger.error("渲染失败(unified): %s — %s", plan_id, result.error_message)
+        _mark_plan_failed(plan_repo, plan_id, gen_task_repo, generation_task_id, result.error_message or "渲染失败")
+        return {"status": "error", "message": result.error_message or "渲染失败"}
 
-    # 上传到 OSS
+    output_path = result.output_path or Path("")
+    output_url = result.output_url
     storage_key = f"rendered/{plan_id}/output.mp4"
-    output_url = upload_to_oss(output_path, storage_key)
 
-    failed_clip_ids: list[str] = []
+    # 用 adapter 返回的 clip 明细（以 adapter 的结果为准）
+    rendered_clip_ids = result.rendered_clip_ids or []
+    failed_clip_ids = result.failed_clip_ids or []
+
     return _finalize_render_success(
         plan=plan,
         plan_repo=plan_repo,
@@ -262,10 +286,10 @@ def _render_with_unified(
         plan_id=plan_id,
         output_url=output_url or "",
         storage_key=storage_key,
-        duration=render_result.duration,
-        file_size=render_result.file_size,
-        width=render_result.width,
-        height=render_result.height,
+        duration=result.duration,
+        file_size=result.file_size,
+        width=result.width,
+        height=result.height,
         rendered_clip_ids=rendered_clip_ids,
         failed_clip_ids=failed_clip_ids,
         generation_task_id=generation_task_id,
@@ -406,128 +430,123 @@ def render_edit_plan(self, plan_id: str) -> dict:
                     )
                     gen_task_repo.update(gen_task)
 
-            # 3. 下载素材并构建 asset_path_map
-            with tempfile.TemporaryDirectory(prefix="edit_plan_") as tmpdir:
-                tmpdir_path = Path(tmpdir)
-                asset_path_map: dict[str, Path] = {}
-                rendered_clip_ids: list[str] = []
-                failed_clip_ids: list[str] = []
+            # 3. 渲染前取消检查
+            if generation_task_id:
+                current_task = gen_task_repo.get(generation_task_id)
+                if current_task:
+                    task_status = (
+                        current_task.status.value if hasattr(current_task.status, "value") else str(current_task.status)
+                    )
+                    if task_status == "cancelled":
+                        logger.info("任务已被取消，中止渲染: plan_id=%s task_id=%s", plan_id, generation_task_id)
+                        from packages.domain.edit_plan import EditPlanStatus
 
-                # 预先批量查询所有素材的 storage_key（file_url）
-                from packages.adapters.sqlalchemy_impl.models import AssetModel
+                        if plan.status.value == "rendering":
+                            try:
+                                plan.resume_editing()
+                                plan_repo.update(plan)
+                            except ValueError:
+                                pass
+                        return {"status": "cancelled", "plan_id": plan_id, "message": "任务已取消"}
 
-                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 = 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}
+            # 4. 根据引擎选择渲染方式
+            if engine == "unified":
+                # ── unified 路径：RenderAdapter 统一处理（下载 + BGM + ASR + 渲染 + 上传）
+                result = _render_with_unified(
+                    plan=plan,
+                    clips=clips,
+                    plan_id=plan_id,
+                    generation_task_id=generation_task_id,
+                    plan_repo=plan_repo,
+                    clip_repo=clip_repo,
+                    gen_task_repo=gen_task_repo,
+                    db=db,
+                )
+            else:
+                # ── legacy 路径：原有的素材下载 + VideoComposeService
+                with tempfile.TemporaryDirectory(prefix="edit_plan_") as tmpdir:
+                    tmpdir_path = Path(tmpdir)
+                    asset_path_map: dict[str, Path] = {}
+                    rendered_clip_ids: list[str] = []
+                    failed_clip_ids: list[str] = []
 
-                for clip in clips:
-                    if not clip.asset_id:
-                        # 没有素材的片段跳过，标记为失败
-                        clip.mark_failed()
-                        clip_repo.update(clip)
-                        failed_clip_ids.append(clip.id)
-                        continue
+                    # 预先批量查询所有素材的 storage_key（file_url）
+                    from packages.adapters.sqlalchemy_impl.models import AssetModel
 
-                    if clip.asset_id in asset_path_map:
-                        # 同一素材已下载（多个 clip 共享同一素材）
-                        rendered_clip_ids.append(clip.id)
-                        continue
+                    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 = 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}
 
-                    storage_key = asset_storage_map.get(clip.asset_id)
-                    if not storage_key:
-                        logger.warning(
-                            "片段素材无 storage_key，跳过: clip_id=%s asset_id=%s",
-                            clip.id,
-                            clip.asset_id,
-                        )
-                        clip.mark_failed()
-                        clip_repo.update(clip)
-                        failed_clip_ids.append(clip.id)
-                        continue
+                    for clip in clips:
+                        if not clip.asset_id:
+                            # 没有素材的片段跳过，标记为失败
+                            clip.mark_failed()
+                            clip_repo.update(clip)
+                            failed_clip_ids.append(clip.id)
+                            continue
 
-                    # 下载素材
-                    ext = Path(storage_key).suffix or ".mp4"
-                    local_path = tmpdir_path / f"clip_{clip.order:04d}{ext}"
-                    if download_asset(storage_key, local_path):
-                        asset_path_map[clip.asset_id] = local_path
-                        rendered_clip_ids.append(clip.id)
-                    else:
-                        clip.mark_failed()
-                        clip_repo.update(clip)
-                        failed_clip_ids.append(clip.id)
+                        if clip.asset_id in asset_path_map:
+                            # 同一素材已下载（多个 clip 共享同一素材）
+                            rendered_clip_ids.append(clip.id)
+                            continue
 
-                if not asset_path_map:
-                    logger.error("所有片段素材下载失败: %s", plan_id)
-                    plan.mark_failed()
-                    plan_repo.update(plan)
+                        storage_key = asset_storage_map.get(clip.asset_id)
+                        if not storage_key:
+                            logger.warning(
+                                "片段素材无 storage_key，跳过: clip_id=%s asset_id=%s",
+                                clip.id,
+                                clip.asset_id,
+                            )
+                            clip.mark_failed()
+                            clip_repo.update(clip)
+                            failed_clip_ids.append(clip.id)
+                            continue
+
+                        # 下载素材
+                        ext = Path(storage_key).suffix or ".mp4"
+                        local_path = tmpdir_path / f"clip_{clip.order:04d}{ext}"
+                        if download_asset(storage_key, local_path):
+                            asset_path_map[clip.asset_id] = local_path
+                            rendered_clip_ids.append(clip.id)
+                        else:
+                            clip.mark_failed()
+                            clip_repo.update(clip)
+                            failed_clip_ids.append(clip.id)
+
+                    if not asset_path_map:
+                        logger.error("所有片段素材下载失败: %s", plan_id)
+                        plan.mark_failed()
+                        plan_repo.update(plan)
+                        if generation_task_id:
+                            gen_task = gen_task_repo.get(generation_task_id)
+                            if gen_task:
+                                gen_task.status = "failed"
+                                gen_task.error_message = "所有片段素材下载失败"
+                                gen_task.completed_at = datetime.now(timezone.utc)
+                                gen_task.append_log(
+                                    stage="download_failed",
+                                    message="所有片段素材下载失败",
+                                    level="ERROR",
+                                )
+                                gen_task_repo.update(gen_task)
+                        return {"status": "error", "message": "所有片段素材下载失败"}
+
+                    # 素材下载完成，记录日志
                     if generation_task_id:
                         gen_task = gen_task_repo.get(generation_task_id)
                         if gen_task:
-                            gen_task.status = "failed"
-                            gen_task.error_message = "所有片段素材下载失败"
-                            gen_task.completed_at = datetime.now(timezone.utc)
                             gen_task.append_log(
-                                stage="download_failed",
-                                message="所有片段素材下载失败",
-                                level="ERROR",
+                                stage="download_done",
+                                message=f"素材下载完成，成功 {len(asset_path_map)} 个，失败 {len(failed_clip_ids)} 个",
+                                level="INFO",
+                                success_count=len(asset_path_map),
+                                failed_count=len(failed_clip_ids),
                             )
+                            gen_task.progress = 30.0
                             gen_task_repo.update(gen_task)
-                    return {"status": "error", "message": "所有片段素材下载失败"}
 
-                # 素材下载完成，记录日志
-                if generation_task_id:
-                    gen_task = gen_task_repo.get(generation_task_id)
-                    if gen_task:
-                        gen_task.append_log(
-                            stage="download_done",
-                            message=f"素材下载完成，成功 {len(asset_path_map)} 个，失败 {len(failed_clip_ids)} 个",
-                            level="INFO",
-                            success_count=len(asset_path_map),
-                            failed_count=len(failed_clip_ids),
-                        )
-                        gen_task.progress = 30.0
-                        gen_task_repo.update(gen_task)
-
-                # 4. 根据引擎选择渲染方式
-                # 取消检查：素材下载完后，确认任务没有被用户取消
-                if generation_task_id:
-                    current_task = gen_task_repo.get(generation_task_id)
-                    if current_task:
-                        task_status = (
-                            current_task.status.value
-                            if hasattr(current_task.status, "value")
-                            else str(current_task.status)
-                        )
-                        if task_status == "cancelled":
-                            logger.info("任务已被取消，中止渲染: plan_id=%s task_id=%s", plan_id, generation_task_id)
-                            # 计划回到 editing 状态，用户可以继续编辑
-                            from packages.domain.edit_plan import EditPlanStatus
-
-                            if plan.status.value == "rendering":
-                                try:
-                                    plan.resume_editing()
-                                    plan_repo.update(plan)
-                                except ValueError:
-                                    pass
-                            return {"status": "cancelled", "plan_id": plan_id, "message": "任务已取消"}
-
-                if engine == "unified":
-                    result = _render_with_unified(
-                        plan=plan,
-                        clips=clips,
-                        asset_path_map=asset_path_map,
-                        tmpdir_path=tmpdir_path,
-                        rendered_clip_ids=rendered_clip_ids,
-                        plan_id=plan_id,
-                        generation_task_id=generation_task_id,
-                        plan_repo=plan_repo,
-                        clip_repo=clip_repo,
-                        gen_task_repo=gen_task_repo,
-                        db=db,
-                    )
-                else:
                     result = _render_with_legacy(
                         plan=plan,
                         clips=clips,
@@ -542,8 +561,8 @@ def render_edit_plan(self, plan_id: str) -> dict:
                         db=db,
                     )
 
-                result["engine"] = engine
-                return result
+            result["engine"] = engine
+            return result
 
         except Exception as exc:
             logger.exception("渲染剪辑计划异常: %s", plan_id)

tests/unit/test_render_adapter.py

@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
+from pathlib import Path
 from typing import Any
 from unittest.mock import MagicMock, patch
 
@@ -386,11 +387,13 @@ class TestDownloadAssets:
             _make_clip("c2", order=1, asset_id="key2.mp4"),
         ]
 
-        result = RenderAdapter._download_assets(clips, tmp_path)
+        asset_path_map, rendered_ids, failed_ids = RenderAdapter._download_assets(clips, tmp_path)
 
-        assert len(result) == 2
-        assert "key1.mp4" in result
-        assert "key2.mp4" in result
+        assert len(asset_path_map) == 2
+        assert "key1.mp4" in asset_path_map
+        assert "key2.mp4" in asset_path_map
+        assert len(rendered_ids) == 2
+        assert len(failed_ids) == 0
         assert mock_download.call_count == 2
 
     @patch("video_processing.render_adapter.download_asset")
@@ -402,10 +405,12 @@ class TestDownloadAssets:
         ]
         mock_download.return_value = True
 
-        result = RenderAdapter._download_assets(clips, tmp_path)
+        asset_path_map, rendered_ids, failed_ids = RenderAdapter._download_assets(clips, tmp_path)
 
-        assert len(result) == 1
-        assert "key2.mp4" in result
+        assert len(asset_path_map) == 1
+        assert "key2.mp4" in asset_path_map
+        assert "c1" in failed_ids
+        assert "c2" in rendered_ids
         assert mock_download.call_count == 1  # 只调用了一次下载
 
     @patch("video_processing.render_adapter.download_asset")
@@ -417,6 +422,392 @@ class TestDownloadAssets:
             _make_clip("c1", order=0, asset_id="key1.mp4"),
         ]
 
-        result = RenderAdapter._download_assets(clips, tmp_path)
+        asset_path_map, rendered_ids, failed_ids = RenderAdapter._download_assets(clips, tmp_path)
 
-        assert len(result) == 0
+        assert len(asset_path_map) == 0
+        assert len(rendered_ids) == 0
+        assert "c1" in failed_ids
+
+    @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}
+
+        def _fake_download(asset_id, local_path):
+            local_path.parent.mkdir(parents=True, exist_ok=True)
+            local_path.write_bytes(b"fake")
+            return results.get(asset_id, False)
+
+        mock_download.side_effect = _fake_download
+
+        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"),
+        ]
+
+        asset_path_map, rendered_ids, failed_ids = RenderAdapter._download_assets(clips, tmp_path)
+
+        assert len(asset_path_map) == 2
+        assert "c1" in rendered_ids
+        assert "c3" in rendered_ids
+        assert "c2" in failed_ids
+
+    @patch("video_processing.render_adapter.download_asset")
+    def test_duplicate_asset_downloaded_once(self, mock_download, tmp_path):
+        """同一素材被多个 clip 引用时只下载一次。"""
+        mock_download.return_value = True
+
+        clips = [
+            _make_clip("c1", order=0, asset_id="shared.mp4"),
+            _make_clip("c2", order=1, asset_id="shared.mp4"),
+        ]
+
+        asset_path_map, rendered_ids, failed_ids = RenderAdapter._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
+
+
+# ── _prepare_bgm 测试 ────────────────────────────────────────────────────────
+
+
+class TestPrepareBgm:
+    """BGM 音频准备逻辑测试。"""
+
+    def test_bgm_disabled_returns_none(self, tmp_path):
+        """BGM 未启用时返回 None。"""
+        plan = FakePlan(config={"bgm": {"enabled": False}})
+        adapter = RenderAdapter(MagicMock())
+        result = adapter._prepare_bgm(plan, tmp_path, "plan_001")
+        assert result is None
+
+    def test_bgm_no_config_returns_none(self, tmp_path):
+        """没有 BGM 配置时返回 None。"""
+        plan = FakePlan(config={})
+        adapter = RenderAdapter(MagicMock())
+        result = adapter._prepare_bgm(plan, tmp_path, "plan_001")
+        assert result is None
+
+    def test_bgm_from_url(self, tmp_path):
+        """从 URL 下载 BGM。"""
+        plan = FakePlan(
+            config={
+                "bgm": {
+                    "enabled": True,
+                    "audio_url": "https://example.com/bgm.mp3",
+                }
+            }
+        )
+        adapter = RenderAdapter(MagicMock())
+
+        with patch("video_processing.url_security.safe_download_file") as mock_download:
+
+            def _fake_download(url, path, **kwargs):
+                Path(path).parent.mkdir(parents=True, exist_ok=True)
+                Path(path).write_bytes(b"fake mp3 data")
+
+            mock_download.side_effect = _fake_download
+
+            result = adapter._prepare_bgm(plan, tmp_path, "plan_001")
+
+        assert result is not None
+        assert Path(result).exists()
+
+    def test_bgm_from_url_failure_falls_through(self, tmp_path):
+        """URL 下载失败时不抛异常，继续尝试其他来源。"""
+        plan = FakePlan(
+            config={
+                "bgm": {
+                    "enabled": True,
+                    "audio_url": "https://example.com/bgm.mp3",
+                    "preset_id": "preset_001",
+                }
+            }
+        )
+        adapter = RenderAdapter(MagicMock())
+
+        with patch("video_processing.url_security.safe_download_file") as mock_download:
+            mock_download.side_effect = Exception("download failed")
+
+            # 预设库返回 None（没有这个预设）
+            with patch("packages.domain.preset_bgm.get_preset_bgm") as mock_preset:
+                mock_preset.return_value = None
+                result = adapter._prepare_bgm(plan, tmp_path, "plan_001")
+
+        assert result is None  # 所有来源都失败时返回 None
+
+    def test_bgm_from_asset_library(self, tmp_path):
+        """从素材库下载 BGM。"""
+        plan = FakePlan(
+            config={
+                "bgm": {
+                    "enabled": True,
+                    "asset_id": "asset_bgm_001",
+                }
+            }
+        )
+
+        mock_db = MagicMock()
+        mock_asset = MagicMock()
+        mock_asset.file_url = "oss://bgm/sample.mp3"
+        mock_query = MagicMock()
+        mock_query.first.return_value = mock_asset
+        mock_db.query.return_value = mock_query
+
+        adapter = RenderAdapter(mock_db)
+
+        with patch("video_processing.render_adapter.download_asset") as mock_download:
+
+            def _fake_download(storage_key, local_path):
+                Path(local_path).parent.mkdir(parents=True, exist_ok=True)
+                Path(local_path).write_bytes(b"fake bgm")
+                return True
+
+            mock_download.side_effect = _fake_download
+
+            result = adapter._prepare_bgm(plan, tmp_path, "plan_001")
+
+        assert result is not None
+        assert Path(result).exists()
+
+    def test_bgm_from_preset_library(self, tmp_path):
+        """从预设 BGM 库下载。"""
+        plan = FakePlan(
+            config={
+                "bgm": {
+                    "enabled": True,
+                    "preset_id": "preset_calm",
+                }
+            }
+        )
+        adapter = RenderAdapter(MagicMock())
+
+        with patch("packages.domain.preset_bgm.get_preset_bgm") as mock_preset:
+            mock_preset.return_value = MagicMock(audio_url="https://cdn.example.com/preset_calm.mp3")
+
+            with patch("video_processing.url_security.safe_download_file") as mock_download:
+
+                def _fake_download(url, path, **kwargs):
+                    Path(path).parent.mkdir(parents=True, exist_ok=True)
+                    Path(path).write_bytes(b"preset bgm data")
+
+                mock_download.side_effect = _fake_download
+
+                result = adapter._prepare_bgm(plan, tmp_path, "plan_001")
+
+        assert result is not None
+        assert Path(result).exists()
+
+    def test_bgm_url_takes_priority_over_asset(self, tmp_path):
+        """URL 优先级高于素材库。"""
+        plan = FakePlan(
+            config={
+                "bgm": {
+                    "enabled": True,
+                    "audio_url": "https://example.com/bgm.mp3",
+                    "asset_id": "asset_bgm_001",
+                }
+            }
+        )
+        adapter = RenderAdapter(MagicMock())
+
+        with patch("video_processing.url_security.safe_download_file") as mock_url_download:
+
+            def _fake_url_download(url, path, **kwargs):
+                Path(path).parent.mkdir(parents=True, exist_ok=True)
+                Path(path).write_bytes(b"url bgm")
+
+            mock_url_download.side_effect = _fake_url_download
+
+            with patch("video_processing.render_adapter.download_asset") as mock_asset_download:
+                result = adapter._prepare_bgm(plan, tmp_path, "plan_001")
+
+        assert result is not None
+        mock_url_download.assert_called_once()
+        mock_asset_download.assert_not_called()  # URL 成功后不会再走素材库
+
+
+# ── _get_asr_service 测试 ────────────────────────────────────────────────────
+
+
+class TestGetAsrService:
+    """ASR 服务初始化测试。"""
+
+    def test_returns_service_when_available(self):
+        """ASR 服务可用时返回实例。"""
+        mock_service = MagicMock()
+        with patch("services.asr_service_factory.get_asr_service") as mock_get:
+            mock_get.return_value = mock_service
+            result = RenderAdapter._get_asr_service()
+
+        assert result is mock_service
+
+    def test_returns_none_when_import_fails(self):
+        """ASR 服务导入失败时返回 None（不阻断主流程）。"""
+        with patch("services.asr_service_factory.get_asr_service") as mock_get:
+            mock_get.side_effect = ImportError("asr module not found")
+            result = RenderAdapter._get_asr_service()
+
+        assert result is None
+
+    def test_returns_none_when_init_fails(self):
+        """ASR 服务初始化失败时返回 None（不阻断主流程）。"""
+        with patch("services.asr_service_factory.get_asr_service") as mock_get:
+            mock_get.side_effect = RuntimeError("ASR init failed")
+            result = RenderAdapter._get_asr_service()
+
+        assert result is None
+
+
+# ── render_plan BGM/ASR 集成测试 ─────────────────────────────────────────────
+
+
+class TestRenderPlanWithBgmAsr:
+    """渲染流程中 BGM 和 ASR 的集成测试。"""
+
+    @patch("video_processing.render_adapter.upload_to_oss")
+    @patch("video_processing.render_adapter.UnifiedRenderService")
+    @patch("video_processing.render_adapter.download_asset")
+    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):
+            local_path.parent.mkdir(parents=True, exist_ok=True)
+            local_path.write_bytes(b"fake video")
+            return True
+
+        mock_download.side_effect = _fake_download
+
+        mock_render = MagicMock()
+        mock_render.render.return_value = MagicMock(
+            output_path=tmp_path / "out.mp4",
+            duration=5.0,
+            file_size=1024,
+            width=1280,
+            height=720,
+        )
+        mock_render_cls.return_value = mock_render
+        mock_upload.return_value = "https://oss.example.com/out.mp4"
+
+        plan = FakePlan(
+            id="plan_bgm",
+            config={
+                "bgm": {
+                    "enabled": True,
+                    "preset_id": "preset_001",
+                }
+            },
+        )
+        clips = [_make_clip("c1", order=0, duration=5.0)]
+
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+        # Mock BGM 下载
+        with patch.object(adapter, "_prepare_bgm", return_value=str(tmp_path / "bgm.mp3")):
+            with patch.object(adapter, "_get_asr_service", return_value=None):
+                result = adapter.render_plan(
+                    "plan_bgm",
+                    work_dir=tmp_path / "work",
+                )
+
+        assert result.success
+        # 验证 UnifiedRenderService 收到了 bgm_path
+        call_kwargs = mock_render_cls.call_args
+        assert call_kwargs.kwargs["bgm_path"] == str(tmp_path / "bgm.mp3")
+
+    @patch("video_processing.render_adapter.upload_to_oss")
+    @patch("video_processing.render_adapter.UnifiedRenderService")
+    @patch("video_processing.render_adapter.download_asset")
+    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):
+            local_path.parent.mkdir(parents=True, exist_ok=True)
+            local_path.write_bytes(b"fake video")
+            return True
+
+        mock_download.side_effect = _fake_download
+
+        mock_render = MagicMock()
+        mock_render.render.return_value = MagicMock(
+            output_path=tmp_path / "out.mp4",
+            duration=5.0,
+            file_size=1024,
+            width=1280,
+            height=720,
+        )
+        mock_render_cls.return_value = mock_render
+        mock_upload.return_value = "https://oss.example.com/out.mp4"
+
+        mock_asr = MagicMock()
+        plan = FakePlan(id="plan_asr")
+        clips = [_make_clip("c1", order=0, duration=5.0)]
+
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+        with patch.object(adapter, "_prepare_bgm", return_value=None):
+            with patch.object(adapter, "_get_asr_service", return_value=mock_asr):
+                result = adapter.render_plan(
+                    "plan_asr",
+                    work_dir=tmp_path / "work",
+                )
+
+        assert result.success
+        call_kwargs = mock_render_cls.call_args
+        assert call_kwargs.kwargs["asr_service"] is mock_asr
+
+    @patch("video_processing.render_adapter.upload_to_oss")
+    @patch("video_processing.render_adapter.UnifiedRenderService")
+    @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 区分
+        download_map = {
+            "good_001.mp4": True,
+            "bad_002.mp4": False,
+            "good_003.mp4": True,
+        }
+
+        def _fake_download(asset_id, local_path):
+            local_path.parent.mkdir(parents=True, exist_ok=True)
+            ok = download_map.get(asset_id, False)
+            if ok:
+                local_path.write_bytes(b"fake video")
+            return ok
+
+        mock_download.side_effect = _fake_download
+
+        mock_render = MagicMock()
+        mock_render.render.return_value = MagicMock(
+            output_path=tmp_path / "out.mp4",
+            duration=3.0,
+            file_size=512,
+            width=1280,
+            height=720,
+        )
+        mock_render_cls.return_value = mock_render
+        mock_upload.return_value = "https://oss.example.com/out.mp4"
+
+        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"),
+        ]
+
+        adapter, _, _ = _make_adapter(plan=plan, clips=clips)
+
+        result = adapter.render_plan(
+            "plan_mixed",
+            work_dir=tmp_path / "work",
+        )
+
+        assert result.success
+        assert "c_good1" in result.rendered_clip_ids
+        assert "c_good2" in result.rendered_clip_ids
+        assert "c_bad" in result.failed_clip_ids
+        assert len(result.rendered_clip_ids) == 2
+        assert len(result.failed_clip_ids) == 1