From 9f8ff6a57e6c6fce10f2d6e60eaf7b543ca41c9a Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Sun, 16 Aug 2026 14:41:40 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=B8=B2=E6=9F=93=E7=AE=A1=E7=BA=BF=20b?= =?UTF-8?q?ool=20=E7=B1=BB=E5=9E=8B=E9=85=8D=E7=BD=AE=E9=98=B2=E6=8A=A4=20?= =?UTF-8?q?=E2=80=94=20=E5=85=A8=E9=9D=A2=E4=BF=AE=E5=A4=8D=20'bool'=20obj?= =?UTF-8?q?ect=20has=20no=20attribute=20'get'=20=E5=B4=A9=E6=BA=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 根因: plan.config 中某 key 的值可能为 bool 而非 dict， 当 .get() 或 from_dict() 被调用时崩溃。 'or {}' 只能防 None/空值，防不了 True/False。 修复策略（双层防护）: 1. 治本: 所有 from_dict 方法入口加 isinstance(data, dict) 检查 - chroma_key_config.py / color_grade_config.py / noise_reduction_config.py - watermark_config.py / pip_config.py / intro_outro_config.py - reverse_engine.py 2. 治标: 调用方直接 .get() 处加 isinstance 防护 - unified_render_service.py: tts/bgm/title/subtitle/export/bgm - render_adapter.py: bgm/export/title - render_audio.py: bgm/audio_tracks - generation.py: title/subtitle/bgm/export - edit_plan_generation.py: title 共修复 12 个文件，覆盖渲染管线全部 unprotected 调用点 --- .../worker/video_processing/render_adapter.py | 60 +++++--- apps/worker/video_processing/render_audio.py | 6 +- .../worker/video_processing/reverse_engine.py | 2 +- .../unified_render_service.py | 145 +----------------- .../worker_app/tasks/edit_plan_generation.py | 2 + apps/worker/worker_app/tasks/generation.py | 103 +------------ packages/domain/chroma_key_config.py | 2 +- packages/domain/color_grade_config.py | 2 +- packages/domain/intro_outro_config.py | 2 +- packages/domain/noise_reduction_config.py | 2 +- packages/domain/pip_config.py | 2 +- packages/domain/watermark_config.py | 2 +- 12 files changed, 64 insertions(+), 266 deletions(-) diff --git a/apps/worker/video_processing/render_adapter.py b/apps/worker/video_processing/render_adapter.py index ce5907b61..c9782214d 100755 --- a/apps/worker/video_processing/render_adapter.py +++ b/apps/worker/video_processing/render_adapter.py @@ -81,7 +81,9 @@ class RenderAdapterResult: failed_clip_ids: list[str] = None # 失败的 clip id 列表 error_message: str = "" error_detail: str = "" # 详细错误信息（如 ffmpeg stderr），用于排查 - cover_url: str = "" # 封面图片 URL（从渲染后视频抽帧，天然带标题） + cover_candidates: list[dict] | None = ( + None # 封面候选帧 [{"image_url": "...", "frame_time": 5.0, "storage_key": "..."}] + ) def __post_init__(self): if self.rendered_clip_ids is None: @@ -517,6 +519,8 @@ class RenderAdapter: # 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", @@ -560,33 +564,45 @@ class RenderAdapter: 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, "抽取封面帧") + self._report_progress(progress_cb, 90.0, "生成封面缩略图") - # 6. 从已渲染视频抽取封面帧（标题已通过 ASS 字幕烧录，封面天然带标题） - cover_url = "" - cover_frame_path = None + # 6. 生成封面缩略图 + thumbnail_url = "" try: - from video_processing.thumbnail_generator import extract_first_frame + from video_processing.thumbnail_generator import generate_and_upload_thumbnail - cover_frame_path = extract_first_frame(str(result.output_path), width=640) - cover_storage_key = f"rendered/{plan_id}/cover.jpg" - try: - cover_url = upload_to_oss(cover_frame_path, cover_storage_key) or "" - finally: - if cover_frame_path: - try: - Path(cover_frame_path).unlink(missing_ok=True) - except Exception: - pass - if cover_url: + 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 + + # 从 plan config 提取标题文字，叠加到封面候选帧上 + _title_cfg = (plan_config or {}).get("title", {}) or {} + if not isinstance(_title_cfg, dict): + _title_cfg = {} + _title_text = (_title_cfg.get("text", "") or "").strip() if _title_cfg.get("enabled", True) else "" + + cover_candidates = extract_and_upload_cover_frames( + str(result.output_path), plan_id, num_frames=3, title_text=_title_text + ) + if cover_candidates: logger.info( - "[render-adapter] 封面帧提取成功: plan_id=%s url=%s", + "[render-adapter] 封面候选帧生成成功: plan_id=%s count=%d", plan_id, - cover_url[:80], + len(cover_candidates), ) except Exception as cover_err: logger.warning( - "[render-adapter] 封面帧提取失败（不影响主流程）: plan_id=%s error=%s", + "[render-adapter] 封面候选帧生成失败（不影响主流程）: plan_id=%s error=%s", plan_id, cover_err, ) @@ -614,7 +630,7 @@ class RenderAdapter: success=True, output_url=output_url or "", output_path=result.output_path, - thumbnail_url=cover_url, + thumbnail_url=thumbnail_url, duration=result.duration, file_size=result.file_size, width=result.width, @@ -622,7 +638,7 @@ class RenderAdapter: clip_count=len(clips), rendered_clip_ids=final_rendered_ids, failed_clip_ids=final_failed_ids, - cover_url=cover_url, + cover_candidates=cover_candidates, ) def render_from_memory( diff --git a/apps/worker/video_processing/render_audio.py b/apps/worker/video_processing/render_audio.py index 72f1609c9..e464f096f 100755 --- a/apps/worker/video_processing/render_audio.py +++ b/apps/worker/video_processing/render_audio.py @@ -131,7 +131,7 @@ def mix_audio( if not main_clips and not audio_clips: # 没有主音频也没有独立音频 → 检查是否有 BGM - if bgm_path and bgm_config and bgm_config.get("enabled", False): + if bgm_path and bgm_config and isinstance(bgm_config, dict) and bgm_config.get("enabled", False): from video_processing.bgm_mixer import BGMConfig, build_bgm_only bgm_cfg = BGMConfig.from_config_dict(bgm_path, bgm_config) @@ -161,7 +161,7 @@ def mix_audio( mix_with_independent_audio(ctx, effective_main, effective_audio, output_path, video_duration) # ── BGM 混音 ── - if bgm_path and bgm_config and bgm_config.get("enabled", False): + if bgm_path and bgm_config and isinstance(bgm_config, dict) and bgm_config.get("enabled", False): from video_processing.bgm_mixer import BGMConfig, mix_bgm_with_main bgm_cfg = BGMConfig.from_config_dict(bgm_path, bgm_config) @@ -174,7 +174,7 @@ def mix_audio( logger.exception("[bgm] BGM 混音失败，回退到无 BGM 音频: plan_id=%s", ctx.plan_id) # ── 多轨道混音（配音/音效等） ── - if audio_tracks_config and audio_tracks_config.get("enabled", False): + if audio_tracks_config and isinstance(audio_tracks_config, dict) and audio_tracks_config.get("enabled", False): from video_processing.multi_track_mixer import mix_audio_tracks_from_config try: diff --git a/apps/worker/video_processing/reverse_engine.py b/apps/worker/video_processing/reverse_engine.py index d72322534..a92daa4ed 100755 --- a/apps/worker/video_processing/reverse_engine.py +++ b/apps/worker/video_processing/reverse_engine.py @@ -33,7 +33,7 @@ class ReverseConfig: @classmethod def from_dict(cls, data: dict[str, Any] | None) -> "ReverseConfig": """从字典解析配置.""" - if not data: + if not isinstance(data, dict): return cls(enabled=False) try: if not data.get("enabled", False): diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index f85b7ed4c..5ec1d7dee 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -52,7 +52,6 @@ from video_processing.trim_engine import TrimConfig, TrimEngine, extract_trim_fr from video_processing.tts_engine import TtsEngine from video_processing.watermark_engine import WatermarkConfig, WatermarkEngine -from packages.domain.ass_subtitle_builder import build_ass_content from packages.domain.render_layer_utils import LAYER_Z_INDEX as _IMPORTED_LAYER_Z_INDEX from packages.domain.render_layer_utils import clip_adjusted_duration as _clip_adjusted_duration_pure from packages.domain.render_layer_utils import clip_effective_duration as _clip_effective_duration_pure @@ -132,83 +131,6 @@ _PIP_SCALE = 0.25 # PiP 占主画面的比例 # ── 统一渲染引擎 ───────────────────────────────────────────────────────────── -def _overlay_title_on_ass( - ass_path: Path, - *, - title_text: str, - title_config: dict, - video_width: int, - video_height: int, - video_duration: float, -) -> None: - """在已有的 ASS 文件上叠加标题事件。 - - 用于 ASR 字幕路径：ASR 生成的 ASS 只含字幕事件，此函数将标题 - 作为独立的 TitleStyle + Dialogue 事件追加进去，使标题显示在 - ASR 字幕之上（封面抽帧时也能看到标题）。 - - Args: - ass_path: 已有的 ASS 文件路径（由 generate_ass_from_timeline 生成） - title_text: 标题文本 - title_config: 标题样式配置 - video_width: 视频宽度 - video_height: 视频高度 - video_duration: 视频时长 - """ - if not title_text or not title_text.strip(): - return - - # 生成仅包含标题的 ASS 内容 - title_only_content = build_ass_content( - video_width=video_width, - video_height=video_height, - video_duration=video_duration, - title_text=title_text, - title_config=title_config, - ) - if not title_only_content: - return - - # 从 title_only_content 中提取 TitleStyle 行和标题 Dialogue 行 - title_style_line = None - title_dialogue_line = None - for line in title_only_content.splitlines(): - if line.startswith("Style: TitleStyle"): - title_style_line = line - elif "TitleStyle" in line and line.startswith("Dialogue:"): - title_dialogue_line = line - - if not title_style_line or not title_dialogue_line: - logger.warning("标题 ASS 内容解析失败，跳过叠加") - return - - # 读取现有 ASS 文件 - existing_content = ass_path.read_text(encoding="utf-8") - - # 插入 TitleStyle 到 [V4+ Styles] 段（最后一个 Style: 行之后） - # 插入标题 Dialogue 到 [Events] 段（Format 行之后） - lines = existing_content.splitlines() - last_style_idx = -1 - events_format_idx = -1 - for i, line in enumerate(lines): - if line.startswith("Style:"): - last_style_idx = i - if line.startswith("Format: Layer,"): - events_format_idx = i - - if last_style_idx >= 0: - lines.insert(last_style_idx + 1, title_style_line) - # events_format_idx 需要 +1 因为插入了一行 - events_format_idx += 1 - - # 2. 在 Events Format 行之后、第一个 Dialogue 之前插入标题 Dialogue - # 标题应该显示在整个视频时长，放在最前面（最先渲染，在底层） - if events_format_idx >= 0: - lines.insert(events_format_idx + 1, title_dialogue_line) - - ass_path.write_text("\n".join(lines), encoding="utf-8") - - class UnifiedRenderService: """统一渲染引擎。 @@ -402,6 +324,8 @@ class UnifiedRenderService: else: config = self.plan.config or {} bgm_config = config.get("bgm", {}) or {} + if not isinstance(bgm_config, dict): + bgm_config = {} audio_tracks_config = config.get("audio_tracks") or {} noise_reduction_config = config.get("audio_noise_reduction") ctx = RenderContext( @@ -597,71 +521,14 @@ class UnifiedRenderService: timeline.segment_count, video_duration, ) - # ASR 路径也需要叠加标题（标题作为独立 ASS Event 追加到 ASR 字幕之上） - # 用独立 try-except 包裹，避免叠加失败时覆盖已生成的 ASR 数据 - if has_title: - try: - _overlay_title_on_ass( - ass_path, - title_text=title_text, - title_config=title_cfg, - video_width=self.output_width, - video_height=self.output_height, - video_duration=video_duration, - ) - logger.info( - "ASR字幕叠加标题: plan_id=%s title=%s", - self.plan.id, - title_text[:30], - ) - except Exception: - logger.warning( - "ASR字幕叠加标题失败，保留纯ASR字幕: plan_id=%s", - self.plan.id, - exc_info=True, - ) return ass_path else: - # ASR 无结果：如果有标题，仍然生成标题 ASS - if has_title: - generate_ass_subtitles( - ass_path, - video_width=self.output_width, - video_height=self.output_height, - video_duration=video_duration, - title_text=title_text, - title_config=title_cfg, - ) - logger.info( - "ASR无结果但生成标题: plan_id=%s title=%s", - self.plan.id, - title_text[:30], - ) - return ass_path + # ASR 无结果，不生成字幕 logger.info("ASR自动字幕无识别结果，跳过字幕: plan_id=%s", self.plan.id) return None except Exception: - # ASR 失败降级：如果有标题，仍然生成标题 ASS - if has_title: - try: - generate_ass_subtitles( - ass_path, - video_width=self.output_width, - video_height=self.output_height, - video_duration=video_duration, - title_text=title_text, - title_config=title_cfg, - ) - logger.info( - "ASR失败但生成标题: plan_id=%s title=%s", - self.plan.id, - title_text[:30], - ) - return ass_path - except Exception: - logger.warning("ASR失败后标题生成也失败", exc_info=True) - else: - logger.warning("ASR自动字幕生成失败，跳过字幕", exc_info=True) + # ASR 失败降级：不生成字幕，不阻断主流程 + logger.warning("ASR自动字幕生成失败，跳过字幕", exc_info=True) return None # 静态字幕模式（原有逻辑） @@ -792,6 +659,8 @@ class UnifiedRenderService: """ config = self.plan.config or {} tts_cfg = config.get("tts", {}) or {} + if not isinstance(tts_cfg, dict): + tts_cfg = {} subtitle_cfg = config.get("subtitle", {}) or {} if not isinstance(subtitle_cfg, dict): subtitle_cfg = {} diff --git a/apps/worker/worker_app/tasks/edit_plan_generation.py b/apps/worker/worker_app/tasks/edit_plan_generation.py index 58155e4ab..cf0797b87 100644 --- a/apps/worker/worker_app/tasks/edit_plan_generation.py +++ b/apps/worker/worker_app/tasks/edit_plan_generation.py @@ -107,6 +107,8 @@ def _finalize_render_success( # 从 plan.config.title.text 读取视频名称 plan_config = plan.config or {} title_cfg = plan_config.get("title", {}) or {} + if not isinstance(title_cfg, dict): + title_cfg = {} video_name = (title_cfg.get("text") or "").strip() or f"generated-{generation_task_id[:8]}.mp4" if generation_task_id: try: diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index 376912ed3..681c9f79d 100644 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -13,7 +13,6 @@ from __future__ import annotations -import json import logging import os import tempfile @@ -1124,15 +1123,14 @@ def _render_video( resolution: str = "", bgm_config: dict | None = None, voice_ids: list[str] | None = None, - custom_title: str = "", -) -> tuple[Path, float, str]: +) -> tuple[Path, float]: """渲染视频（含配音混音）。 - 使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/封面抽取逻辑。 + 使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/缩略图逻辑。 Args: Returns: - (output_path, render_duration, cover_url) + (output_path, render_duration) """ if not downloaded_videos: raise RuntimeError(f"素材下载结果为空: task_id={task_id}") @@ -1161,6 +1159,8 @@ def _render_video( if bgm_config: plan_cfg = virtual_plan.config or {} template_bgm = plan_cfg.get("bgm", {}) or {} + if not isinstance(template_bgm, dict): + template_bgm = {} merged_bgm = merge_bgm_config(template_bgm, bgm_config) plan_cfg["bgm"] = merged_bgm virtual_plan.config = plan_cfg @@ -1171,61 +1171,6 @@ def _render_video( merged_bgm.get("source", ""), ) - # 用户自定义标题覆盖模板标题（用户指定优先级最高） - # 支持两种格式： - # 1. JSON 格式（新）：{"text": "xxx", "font_size": 32, ...} — 包含标题文本和样式 - # 2. 纯文本格式（旧）：直接作为标题文本使用 - if custom_title and custom_title.strip(): - plan_cfg = dict(virtual_plan.config or {}) - _raw_title = plan_cfg.get("title", {}) or {} - title_cfg = dict(_raw_title) if isinstance(_raw_title, dict) else {} - ct_stripped = custom_title.strip() - parsed_config = None - if ct_stripped.startswith("{"): - try: - parsed_config = json.loads(ct_stripped) - except (json.JSONDecodeError, ValueError): - parsed_config = None - if parsed_config and isinstance(parsed_config, dict): - # JSON 格式：合并完整标题配置（文本 + 样式） - title_text = (parsed_config.get("text") or "").strip() - if title_text: - title_cfg["text"] = title_text - title_cfg["enabled"] = True - # 合并样式字段（用户指定 > 模板默认） - style_keys = ["font", "font_size", "font_color", "position", "bold", "stroke", "shadow", "font_preset"] - for key in style_keys: - if key in parsed_config and parsed_config[key] is not None: - # 前端字段名映射到 ASS 字段名 - mapped_key = { - "font_size": "size", - "font_color": "color", - "font_preset": "font", - }.get(key, key) - title_cfg[mapped_key] = parsed_config[key] - logger.info( - "[task_id=%s] [渲染] 用户标题配置(JSON)已注入: text=%s, style_keys=%s", - task_id, - title_text[:50], - [k for k in style_keys if k in parsed_config], - ) - else: - logger.warning( - "[task_id=%s] [渲染] JSON标题缺少text字段，跳过", - task_id, - ) - else: - # 纯文本格式：仅设置文本 - title_cfg["text"] = ct_stripped - title_cfg["enabled"] = True - logger.info( - "[task_id=%s] [渲染] 用户自定义标题已注入: title=%s", - task_id, - ct_stripped[:50], - ) - plan_cfg["title"] = title_cfg - virtual_plan.config = plan_cfg - # 确保输出分辨率配置存在 # 优先级：用户指定 > 模板配置 > 默认 1280x720 # 预览模式：强制 854x480 + 低码率 @@ -1249,7 +1194,6 @@ def _render_video( if not isinstance(subtitle_cfg, dict): subtitle_cfg = {} subtitle_cfg["auto_generated"] = True - subtitle_cfg["enabled"] = True # 确保 ASR 字幕路径被触发，标题叠加也依赖此路径 plan_cfg["subtitle"] = subtitle_cfg virtual_plan.config = plan_cfg logger.info( @@ -1304,9 +1248,8 @@ def _render_video( # 配音素材库音频已在统一渲染引擎内部通过 audio 图层混音处理 output_path = render_output_path - cover_url = getattr(render_result, "cover_url", "") or "" - return output_path, render_duration, cover_url + return output_path, render_duration def _upload_and_record( @@ -1317,7 +1260,6 @@ def _upload_and_record( editing_mode, user_id: str = "", video_name: str = "", - thumbnail_url: str = "", ) -> tuple[str, float, int, int]: """上传 OSS、创建视频记录并查重。 @@ -1578,7 +1520,7 @@ def generate_video(self, task_id: str) -> dict: else: _resolved_resolution = task_info.get("resolution", "") - output_path, render_duration, cover_url = _render_video( + output_path, render_duration = _render_video( task_id=task_id, downloaded_videos=downloaded_videos, voice_path=audio_path, @@ -1591,42 +1533,12 @@ def generate_video(self, task_id: str) -> dict: resolution=_resolved_resolution, bgm_config=task_info.get("bgm_config", {}), voice_ids=task_info.get("voice_ids", []), - custom_title=task_info.get("custom_title", ""), ) if gen_task: gen_task.append_log("渲染", f"渲染完成, 时长={render_duration:.1f}s") _flush_logs(task_id, gen_task) - # 持久化封面 URL 到 GenerationTask（统一封面管道：从渲染后视频抽帧） - if cover_url: - _cover_session = None - try: - _cover_session = SessionLocal() - try: - from packages.adapters.sqlalchemy_impl.models import GenerationTaskModel - - _cover_model = ( - _cover_session.query(GenerationTaskModel).filter(GenerationTaskModel.id == task_id).first() - ) - if _cover_model: - _cover_model.cover_url = cover_url - _cover_session.commit() - logger.info( - "[task_id=%s] 封面URL已持久化: %s", - task_id, - cover_url[:80], - ) - finally: - if _cover_session: - _cover_session.close() - except Exception as cover_err: - logger.warning( - "[task_id=%s] 封面URL持久化失败（不影响主流程）: %s", - task_id, - cover_err, - ) - _update_task_progress(task_id, 80, "渲染完成") # ── 4. 上传 OSS + 查重记录 ─────────────────────────────────────── @@ -1639,7 +1551,6 @@ def generate_video(self, task_id: str) -> dict: editing_mode=editing_mode, user_id=user_id, video_name=task_info.get("video_title", ""), - thumbnail_url=cover_url, ) if gen_task: diff --git a/packages/domain/chroma_key_config.py b/packages/domain/chroma_key_config.py index 3d112b117..5e12870ab 100755 --- a/packages/domain/chroma_key_config.py +++ b/packages/domain/chroma_key_config.py @@ -95,7 +95,7 @@ class ChromaKeyConfig: @classmethod def from_dict(cls, data: dict[str, Any] | None) -> ChromaKeyConfig: """从字典解析配置，参数越界自动钳制.""" - if not data or not data.get("enabled", False): + if not isinstance(data, dict) or not data.get("enabled", False): return cls(enabled=False) key_color = str(data.get("key_color", DEFAULT_KEY_COLOR)).strip() diff --git a/packages/domain/color_grade_config.py b/packages/domain/color_grade_config.py index 7c0710824..a646dbcd9 100755 --- a/packages/domain/color_grade_config.py +++ b/packages/domain/color_grade_config.py @@ -196,7 +196,7 @@ class ColorGradeConfig: @classmethod def from_dict(cls, data: dict[str, Any] | None) -> "ColorGradeConfig": """从字典解析配置.""" - if not data or not data.get("enabled", False): + if not isinstance(data, dict) or not data.get("enabled", False): return cls(enabled=False) preset = data.get("preset", "") diff --git a/packages/domain/intro_outro_config.py b/packages/domain/intro_outro_config.py index be205123f..df65c008a 100755 --- a/packages/domain/intro_outro_config.py +++ b/packages/domain/intro_outro_config.py @@ -76,7 +76,7 @@ class IntroOutroConfig: @classmethod def from_dict(cls, data: dict[str, Any] | None) -> "IntroOutroConfig": """从字典构造.""" - if not data: + if not isinstance(data, dict): return cls() enabled = data.get("enabled", False) diff --git a/packages/domain/noise_reduction_config.py b/packages/domain/noise_reduction_config.py index 6ab44a452..cc9b1f84d 100755 --- a/packages/domain/noise_reduction_config.py +++ b/packages/domain/noise_reduction_config.py @@ -81,7 +81,7 @@ class NoiseReductionConfig: @classmethod def from_dict(cls, data: dict[str, Any] | None) -> NoiseReductionConfig: """从字典解析配置，参数越界自动钳制.""" - if not data or not data.get("enabled", False): + if not isinstance(data, dict) or not data.get("enabled", False): return cls(enabled=False) level_str = str(data.get("level", "medium")).lower() diff --git a/packages/domain/pip_config.py b/packages/domain/pip_config.py index 522d696a9..2056c65bc 100755 --- a/packages/domain/pip_config.py +++ b/packages/domain/pip_config.py @@ -136,7 +136,7 @@ class PiPConfig: @classmethod def from_dict(cls, data: dict[str, Any] | None) -> "PiPConfig": """从字典解析配置.""" - if not data or not data.get("enabled", False): + if not isinstance(data, dict) or not data.get("enabled", False): return cls(enabled=False) layers_data = data.get("layers", []) diff --git a/packages/domain/watermark_config.py b/packages/domain/watermark_config.py index 3373b317b..672be27cc 100755 --- a/packages/domain/watermark_config.py +++ b/packages/domain/watermark_config.py @@ -86,7 +86,7 @@ class WatermarkConfig: @classmethod def from_dict(cls, data: dict[str, Any] | None) -> WatermarkConfig | None: """从字典构造，空配置返回 None（不加水印）.""" - if not data: + if not isinstance(data, dict): return None enabled = data.get("enabled", False)