xiaoxia-saas/apps/worker/video_processing/render_adapter.py

"""统一渲染引擎适配层 — Phase 2.

将 EditPlan + EditPlanClips（来自 DB）适配为 UnifiedRenderService 的输入格式，
封装素材下载、BGM 准备、ASR 字幕、渲染执行、结果上传的完整流程。

职责：
  1. 从 DB 读取 EditPlan + EditPlanClips
  2. 下载素材到本地，构建 asset_path_map
  3. 准备 BGM 音频（URL / 素材库 / 预设库）
  4. 初始化 ASR 服务（自动字幕）
  5. 调用 UnifiedRenderService 执行渲染
  6. 上传渲染结果到 OSS
  7. 支持进度回调（对接 JobService）
"""

from __future__ import annotations

import logging
import subprocess
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable

from sqlalchemy.orm import Session
from video_processing.oss_helpers import download_asset, upload_to_oss
from video_processing.unified_render_service import UnifiedRenderService
from video_processing.video_validation import validate_video_output

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

logger = logging.getLogger(__name__)

DEFAULT_OUTPUT_WIDTH = 1080
DEFAULT_OUTPUT_HEIGHT = 1920


def _parse_resolution(resolution_str: str | None) -> tuple[int, int]:
    """解析分辨率字符串，如 '1080x1920' → (1080, 1920)。解析失败返回默认值。"""
    if not resolution_str or "x" not in resolution_str:
        return DEFAULT_OUTPUT_WIDTH, DEFAULT_OUTPUT_HEIGHT
    try:
        w, h = resolution_str.lower().split("x", 1)
        width = int(w.strip())
        height = int(h.strip())
        # 最小 100px 防护：避免前端传入宽高比（如 "9:16"）被 parseInt 截断为极小值
        if width < 100 or height < 100:
            logger.warning(
                "分辨率异常小 (%dx%d)，使用默认值。原始值: %s",
                width,
                height,
                resolution_str,
            )
            return DEFAULT_OUTPUT_WIDTH, DEFAULT_OUTPUT_HEIGHT
        return width, height
    except (ValueError, TypeError):
        return DEFAULT_OUTPUT_WIDTH, DEFAULT_OUTPUT_HEIGHT


# ── 数据结构 ──────────────────────────────────────────────────────────────────


