"""统一渲染引擎适配层 — 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 collections.abc import Callable from dataclasses import dataclass from pathlib import Path from typing import Any 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() stderr_tail = stderr_text[-5000:] if len(stderr_text) > 5000 else stderr_text logger.error( "[render-adapter] ffmpeg渲染失败: plan_id=%s job_id=%s exit_code=%d\nstderr:\n%s", plan_id, job_id, exc.returncode, stderr_tail, ) return RenderAdapterResult( success=False, error_message=f"FFmpeg渲染失败(exit={exc.returncode}): {stderr_text[-500:]}", error_detail=stderr_tail, ) 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 _resolve_clip_has_text(self, clips: list[Any]) -> list[bool] | None: """#1970：按源视频片段顺序解析 atom_clip.ai_tags.has_text。 顺序与 UnifiedRenderService 的「非 audio 源片段」口径一致。 仅当 atom_clip 存在 ai_tags 字典且 has_text 显式为 False 时标记为 无文字（允许 hflip）；atom_clip_id 缺失、ai_tags 未生成、has_text 为 true/null/非布尔值时一律按有文字处理（保守不翻转）。 查询失败时返回 None，渲染层回退到全保守路径。 """ video_clips = [c for c in clips if getattr(c, "clip_type", "main") != "audio"] atom_ids: list[str] = [] seen: set[str] = set() for c in video_clips: atom_id = getattr(c, "atom_clip_id", "") or "" if atom_id and atom_id not in seen: seen.add(atom_id) atom_ids.append(atom_id) if not atom_ids: return None try: from packages.adapters.sqlalchemy_impl.asset_atom_clip_repository import ( SQLAlchemyAssetAtomClipRepository, ) atom_clips = SQLAlchemyAssetAtomClipRepository(self._db).find_by_ids(atom_ids) except Exception as exc: logger.warning("[render-adapter] atom_clip ai_tags 查询失败，hflip 全量保守处理: %s", exc) return None has_text_map: dict[str, bool] = {} for ac in atom_clips: ai_tags = getattr(ac, "ai_tags", None) no_text = isinstance(ai_tags, dict) and ai_tags.get("has_text") is False has_text_map[ac.id] = not no_text return [has_text_map.get((getattr(c, "atom_clip_id", "") or ""), True) for c in video_clips] 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. 执行统一渲染 clip_has_text = self._resolve_clip_has_text(clips) 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, clip_has_text=clip_has_text, ) 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. 生成封面缩略图（结果通过 RenderAdapterResult.thumbnail_url 返回给调用方） thumbnail_url = "" try: from video_processing.thumbnail_generator import generate_and_upload_thumbnail # job_id 是 _do_render 方法的参数（参见方法签名） thumb_storage_key = f"rendered/{plan_id}/thumbnails/{job_id}.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, task_id=job_id, num_frames=5, 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() stderr_tail = stderr_text[-5000:] if len(stderr_text) > 5000 else stderr_text logger.error( "[render-adapter] 内存模式渲染失败: plan_id=%s exit_code=%d\nstderr:\n%s", actual_plan_id, exc.returncode, stderr_tail, ) return RenderAdapterResult( success=False, error_message=f"FFmpeg渲染失败(exit={exc.returncode}): {stderr_text[-500:]}", error_detail=stderr_tail, ) 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)