@dataclass
class RenderAdapterResult:
    """渲染适配结果。"""

    success: bool
    output_url: str = ""
    output_path: Path | None = None
    thumbnail_url: str = ""
    duration: float = 0.0
    file_size: int = 0
    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 = ""
    error_detail: str = ""  # 详细错误信息（如 ffmpeg stderr），用于排查
    cover_candidates: list[dict] | None = (
        None  # 封面候选帧 [{"image_url": "...", "frame_time": 5.0, "storage_key": "..."}]
    )
    temp_dir: str | None = None  # 渲染临时目录，成功时由调用方清理，失败时由 finally 清理

    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"""


# ── 适配层主体 ────────────────────────────────────────────────────────────────


class RenderAdapter:
    """统一渲染引擎适配层。

    桥接 EditPlan 领域模型与 UnifiedRenderService 图层模型。

    用法::

        adapter = RenderAdapter(db)
        result = adapter.render_plan(
            plan_id=plan_id,
            job_id=job_id,
            progress_cb=lambda p, s: job_service.update_progress(job_id, p, s),
        )
    """

    def __init__(self, db: Session) -> None:
        self._db = db
        self._plan_repo = SQLAlchemyEditPlanRepository(db)
        self._clip_repo = SQLAlchemyEditPlanClipRepository(db)

    # ── 公开方法 ──────────────────────────────────────────────────────────

    def render_plan(
        self,
        plan_id: str,
        *,
        job_id: str = "",
        work_dir: Path | None = None,
        progress_cb: ProgressCallback | None = None,
        voiceover_audio_path: str | None = None,
    ) -> RenderAdapterResult:
        """渲染一个 EditPlan。

        完整流程：
        1. 加载计划与片段
        2. 下载素材
        3. 执行统一渲染
        4. 上传结果

        Args:
            plan_id: EditPlan ID
            job_id: 关联的 Job ID（用于结果存储路径）
            work_dir: 工作目录，不传则使用临时目录
            progress_cb: 进度回调函数
            voiceover_audio_path: 配音音频本地路径（一键生成场景使用）

        Returns:
            RenderAdapterResult
        """
        temp_dir = None
        try:
            # 0. 准备工作目录
            if work_dir is None:
                temp_dir = tempfile.mkdtemp(prefix="render_")
                work_dir = Path(temp_dir)
            work_dir.mkdir(parents=True, exist_ok=True)

            self._report_progress(progress_cb, 5.0, "加载剪辑计划")

            # 1. 加载计划与片段
            plan = self._plan_repo.get(plan_id)
            if plan is None:
                return RenderAdapterResult(
                    success=False,
                    error_message=f"剪辑计划不存在: {plan_id}",
                )

            clips = self._clip_repo.list_by_plan(plan_id, skip=0, limit=10000)
            ready_clips = [c for c in clips if c.status == EditPlanClipStatus.READY and c.asset_id]
            ready_clips.sort(key=lambda c: c.order)

            if not ready_clips:
                return RenderAdapterResult(
                    success=False,
                    error_message="没有可渲染的就绪片段",
                    clip_count=0,
                )

            logger.info(
                "开始渲染: plan_id=%s job_id=%s ready_clips=%d engine=unified",
                plan_id,
                job_id,
                len(ready_clips),
            )

            self._report_progress(progress_cb, 15.0, f"下载素材（{len(ready_clips)} 个）")

            # 2. 下载素材
            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~6. 统一渲染核心流程（BGM + ASR + 渲染 + 缩略图 + 上传）
            result = self._do_render(
                plan=plan,
                clips=ready_clips,
                asset_path_map=asset_path_map,
                work_dir=work_dir,
                plan_id=plan_id,
                job_id=job_id,
                progress_cb=progress_cb,
                rendered_clip_ids=rendered_clip_ids,
                failed_clip_ids=failed_clip_ids,
                voiceover_audio_path=voiceover_audio_path,
            )
            # 成功时将临时目录所有权转移给调用方，阻止 finally 清理
            if result.success and temp_dir:
                result.temp_dir = temp_dir
                temp_dir = None  # 阻止 finally 块清理
            return result

        except subprocess.CalledProcessError as exc:
            stderr_text = (exc.stderr or "").strip()
            logger.error(
                "[render-adapter] ffmpeg渲染失败: plan_id=%s job_id=%s exit_code=%d\nstderr:\n%s",
                plan_id,
                job_id,
                exc.returncode,
                stderr_text[-2000:] if len(stderr_text) > 2000 else stderr_text,
            )
            return RenderAdapterResult(
                success=False,
                error_message=f"FFmpeg渲染失败(exit={exc.returncode}): {stderr_text[:200]}",
                error_detail=stderr_text[-2000:] if len(stderr_text) > 2000 else stderr_text,
            )
        except Exception as exc:
            logger.exception(
                "[render-adapter] render failed: plan_id=%s job_id=%s engine=unified error=%s",
                plan_id,
                job_id,
                str(exc)[:200],
            )
            return RenderAdapterResult(
                success=False,
                error_message=str(exc)[:500],
            )
        finally:
            # 清理临时目录
            if temp_dir:
                import shutil

                try:
                    shutil.rmtree(temp_dir, ignore_errors=True)
                except Exception as cleanup_err:
                    logger.warning("临时目录清理失败: path=%s error=%s", temp_dir, cleanup_err)

    def validate_plan(self, plan_id: str) -> tuple[bool, list[str], list[str], int, int]:
        """校验计划是否可渲染（兼容 VideoComposeService.validate_compose 接口）。

        Returns:
            (valid, errors, warnings, ready_clip_count, total_clip_count)
        """
        errors: list[str] = []
        warnings: list[str] = []

        plan = self._plan_repo.get(plan_id)
        if plan is None:
            return False, [f"剪辑计划不存在: {plan_id}"], [], 0, 0

        if plan.status not in (EditPlanStatus.EDITING, EditPlanStatus.RENDERING):
            errors.append(f"计划状态不正确，需要 editing 或 rendering，当前: {plan.status}")

        clips = self._clip_repo.list_by_plan(plan_id, skip=0, limit=10000)
        if not clips:
            errors.append("计划没有任何片段")
            return False, errors, warnings, 0, 0

        clips.sort(key=lambda c: c.order)

        ready_count = 0
        pending_count = 0
        no_asset_count = 0

        for clip in clips:
            if clip.status == EditPlanClipStatus.READY:
                ready_count += 1
                if not clip.asset_id:
                    errors.append(f"片段 {clip.id} (order={clip.order}) 没有分配素材")
                    no_asset_count += 1
            elif clip.status == EditPlanClipStatus.PENDING:
                pending_count += 1
            elif clip.status == EditPlanClipStatus.FAILED:
                warnings.append(f"片段 {clip.id} (order={clip.order}) 状态为 failed，已跳过")

        if ready_count == 0:
            errors.append("没有就绪（ready）的片段可以合成")

        if pending_count > 0:
            warnings.append(f"有 {pending_count} 个片段仍处于 pending 状态")

        return len(errors) == 0, errors, warnings, ready_count, len(clips)

    # ── 内部方法 ──────────────────────────────────────────────────────────

    @staticmethod
    def _report_progress(progress_cb: ProgressCallback | None, progress: float, stage: str) -> None:
        """上报进度。"""
        if progress_cb is not None:
            try:
                progress_cb(progress, stage)
            except Exception:
                logger.exception("进度回调失败")

    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 映射（下载成功的）
            - 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()

        # 批量查询素材的 storage_key（OSS 存储路径）
        # 兼容存量数据：storage_key 为空时 fallback 到 file_url
        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.storage_key or a.file_url or "") for a in assets if a.storage_key or a.file_url
            }

        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)

            # 从素材表获取 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(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])
            else:
                failed_clip_ids.append(clip.id)
                logger.warning("素材下载失败: clip_id=%s asset_id=%s", clip.id, asset_id[:60])

        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.storage_key or model.file_url):
                    # 兼容存量数据：storage_key 为空时 fallback 到 file_url
                    storage_key = model.storage_key or 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 apps.worker.services.asr_service_factory import get_asr_service

            return get_asr_service()
        except Exception as e:
            logger.warning("ASR 服务初始化失败，自动字幕将不可用: %s", e)
            return None

    def _do_render(
        self,
        plan: Any,
        clips: list[Any],
        asset_path_map: dict[str, Path],
        work_dir: Path,
        *,
        plan_id: str,
        job_id: str = "",
        progress_cb: ProgressCallback | None = None,
        rendered_clip_ids: list[str] | None = None,
        failed_clip_ids: list[str] | None = None,
        voiceover_audio_path: str | None = None,
    ) -> RenderAdapterResult:
        """执行统一渲染核心流程（BGM + ASR + 渲染 + 缩略图 + 上传）。

        render_plan 和 render_from_memory 共用此方法。

        Args:
            rendered_clip_ids: 成功下载/准备的 clip id 列表（render_plan 从下载阶段传入）
            failed_clip_ids: 失败的 clip id 列表
            voiceover_audio_path: 配音素材库音频本地路径（一键生成场景使用）

        Returns:
            RenderAdapterResult
        """
        # 1. 准备 BGM
        bgm_path = self._prepare_bgm(plan, work_dir, plan_id)

        self._report_progress(progress_cb, 40.0, "执行视频渲染")

        # 2. 初始化 ASR
        plan_config = plan.config or {}
        asr_service = self._get_asr_service()

        # 3. 读取输出分辨率
        export_config = plan_config.get("export", {}) or {}
        if not isinstance(export_config, dict):
            export_config = {}
        output_width, output_height = _parse_resolution(export_config.get("resolution"))
        logger.info(
            "渲染输出分辨率: plan_id=%s resolution=%dx%d source=%s",
            plan_id,
            output_width,
            output_height,
            "config" if export_config.get("resolution") else "default",
        )

        # 4. 执行统一渲染
        render_svc = UnifiedRenderService(
            plan=plan,
            clips=clips,
            asset_path_map=asset_path_map,
            work_dir=work_dir,
            output_width=output_width,
            output_height=output_height,
            bgm_path=bgm_path,
            asr_service=asr_service,
            voiceover_audio_path=voiceover_audio_path,
        )
        result = render_svc.render()

        # 4.5 渲染后校验输出完整性
        validation = validate_video_output(result.output_path)
        if not validation.valid:
            logger.error(
                "[render-adapter] 渲染输出校验失败: plan_id=%s job_id=%s error=%s",
                plan_id,
                job_id,
                validation.error_message,
            )
            return RenderAdapterResult(
                success=False,
                error_message=f"渲染输出校验失败: {validation.error_message}",
                error_detail=validation.error_message,
            )
        self._report_progress(progress_cb, 80.0, "上传渲染结果")

        # 5. 上传结果
        storage_key = f"rendered/{plan_id}/{job_id or plan_id}.mp4"
        output_url = upload_to_oss(result.output_path, storage_key)

        self._report_progress(progress_cb, 90.0, "生成封面缩略图")

        # 6. 生成封面缩略图
        thumbnail_url = ""
        try:
            from video_processing.thumbnail_generator import generate_and_upload_thumbnail

            thumb_storage_key = f"rendered/{plan_id}/thumbnail.jpg"
            thumbnail_url = generate_and_upload_thumbnail(str(result.output_path), thumb_storage_key)
        except Exception as thumb_err:
            logger.warning(
                "[render-adapter] 缩略图生成失败（不影响主流程）: plan_id=%s error=%s",
                plan_id,
                thumb_err,
            )

        # 7. 抽取封面候选帧并上传 OSS（失败不阻断主流程）
        cover_candidates = None
        try:
            from video_processing.thumbnail_generator import extract_and_upload_cover_frames

            # 已渲染视频在统一渲染阶段已通过 ASS 字幕把标题烧录进画面，
            # 抽帧天然带标题，因此这里传空字符串，避免 Pillow 二次叠加导致重影。
            # Pillow 叠加仅用于 API 从源素材抽帧（源素材本身无标题）的兜底场景。
            cover_candidates = extract_and_upload_cover_frames(
                str(result.output_path), plan_id, num_frames=3, title_text=""
            )
            if cover_candidates:
                logger.info(
                    "[render-adapter] 封面候选帧生成成功: plan_id=%s count=%d",
                    plan_id,
                    len(cover_candidates),
                )
        except Exception as cover_err:
            logger.warning(
                "[render-adapter] 封面候选帧生成失败（不影响主流程）: plan_id=%s error=%s",
                plan_id,
                cover_err,
            )

        self._report_progress(progress_cb, 100.0, "渲染完成")

        logger.info(
            "[render-adapter] render success: plan_id=%s job_id=%s engine=unified "
            "duration=%.2fs file_size=%d resolution=%dx%d clip_count=%d",
            plan_id,
            job_id,
            result.duration,
            result.file_size,
            result.width,
            result.height,
            len(clips),
        )

        final_rendered_ids = (
            rendered_clip_ids if rendered_clip_ids is not None else [c.id for c in clips if hasattr(c, "id")]
        )
        final_failed_ids = failed_clip_ids if failed_clip_ids is not None else []

        return RenderAdapterResult(
            success=True,
            output_url=output_url or "",
            output_path=result.output_path,
            thumbnail_url=thumbnail_url,
            duration=result.duration,
            file_size=result.file_size,
            width=result.width,
            height=result.height,
            clip_count=len(clips),
            rendered_clip_ids=final_rendered_ids,
            failed_clip_ids=final_failed_ids,
            cover_candidates=cover_candidates,
        )

    def render_from_memory(
        self,
        plan: Any,
        clips: list[Any],
        asset_path_map: dict[str, Path],
        *,
        plan_id: str = "",
        job_id: str = "",
        work_dir: Path | None = None,
        progress_cb: ProgressCallback | None = None,
        voiceover_audio_path: str | None = None,
    ) -> RenderAdapterResult:
        """使用内存中的 plan/clips/asset_path_map 直接渲染。

        适用于一键生成等不写DB剪辑计划的场景，复用统一的 BGM/ASR/分辨率/渲染/缩略图逻辑。

        Args:
            plan: 类 EditPlan 的对象（鸭子类型，需有 id/config 等属性）
            clips: 类 EditPlanClip 的对象列表
            asset_path_map: asset_id → local_path 映射
            plan_id: 用于日志的计划标识（不传则用 plan.id）
            job_id: 关联的 Job ID
            work_dir: 工作目录，不传则用临时目录
            progress_cb: 进度回调
            voiceover_audio_path: 配音素材库音频本地路径

        Returns:
            RenderAdapterResult
        """
        actual_plan_id = plan_id or getattr(plan, "id", "memory_plan")
        temp_dir = None
        try:
            if work_dir is None:
                temp_dir = tempfile.mkdtemp(prefix="render_mem_")
                work_dir = Path(temp_dir)
            work_dir.mkdir(parents=True, exist_ok=True)

            if not clips:
                return RenderAdapterResult(
                    success=False,
                    error_message="没有可渲染的片段",
                    clip_count=0,
                )

            if not asset_path_map:
                return RenderAdapterResult(
                    success=False,
                    error_message="素材路径映射为空",
                    clip_count=len(clips),
                )

            logger.info(
                "开始内存模式渲染: plan_id=%s job_id=%s clip_count=%d engine=unified",
                actual_plan_id,
                job_id,
                len(clips),
            )

            self._report_progress(progress_cb, 35.0, "准备 BGM 音频")

            return self._do_render(
                plan=plan,
                clips=clips,
                asset_path_map=asset_path_map,
                work_dir=work_dir,
                plan_id=actual_plan_id,
                job_id=job_id,
                progress_cb=progress_cb,
                voiceover_audio_path=voiceover_audio_path,
            )

        except subprocess.CalledProcessError as exc:
            stderr_text = (exc.stderr or "").strip()
            logger.error(
                "[render-adapter] 内存模式渲染失败: plan_id=%s exit_code=%d\nstderr:\n%s",
                actual_plan_id,
                exc.returncode,
                stderr_text[-2000:] if len(stderr_text) > 2000 else stderr_text,
            )
            return RenderAdapterResult(
                success=False,
                error_message=f"FFmpeg渲染失败(exit={exc.returncode}): {stderr_text[:200]}",
                error_detail=stderr_text[-2000:] if len(stderr_text) > 2000 else stderr_text,
            )
        except Exception as exc:
            logger.exception(
                "[render-adapter] 内存模式渲染失败: plan_id=%s error=%s",
                actual_plan_id,
                str(exc)[:200],
            )
            return RenderAdapterResult(
                success=False,
                error_message=str(exc)[:500],
            )
        finally:
            if temp_dir:
                import shutil

                try:
                    shutil.rmtree(temp_dir, ignore_errors=True)
                except Exception as cleanup_err:
                    logger.warning("临时目录清理失败: path=%s error=%s", temp_dir, cleanup_err)